import json from shapely.geometry import LineString, Point, box from shapely.ops import nearest_points import unittest from collections import defaultdict import bisect import networkx as nx import matplotlib.pyplot as plt from itertools import pairwise, combinations, permutations 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: def __init__(self): #self._points_by_x = [] # [(x, y)] #self._points_by_y = [] # [(y, x)] self.points = [] def add_point(self, point:Point): # bisect.insort(self._points_by_x, (x, y)) # bisect.insort(self._points_by_y, (y, x)) self.points.append(point) def add_points(self, points:list[Point]): for p in points: self.add_point(p) def query_box(self, x1, x2, y1, y2): # Suche nach x-Grenzen ix1 = bisect.bisect_left(self._points_by_x, (x1, -float('inf'))) ix2 = bisect.bisect_right(self._points_by_x, (x2, float('inf'))) candidates = self._points_by_x[ix1:ix2] # Filtere nach y ret = list() for (x,y) in candidates: if y1 <= y <= y2: ret.append(Point(x,y)) return ret def get_sorted_by_x(self): # Sortiere nach x # ret = list() # for (x,y) in self._points_by_x: # ret.append(Point(x,y)) # return ret return sorted(self.points, key = lambda p: p.x) def get_sorted_by_y(self): # # Sortiere nach y # ret = list() # for (y,x) in self._points_by_y: # ret.append(Point(x,y)) # return ret 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(): def __init__(self, points=[]): self._counter = 0 self._cord2id = dict() self._id2cord = dict() self.add_points(points) def add_point(self, point:Point): 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]: ret = list() for n in nids: c = self.get_point(n) ret.append(c) return ret def show(self): return self._id2cord class RackIDs(): 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): 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): 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() #(pa, pe) = self._rack2begend[name] 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): 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): 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): [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(): tunbeg = pos[0] tunend = pos[1] trackname = f"t-Rack-{tname}" self._racks.add_rack(tunbeg, tunend, trackname) 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_lengths"] = { rname: round(length, 1) 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()