import json from shapely.geometry import LineString, Point, box from shapely.ops import nearest_points import unittest from collections import defaultdict import networkx as nx import matplotlib.pyplot as plt from itertools import pairwise import re from shapely.strtree import STRtree import math import shapely # Globale Variable, die in main aufgerufen wird und steuert ob Graphen in unittests gezeichnet werden draw = False class PointSorter: ''' Klasse, die Punkte sortiert. Die Punkte werden in der Liste self.points gespeichert. Die Punkte werden sortiert nach x- und y-Koordinate. ''' def __init__(self): self.points = [] def add_point(self, point:Point): self.points.append(point) def add_points(self, points:list[Point]): for p in points: self.add_point(p) def get_sorted_by_x(self): return sorted(self.points, key = lambda p: p.x) def get_sorted_by_y(self): return sorted(self.points, key = lambda p: p.y) def to_json(d, pretty: bool = True) -> str: return json.dumps(d, indent=2 if pretty else None, default=str) #ensure_ascii false für darstellung von "ue" class NodeIDs(): ''' Klasse, die Punkte verwaltet und NodeIDs zu Punkten zuordnet. Die NodeIDs sind ganze Zahlen, die die Position der Punkte in der Liste self.points repräsentieren. ''' def __init__(self, points=[]): self._counter = 0 self._cord2id = dict() self._id2cord = dict() self.add_points(points) def add_point(self, point:Point): ''' Fügt den Punkt unter einer neuen NodeId hinzu. ''' if self.point_exists(point): return True self._counter += 1 self._cord2id[f"{point.x} {point.y}"] = self._counter self._id2cord[self._counter] = point def point_exists(self, point:Point) -> bool: return f"{point.x} {point.y}" in self._cord2id def nid_exists(self, nid:int) -> bool: return nid in self._id2cord def add_points(self, points:list[Point]): for p in points: self.add_point(p) def get_id(self, point:Point) -> int: if f"{point.x} {point.y}" not in self._cord2id: raise Exception(f"Point not found!, {point.x},{point.y}") return self._cord2id[f"{point.x} {point.y}"] def get_point(self, nid:int) -> Point: if nid not in self._id2cord: raise Exception(f"NodeID nnot found! {nid}") return self._id2cord[nid] def get_ids(self, points:list[Point]) -> list[int]: ret = list() for p in points: nid = self.get_id(p) ret.append(nid) return ret def size_of(self): return len(self._cord2id.keys()) def get_points(self, nids:list[int]) -> list[Point]: ''' Gibt zu einer Liste von NodeIDs die zugehörigen Punkte zurück. Die Punkte werden in der gleichen Reihenfolge wie die NodeIDs zurückgegeben. ''' ret = list() for n in nids: c = self.get_point(n) ret.append(c) return ret def show(self): return self._id2cord class RackIDs(): ''' Klasse, die Racks verwaltet und Rack-Racks und Rack-Equipment miteinander verbindet. Die Racks werden als Linien in den STR-Baum eingefügt. ''' def __init__(self, tol_snap = 200.0): self._point2rack = dict() self._rack2begend = dict() # Toleranzen zur Rack anbindung aneinander (Rack Snap) self._tol_snap = tol_snap # STR-Baum, der die Racks verwaltet und zur Verbdinungssuche Rack-Rack & Rack-Equipment verwendet wird self._rack_tree = None def add_rack(self, beg:Point, end:Point, name:str): ''' Fügt einen Rack hinzu. Fügt die Anfangs- und Endpunkte des Racks zu den Racks hinzu. ''' self.add_point_to_rack(beg, name) self.add_point_to_rack(end, name) self._rack2begend[name] = (beg, end) # Anfangs und Endpunkte zu Rack Namen merken def add_racks(self, racks:dict): ''' Fügt Racks hinzu. Fügt die Anfangs- und Endpunkte der Racks zu den Racks hinzu. ''' for name,v in racks.items(): if len(v) == 2: self.add_rack(v[0], v[1], name) else: counter = 0 for start, end in pairwise(v): counter +=1 self.add_rack(start, end, f"{name}-{counter}") def get_racks_borders(self) -> dict: ''' Gibt Rack nur mit Anfangs und Endpunkt zurück. {Rack_1_0: "Point(0, 0), Point(0,15)", ... } ''' return self._rack2begend def get_racks_from_all_points(self) -> dict: ''' Gibt zu einem Punkt, diejenigen Racks zurück, auf denen der Punkt liegt. {Point(0, 0): ["Rack_1-0", "Rack_2-0", ...]} ''' return self._point2rack def get_rack_names(self) -> list: return list(self._rack2begend.keys()) def add_point_to_rack(self, point:Point, name:str): ''' Fügt einen Punkt zu genanntem Rack hinzu. (z.B Zwischenpunkt in der Mitte)''' if point not in self._point2rack: self._point2rack[point] = set() self._point2rack[point].add(name) def get_racks_from_point(self, point:Point) -> set[str]: return self._point2rack[point] def get_points_from_rack(self, name:str) -> list[Point]: ''' Gibt zu Namen von Rack zugehörige Punkte aus und sortiert Punkte''' ret = list() pin = PointSorter() for p, l_racks in self._point2rack.items(): if name in l_racks: ret.append(p) pin.add_points(ret) ret_sorted = list() if self.rack_is_horizontal(name): ret_sorted = pin.get_sorted_by_x() else: ret_sorted = pin.get_sorted_by_y() return ret_sorted def _build_rack_strtree(self): ''' Erzeugt einen STR-Baum aus den Racks. Die Racks werden als Linien in den STR-Baum eingefügt. ''' self._rack_lines = [] self._rack_map = {} for r_name, pts in self.get_racks_borders().items(): line = LineString([pts[0], pts[-1]]) self._rack_lines.append(line) self._rack_map[line] = r_name self._rack_tree = STRtree(self._rack_lines) def join_racks_str(self): ''' Verbindet Racks miteinander. Die Racks werden als Linien in den STR-Baum eingefügt. ''' if self._rack_tree is None: self._build_rack_strtree() rack_tree = self._rack_tree rnames = self._rack_map allracks = self._rack_lines # Erzeugung von BoundingBox for i, l1 in enumerate(allracks): bbox = box(*l1.bounds).buffer(self._tol_snap) candidates = rack_tree.query(bbox) candidates = [self._rack_lines[idx] for idx in candidates] for l2 in candidates: if l1.equals(l2): continue # Echte Schnittpunkte if l1.intersects(l2): inter = l1.intersection(l2) if inter.geom_type == "Point": self.add_point_to_rack(inter, rnames[l1]) self.add_point_to_rack(inter, rnames[l2]) continue # Beinahe Schnittpunkte -> Snapping for pt in [Point(l2.coords[0]), Point(l2.coords[-1])]: if l1.distance(pt) <= self._tol_snap: snap_point = l1.interpolate(l1.project(pt)) self.add_point_to_rack(snap_point, rnames[l1]) connrackname = f"c-{rnames[l2]}-{rnames[l1]}" self.add_rack(pt, snap_point, connrackname) def rack_is_horizontal(self, name): ''' Gibt True zurück, wenn der Rack horizontal ist. False, wenn der Rack vertikal ist. ''' [pa, pe] = self._rack2begend[name] if pa.y == pe.y: return True else: return False class Anlage(): r""" Baut eine Anlage besteend aus Kabeltrassen (Racks), Sensoren und Unterverteilern auf. Ermöglicht die Berechnung der günstigsten Kabelwege und gibt die Kabellängen von jedem Sensor zum zugehörigen Unterverteiler aus. Parameters ---------- Returns ------- Examples -------- >>> # Erstelle Anlage >>> an = Anlage() >>> # Füge racks aus Daten hinzu >>> an.set_racks(rack_segs) >>> # Verbinde Racks miteinander (ggf. verlängere ungenaue Racks) >>> an.join_racks() >>> # Füge Sensoren als Knoten hinzu >>> an.add_sensors(sensors) >>> # Verbinde Sensoren mit deren naheliegendsten Racks >>> an.connect_sensors_to_racks() >>> # Füge UV hinzu >>> an.add_distributors(distributors) >>> # Verbinde UV mit deren naheliegendsten Racks >>> an.connect_distributor_to_racks() >>> # Verknüpfe Sensoren mit zugehörigem UV >>> an.map_distributors_to_sensors(mapping) >>> >>> # Initialisiere Graph >>> G3 = nx.Graph() >>> # Fülle eben erstellten Graphen mit Daten >>> pos = an.generate_graph(G3) >>> # Ermittle kürzeste Wege von Unterverteilern zu zugehörigen Sensoren >>> paths = an.create_cable_paths(G3) >>> print(paths['Dist_1-Sens_1']["path_coords"]) ... [Point(-1, 9), Point(0, 9), Point(0, 3), Point(0, 1), Point(1, 1)] >>> print(paths['Dist_1-Sens_2']["path_coords"]) ... [Point(-1, 9), Point(0, 9), Point(0, 3), Point(2, 3), Point(2, 4)] >>> print(paths['Dist_2-Sens_3']["path_coords"]) ... [Point(11, 0), Point(10, 0), Point(10, 2), Point(9, 2)] >>> print(paths['Dist_1-Sens_1']["path_length"]) ... 10 """ def __init__(self, tol_snap=200.0, tol_connect=5000.0): # Container für alle Racks self._racks = RackIDs(tol_snap=tol_snap) self._rack_lengths = defaultdict(float) # zuordnung zwischen KnotenID und Punkt self._nodeids = NodeIDs() # Container für alle Sensoren self._sensors = dict() self._sensor_onpoints = dict() # Container für alle Unterverteiler self._distributors = dict() self._distributors_onpoints = dict() # Container für alle Tunnel self._tunnels = dict() self._tunnel_onpoints = dict() self._tunnel_lengths = dict() #Container für alle Wege self._sensor2dist = dict() # Container für Sensor Artikelnummern self._sensor_artnrs = dict() # Toleranzen zur Rack anbindung aneinander (Rack Snap) self._tol_snap = tol_snap # Toleranzen zur Anbindung von Sensoren / Verteilern zu Racks self._tol_connect = tol_connect # Infos zum zeichnen des Graphen self._node_positions = dict() def set_racks(self, racks:dict[str, list[Point]]): r""" Fügt racks aus eingelsener Datei zu Anlage hinzu. Parameters ---------- racks - dict{"Rack_1-1": [Point(1, 0), Point(2,0), ...]} """ return self._racks.add_racks(racks) def get_racks(self) -> dict: r""" Gibt in Anlage enthaltene Racks zurück. Returns ---------- dict{Point(0,0): ["Rack_1", "Rack_2", ...]} """ return self._racks._point2rack def add_point_to_rack(self, point:Point, rname:str): r""" Fügt einen Punkt zu einem angegebenen Rack hinzu. """ return self._racks.add_point_to_rack(point, rname) def get_all_rack_points(self): r""" Gibt einer Liste von allen Punkten allen Racks zurück. Returns ---------- [Point(0,0), Point(1,5), Point(2,6)] """ ret = list() for rname in self._racks.get_rack_names(): ret.extend(self.get_points_from_rack(rname)) return list(set(ret)) def get_rack_names(self) -> list: return self._racks.get_rack_names() def get_points_from_rack(self, rname) -> list: ''' Gibt zu Namen von Rack zugehörige Punkte aus und sortiert Punkte''' return self._racks.get_points_from_rack(rname) def get_points_from_sensors(self): '''gibt die Aufpunkte aller Sensoren auf den Racks zurück''' return self._sensors.values() def get_distributor_onpoints(self): '''gibt die Aufpunkte aller Unterverteiler auf den Racks zurück''' return self._distributors_onpoints.values() def get_sensor_onpoints(self): '''gibt die Aufpunkte aller Sensoren auf den Racks zurück''' return self._sensor_onpoints.values() def add_sensor(self, sname: str, pos:Point): self._sensors[sname] = pos def add_sensors(self, sensors:dict): for sname,pos in sensors.items(): self.add_sensor(sname, pos) def get_sensor_point(self, sname:str) -> Point: if sname in self._sensors: return self._sensors[sname] raise Exception("Sensor not found") def set_sensor_artnrs(self, artnr_dict: dict): self._sensor_artnrs = artnr_dict def connect_sensors_to_racks(self) -> list: '''verbindet die Sensoren mit den Racks. die Rückgabe enthält ein Tuple, welche Sensoren keinem Rack zugeordnet werden konnten ''' return self.connect_equipment_to_racks(self._sensors, self._sensor_onpoints) def add_distributor(self, dname: str, pos:Point): self._distributors[dname] = pos def add_distributors(self, distributors:dict): for dname,pos in distributors.items(): self.add_distributor(dname, pos) def get_distributor_point(self, dname:str) -> Point: if dname in self._distributors: return self._distributors[dname] raise Exception("Distributor not found") def connect_distributor_to_racks(self) -> list: '''verbindet die Unterverteiler mit den Racks die Rückage enthält ein Tuple, welche Unterverteiler keinem Rack zugeordnet werden konnten ''' return self.connect_equipment_to_racks(self._distributors, self._distributors_onpoints) def get_tunnel_onpoints(self): return self._tunnel_onpoints.values() def add_tunnel(self, tname: str, pos:Point): self._tunnels[tname] = pos def set_tunnel_length(self, tunlength: dict): self._tunnel_lengths = tunlength def get_tunnel_length(self, tname: str): if tname in self._tunnels: return float(self._tunnel_lengths[tname]) raise Exception("Tunnel-Length not found") def add_tunnels(self, tunnels:dict): for tname,pos in tunnels.items(): self.add_tunnel(tname, pos) def get_tunnel_point(self, tname:str) -> Point: if tname in self._tunnels: return self._tunnels[tname] raise Exception("Tunnel not found") def connect_tunnels(self) -> list: '''verbindet die Anfangs und Endpunkte der Tunnel miteiandner als t-Rack Verbindet sowohl Tunnel anfang als auch Tunnel ende jeweils mit nahegelgenem Rack (standard Aufruf wie Sensoren / Distributoren) die Rückage enthält ein Tuple, welche Tunnel keinem Rack zugeordnet werden konnten ''' for tname, pos in self._tunnels.items(): if len(pos) == 2: tunbeg = pos[0] tunend = pos[1] trackname = f"t-Rack-{tname}" self._racks.add_rack(tunbeg, tunend, trackname) else: raise Exception("Tunnel has more than one begin and end/ just one begin") errors = list() for name, pos in self._tunnels.items(): p1 = self._tunnels[name][0] p2 = self._tunnels[name][1] ret = self.connect_equipment_to_racks({f"{name}-A": p1}, self._tunnel_onpoints) if ret: errors.append(ret) ret = self.connect_equipment_to_racks({f"{name}-E": p2}, self._tunnel_onpoints) if ret: errors.append(ret) return errors def coord_is_in_tunnel(self, coord): for tname, pos in self._tunnels.items(): if coord == pos[0]: return True elif coord == pos[1]: return True return False def join_racks(self): self._racks.join_racks_str() def get_rack_lengths(self) -> dict[str, float]: return self._rack_lengths def find_nearest_rack_from_point_STR_bbox(self, max_dist, pt:Point) -> tuple[Point, str]: if self._racks._rack_tree is None: self._racks._build_rack_strtree() minx, miny, maxx, maxy = pt.x - max_dist, pt.y - max_dist, pt.x + max_dist, pt.y + max_dist bbox = box(minx, miny, maxx, maxy) candidates = self._racks._rack_tree.query(bbox) if len(candidates) == 0: raise LookupError("no candidates in box found") candidates = [self._racks._rack_lines[idx] for idx in candidates] best_dist = max_dist best_line = candidates[0] for line in candidates: dist = pt.distance(line) if dist < best_dist: best_dist = dist best_line = line rack_name = self._racks._rack_map[best_line] nearest_point = best_line.interpolate(best_line.project(pt)) return nearest_point, rack_name def connect_equipment_to_racks(self, equipment: dict, onpoints: dict) -> list: '''Verbindet Peripherie (Sensoren / Aktoren / Unterverteiler) mit dem nächsten Rack. Eingabe: Dict des Equipments ({'Name': Point}), Dict der Aufpunkte von Sensoren o. Dists (zu Beginn leer) Rückgabe: Liste der nicht mit Racks verbundenen Geräte (z.B. Entfernung zu groß) ''' errors = list() for name, pos in equipment.items(): try: onpoint, rackname = self.find_nearest_rack_from_point_STR_bbox(self._tol_connect, pos) onpoints[name] = (onpoint, rackname) except LookupError: errors.append({"name": name, "coords": {"x":pos.x, "y":pos.y}}) # Name des fehlgeschlagenenen und Position als Koodinaten zurückgeben continue self.add_point_to_rack(onpoint, rackname) virtual_rackname = f"v-{name}-{rackname}" self._racks.add_rack(pos, onpoint, virtual_rackname) return errors def get_node_positions(self): ''' Daten werden durch generate_graph() befüllt''' return self._node_positions def is_sensor(self, p:Point) -> bool: if p in self._sensors.values(): return True else: return False def is_distributor(self, p:Point) -> bool: if p in self._distributors.values(): return True else: return False def generate_graph(self, G:nx.Graph): points = list() points.extend(self.get_all_rack_points()) points.extend(self.get_points_from_sensors()) self._nodeids.add_points(points) for p in points: if self.is_distributor(p): shape = "s" elif self.is_sensor(p): shape = "^" else: shape = "o" nid = self._nodeids.get_id(p) G.add_node(nid, shape=shape) # Knoten für Startpunkt for node in G.nodes: point = self._nodeids.get_point(node) self._node_positions[node] = (point.x, point.y) for rname in self.get_rack_names(): plist = self.get_points_from_rack(rname) for start, end in pairwise(plist): nid_start = self._nodeids.get_id(start) nid_end = self._nodeids.get_id(end) dx = start.x - end.x dy = start.y - end.y if shapely.has_z(start) and shapely.has_z(end): dz = (getattr(start, "z", 0) - getattr(end, "z", 0)) else: dz = 0.0 weight = math.sqrt(dx**2 + dy**2 + dz**2) if re.match("v-.*", rname): color = "red" elif re.match("d-.*", rname): color = "blue" elif re.match("t-.*", rname): color = "orange" tname = rname.split("-")[2] weight = self.get_tunnel_length(tname) elif re.match("c-.*", rname): color = "green" else: color = "black" G.add_edge(nid_start, nid_end, color=color, weight=weight) self._rack_lengths[rname] += round(weight/1000, 1) return self._node_positions def map_distributor_to_sensors(self, dname:str, snamen:list[str]): ''' Gibt zu einem Distributor die zugehörigen Sensoren an, die später zugeordnet werden. Dist_1: ["Sens_3, Sens_5, ...] ''' for sname in snamen: self._sensor2dist[sname] = dname def map_distributors_to_sensors(self, d2sensors:dict[str, list[str]]): ''' Gibt zu einem dict mit Distributors die jeweils zugehörigen Sensoren aus. {Dist_1: ["Sens_3, Sens_5, ...] Dist_2: ["Sens_1, Sens_8, ...]} ''' for dname, listofsensors in d2sensors.items(): self.map_distributor_to_sensors(dname, listofsensors) def create_cable_path(self, G, sname, dname) -> tuple: quelle = self._nodeids.get_id(self.get_distributor_point(dname)) ziel = self._nodeids.get_id(self.get_sensor_point(sname)) pfad_nodes = nx.shortest_path(G, source=quelle, target=ziel, weight='weight') pfad_length = nx.shortest_path_length(G, source=quelle, target=ziel, weight='weight') return pfad_nodes, pfad_length def create_cable_paths(self, G) -> dict: pfade = dict() pfade["kabel"] = list() routing_errors = dict() # Fehler Liste für fehlgeschlagene Kabelverbindungen for sname, dname in self._sensor2dist.items(): try: pfad_nodes, pfad_length = self.create_cable_path(G, sname, dname) except: if dname not in routing_errors: routing_errors[dname] = list() routing_errors[dname].append(sname) continue pfad_coords = self._nodeids.get_points(pfad_nodes) ld = dict() ld['id'] = f"{dname}-{sname}" if len(self._sensor_artnrs) > 0: ld['s_artinr'] = self._sensor_artnrs.get(sname, "") ld["coords"]= [{"x":round(p.x,1), "y":round(p.y,1)} for p in pfad_coords] ld["length"]= round(pfad_length,1) #ld["nodes"]=pfad_nodes ld['tunnel_indices'] = list() for i, p in enumerate(pfad_coords): if self.coord_is_in_tunnel(p): ld['tunnel_indices'].append(i) pfade["kabel"].append(ld) pfade["rack_geometry"] = { rname: { "length": round(length, 1), "coordinates": [ { "x": round(p.x, 1), "y": round(p.y, 1), "z": round(p.z if shapely.has_z(p) else 0.0, 1) } for p in self.get_points_from_rack(rname) ] } for rname, length in self._rack_lengths.items() } pfade["errors_routing"] = dict() lOfDistErrors = list() # sammeln wo keine Verbindung zwischen Dist und Senor möglich war for dname,lofSensornamen in routing_errors.items(): lOfDistErrors.append({"unterverteiler":dname, "sensoren":lofSensornamen}) pfade["errors_routing"] = lOfDistErrors return pfade def show_node_ids(self): return self._nodeids.show() class TestPlant(unittest.TestCase): def test_duplicate_points(self): ''' Testet das Nicht-Hinzufügen von doppelten Punkten''' # Initialisiere die Liste an Knoten nodeids = NodeIDs() # Setze gleichen Knoten doppelt nodeids.add_point(Point(1,1)) nodeids.add_point(Point(1,1)) self.assertEqual(nodeids.size_of(), 1) def test_cut_rack_in_segments(self): ''' Teilt Rack aus Polyline in mehrere Segmente automatisch auf.''' racks_data = { 'Rack_1': [Point(0, 0), Point(0, 10), Point (10, 10)], 'Rack_2': [Point(-5, 5), Point(5, 5)] } # Initialisiere Racks rack = RackIDs() # Füge Racks aus gegebenen Daten hinzu und teile Rack_1 bestehend aus 3 Punkten in 2 Racks auf rack.add_racks(racks_data) self.assertEqual(rack.get_rack_names(), ['Rack_1-1', 'Rack_1-2', 'Rack_2']) def test_intersect_segments(self): ''' Stellt Schnittpunkte zwischen Racks fest und fügt Schnittpunkt zu Rack hinzu. ''' racks_data = { 'Rack_1': [Point(0, 0), Point(0, 10), Point (10, 10)], 'Rack_2': [Point(-5, 5), Point(5, 5)], } # Initialisiere Racks rack = RackIDs() # Füge Racks aus gegebenen Daten hinzu und teile Rack_1 bestehend aus 3 Punkten in 2 Racks auf rack.add_racks(racks_data) # Verknüpfe Racks mit echten Schnittpunkten und füge Schnittpunkte (exakt & beinahe) zu jeweiligem Rack hinzu rack.join_racks_str() self.assertEqual(rack.get_points_from_rack("Rack_1-1"), [Point(0, 0), Point(0, 5), Point (0, 10)]) def test_snap_segments(self): ''' Verlängert Anfangs und Endpunkte von Racks, sodass sie auf naheliegenden Racks liegen''' racks_data = { 'Rack_1': [Point(0, 0), Point(0, 10)], 'Rack_2': [Point(1, 5), Point(5, 5)], 'Rack_3': [Point(1.5, 7.5), Point(5, 7.5)] } # Initialisiere Racks rack = RackIDs(tol_snap=1) # Füge Racks aus gegebenen Daten hinzu und teile Rack_1 bestehend aus 3 Punkten in 2 Racks auf rack.add_racks(racks_data) # Verknüpfe Racks mit echten Schnittpunkten und füge Schnittpunkte (exakt & beinahe) zu jeweiligem Rack hinzu rack.join_racks_str() #Rack 2 wird verlängert auf SP mit Rack 1. Rack 3 ausserhalb der Toleranz self.assertEqual(rack.get_points_from_rack("Rack_1"), [Point(0, 0), Point(0, 5), Point (0, 10)]) def test_ids_to_point(self): ''' Testet, ob gefragter Punkt auf Racks a, b, c liegt''' res_rack_seg = {'Rack_1-0': [Point(1, 0), Point(5, 6)], 'Rack_2-0': [Point(1, 8), Point(1, 0)], 'Rack_2-1': [Point(0, 10), Point(5, 10)]} # 10 |R2-1: #################### # 9 | # 8 | | # 7 | | # 6 | | #### R1-0 # 5 | | # # 4 | | # # 3 | | # # 2 | | # # 1 | |# # 0 |-----#-------------------- # 0 1 2 3 4 5 point2rack = RackIDs() point2rack.add_racks(res_rack_seg) t1 = set(point2rack.get_racks_from_point(Point(1, 0))) self.assertEqual(t1, set(["Rack_1-0", "Rack_2-0"])) t2 = point2rack.get_racks_from_point(Point(5, 6)) self.assertEqual(t2, set(["Rack_1-0"])) self.assertEqual(point2rack.get_points_from_rack("Rack_2-0"), [Point(1, 0), Point(1, 8)]) def test_add_point_interim(self): ''' Testet das hinzufügen und einsortieren eines Zwischenpunktes zwischen Rack-Anfang und Rack-Ende''' res_rack_seg = {'Rack_1-0': [Point(1, 0), Point(5, 6)], 'Rack_2-0': [Point(1, 8), Point(1, 0)], 'Rack_2-1': [Point(0, 10), Point(5, 10)]} point2rack = RackIDs() point2rack.add_racks(res_rack_seg) point2rack.add_point_to_rack(Point(1,4), "Rack_2-0") self.assertEqual(point2rack.get_points_from_rack("Rack_2-0"), [Point(1, 0), Point(1,4), Point(1, 8)]) def test_add_sensor(self): ''' Erzeugt Aufpunkt an dem Sensor nähesten Rack und fügt diesen auf Rack ein (sortiert).''' rack_segs = {'Rack_1-0': [Point(0, 0), Point(0, 10)], 'Rack_2-0': [Point(10, -2), Point(10, 5)], 'Rack_2-1': [Point(0, 3), Point(10, 3)]} sensors = {'Sens_1': Point(1, 1), 'Sens_2': Point(2, 4), 'Sens_3': Point(9, 2)} an = Anlage() point2rack = an.set_racks(rack_segs) an.add_sensors(sensors) plist1 = an.get_points_from_rack("Rack_1-0") an.connect_sensors_to_racks() plist2 = an.get_points_from_rack("Rack_1-0") self.assertEqual(plist1, [Point(0, 0), Point(0, 10)]) self.assertEqual(plist2, [Point(0, 0), Point(0,1), Point(0, 10)]) def test_add_equipment_str_tree(self): racks = {'Rack_1': [Point(0, 0), Point(0, 10)], 'Rack_2': [Point(10, -2), Point(10, 5)], 'Rack_3': [Point(0, 3), Point(10, 3)]} sensors = {'Sens_1': Point(1, 1), 'Sens_2': Point(2, 4), 'Sens_3': Point(9, 2)} distributors = {'Dist_1': Point(-1, 9), 'Dist_2': Point(11, 0)} an = Anlage(tol_snap=1.5, tol_connect=1.5) an.set_racks(racks) an.join_racks() an.add_sensors(sensors) an.add_distributors(distributors) an.connect_equipment_to_racks(an._sensors, an._sensor_onpoints) an.connect_equipment_to_racks(an._distributors, an._distributors_onpoints) plist1 = an.get_points_from_rack("Rack_1") plist2 = an.get_points_from_rack("Rack_2") self.assertEqual(plist1, [Point(0, 0), Point(0, 1), Point(0, 3), Point(0, 9), Point(0, 10)]) self.assertEqual(plist2, [Point(10, -2), Point(10, 0), Point(10, 2), Point(10, 3), Point(10, 5)]) def test_wegsuche_str_tree(self): racks = {'Rack_1-0': [Point(0, 0), Point(0, 10)], 'Rack_2-0': [Point(10, -2), Point(10, 5)], 'Rack_2-1': [Point(0, 3), Point(10, 3)]} sensors = {'Sens_1': Point(1, 1), 'Sens_2': Point(2, 4), 'Sens_3': Point(9, 2)} distributors = {'Dist_1': Point(-1, 9), 'Dist_2': Point(11, 0)} mapping = {'Dist_1': ['Sens_1', 'Sens_2'], 'Dist_2': ['Sens_3']} an = Anlage(tol_snap=1) an.set_racks(racks) an.join_racks() an.add_sensors(sensors) an.add_distributors(distributors) an.connect_equipment_to_racks(an._sensors, an._sensor_onpoints) an.connect_equipment_to_racks(an._distributors, an._distributors_onpoints) an.map_distributors_to_sensors(mapping) G = nx.Graph() # Fülle eben erstellten Graphen mit Daten pos = an.generate_graph(G) # Extrahiere Farb-Informationen der Kanten edge_colors = [G[u][v].get('color', 'black') for u, v in G.edges()] # Zeiche Graphen und zeige in if draw: nx.draw(G, pos, with_labels=False, node_size=10, font_size=8, edge_color=edge_colors) plt.show() # Ermittle kürzeste Wege von Unterverteilern zu zugehörigen Sensoren paths = an.create_cable_paths(G) paths_by_id = {p['id']: p for p in paths["kabel"]} self.assertEqual(paths_by_id['Dist_1-Sens_1']["coords"], [{'x': -1.0, 'y': 9.0}, {'x': 0.0, 'y': 9.0}, {'x': 0.0, 'y': 3.0}, {'x': 0.0, 'y': 1.0}, {'x': 1.0, 'y': 1.0}]) self.assertEqual(paths_by_id['Dist_1-Sens_2']["coords"], [{'x': -1.0, 'y': 9.0}, {'x': 0.0, 'y': 9.0}, {'x': 0.0, 'y': 3.0}, {'x': 2.0, 'y': 3.0}, {'x': 2.0, 'y': 4.0}]) self.assertEqual(paths_by_id['Dist_2-Sens_3']["coords"], [{'x': 11.0, 'y': 0.0}, {'x': 10.0, 'y': 0.0}, {'x': 10.0, 'y': 2.0}, {'x': 9.0, 'y': 2.0}]) self.assertEqual(paths_by_id['Dist_1-Sens_1']["length"], 10) self.assertEqual(paths_by_id['Dist_1-Sens_2']["length"], 10) self.assertEqual(paths_by_id['Dist_2-Sens_3']["length"], 4) def test_generate_graph(self): '''Generiert einen Graphen in 3 unterschiedlichen Ausbaustufen (nur Racks, Racks+Sensoren, Racks+Sensoren+Unterverteiler)''' rack_segs = {'Rack_1-0': [Point(0, 0), Point(0, 10)], 'Rack_2-0': [Point(10, -2), Point(10, 5)], 'Rack_2-1': [Point(0, 3), Point(10, 3)]} sensors = {'Sens_1': Point(1, 1), 'Sens_2': Point(2, 4), 'Sens_3': Point(9, 2)} distributors = {'Dist_1': Point(-1, 9), 'Dist_2': Point(11, 0)} an = Anlage() an.set_racks(rack_segs) an.join_racks G1 = nx.Graph() pos = an.generate_graph(G1) if draw: nx.draw(G1, pos, with_labels=False, node_size=10, font_size=8) plt.show() an.add_sensors(sensors) an.connect_sensors_to_racks() G2 = nx.Graph() pos = an.generate_graph(G2) edge_colors = [G2[u][v].get('color', 'black') for u, v in G2.edges()] if draw: nx.draw(G2, pos, with_labels=False, node_size=10, font_size=8, edge_color=edge_colors) plt.show() an.add_distributors(distributors) an.connect_distributor_to_racks() G3 = nx.Graph() pos = an.generate_graph(G3) edge_colors = [G3[u][v].get('color', 'black') for u, v in G3.edges()] if draw: nx.draw(G3, pos, with_labels=False, node_size=10, font_size=8, edge_color=edge_colors) plt.show() if __name__ == '__main__': # Plot Ausgabe in Unittests steuern draw = False suite = unittest.TestSuite() suite.addTest(TestPlant('test_duplicate_points')) suite.addTest(TestPlant('test_cut_rack_in_segments')) suite.addTest(TestPlant('test_intersect_segments')) suite.addTest(TestPlant('test_snap_segments')) suite.addTest(TestPlant('test_ids_to_point')) suite.addTest(TestPlant('test_add_point_interim')) suite.addTest(TestPlant('test_add_sensor')) suite.addTest(TestPlant('test_add_equipment_str_tree')) suite.addTest(TestPlant('test_wegsuche_str_tree')) suite.addTest(TestPlant('test_generate_graph')) runner = unittest.TextTestRunner() runner.run(suite) #unittest.main()