diff --git a/lib/plant.py b/lib/plant.py index c72e3e5..a4ebcd5 100644 --- a/lib/plant.py +++ b/lib/plant.py @@ -10,7 +10,8 @@ from itertools import pairwise, combinations, permutations import re from shapely.strtree import STRtree - +# 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)] @@ -74,7 +75,6 @@ class NodeIDs(): 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) @@ -99,7 +99,6 @@ class NodeIDs(): def size_of(self): return len(self._cord2id.keys()) - def get_points(self, nids:list[int]) -> list[Point]: ret = list() for n in nids: @@ -115,10 +114,10 @@ class RackIDs(): self._rack2begend = dict() # Toleranzen zur Rack anbindung aneinander (Rack Snap) self._tol_snap = tol_snap - # falls man die rack zu den Sensorpunkten abfragen möchte, ist ein STR Baum nötig + # 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): #Hier wird Rack nur mit Anfang und Ende hinzugefügt -> wie macht man Zwischenpunkte? + def add_rack(self, beg:Point, end:Point, name:str): if beg in self._point2rack: self._point2rack[beg].append(name) else: @@ -180,37 +179,6 @@ class RackIDs(): ret_sorted = pin.get_sorted_by_y() return ret_sorted - def join_racks(self): - allracks = list() - rnames = dict() - for rname, lpoints in self._rack2begend.items(): - ls = LineString(lpoints) - allracks.append(ls) - rnames[ls] = rname - - for (l1, l2) in combinations(allracks,2): - 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]) - - for (l1, l2) in permutations(allracks,2): - first = Point(l2.coords[0]) - last = Point(l2.coords[1]) - if l1.distance(first) <= self._tol_snap: - snap_point = l1.interpolate(l1.project(first)) - self.add_point_to_rack(snap_point, rnames[l1]) - # Füge zusätzliches Rack als Verbindung zwischen Endpunkt und snap_point ein - connrackname = f"c-{rnames[l2]}" - self.add_rack(first, snap_point, connrackname) - if l1.distance(last) <= self._tol_snap: - snap_point = l1.interpolate(l1.project(last)) - self.add_point_to_rack(snap_point, rnames[l1]) - # Füge zusätzliches Rack als Verbindung zwischen Endpunkt und snap_point ein - connrackname = f"c-{rnames[l2]}" - self.add_rack(last, snap_point, connrackname) - def _build_rack_strtree(self): self._rack_lines = [] self._rack_map = {} @@ -253,8 +221,6 @@ class RackIDs(): connrackname = f"c-{rnames[l2]}" self.add_rack(pt, snap_point, connrackname) - - def rack_is_horizontal(self, name): [pa, pe] = self._rack2begend[name] if pa.y == pe.y: @@ -334,8 +300,7 @@ class Anlage(): self._connect_step = tol_connect_step # 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. @@ -406,7 +371,6 @@ class Anlage(): return self._sensors[sname] raise Exception("Sensor not found") - 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 @@ -458,7 +422,6 @@ class Anlage(): nearest_point = best_line.interpolate(best_line.project(sensor)) 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 (Sensoren o. Dists), Dict der Aufpunkte von Sensoren o. Dists @@ -479,91 +442,6 @@ class Anlage(): return errors - def search_connections(self, rack_segments, segment_endpoints, tol, tol_step): - ''' Aus Rack Segmenten und Endpunkten der Racks wird unter Berücksichtigung von Toleranz naheliegende Endpunkte gefunden. - Zuerst echte Schnittpunkte und im Anschluss via Kreissuche neheliegende Punkte und deren gepinnte Berührpunkte - ''' - verbindungen = [] - endpoint_pinned = [] - - # === A: Echte Schnittpunkte zwischen Linien finden === - ''' Alle Segmente mit allen überprüfen, um echte SP zu finden''' - for i, (rack_id1, idx1, line1) in enumerate(rack_segments): - #print(f"\n=== Prüfe {rack_id1}_{idx1} auf echte Schnittpunkte") - for j, (rack_id2, idx2, line2) in enumerate(rack_segments): - if i >= j: - continue # keine Duplikate / sich selbst - - if line1.intersects(line2): - inter = line1.intersection(line2) - if inter.geom_type == "Point": - #print(f"✅ Exakter Schnittpunkt {inter} zwischen {rack_id1}_{idx1} und {rack_id2}_{idx2}") - verbindungen.append((rack_id1, idx1, rack_id2, idx2, inter)) - - # === B: Näherungsweise Verbindung durch Toleranz-Kreise === - ''' Entlanglaufen der Racks und Scan nach Endpunkten im Toleranzbereich''' - for rack_id, idx, line in rack_segments: - #print(f"\n=== Prüfe {rack_id}_{idx1} auf Punkte im Toleranzbereich") - for other_rack_id, other_idx, pt in segment_endpoints: - if rack_id == other_rack_id: - continue # ignoriere eigene Endpunkte - - # Exakte Schnittpunkte ignorieren - if line.intersects(pt): - continue - - dist = line.distance(pt) - if dist < tol: - self.increase_circle(tol, tol_step, line, pt, rack_id, idx, other_rack_id, other_idx, verbindungen, endpoint_pinned) - #print(f"🔍 Punkt {pt} liegt {dist:.2f} von Linie {rack_id}_{idx} entfernt" - - # === Endpunkte aktualisieren === - # Dict erstellen, dass mit dem Key "Rack_id - index" dahinter die Koordinaten von Anfang und Endpunkt speichert - rack_segments_pinned = dict() - - for rack_id, idx, linestring in rack_segments: - key = f"{rack_id}-{idx}" - rack_segments_pinned[key] = [Point(linestring.coords[0]), Point(linestring.coords[1])] #Alle Racks in ihrer eingelesenen Form zum Dict hinzufügen - - for rack_id, idx, old_pt, new_pt, taget_rack in endpoint_pinned: #Durch verschobene Endpunkte laufen... - key = f"{rack_id}-{idx}" - coords = rack_segments_pinned.get(key) - - if coords: #...und bei Übereinstimmung von Start oder Endkoordinate die ursprüngliche (eingelesene) mit der gepinnten überschreiben - # Vergleich mit Startpunkt - if Point(coords[0]).equals(old_pt): - coords[0] = Point(new_pt.x, new_pt.y) #.x bzw .y übergibt x bzw y Koordinate von Objekt POINT - # Vergleich mit Endpunkt - elif Point(coords[1]).equals(old_pt): - coords[1] = Point(new_pt.x, new_pt.y) - - rack_segments_pinned[key] = coords # aktualisieren - - #Dict erstellen, dass alle Punkte die an einem Rack anschließen speichert - d_rack_conn_points = dict() - - for conn_to_rack, conn_to_idx, conn_from_rack, conn_from_idx, conn_point in verbindungen: - key = f"{conn_to_rack}-{conn_to_idx} + {conn_from_rack}-{conn_from_idx}" - d_rack_conn_points[key] = [conn_point] - - - d_rack_to_points = dict() #neues Dict für Rack_id - Idx: Alle Punkte auf dem Rack - - for key, coords in rack_segments_pinned.items(): # Erst Anfangs und Endpunkt aus d_racks_segments holen - # coords = [start_point end_point] - d_rack_to_points[key] = coords.copy() - - for key, point in d_rack_conn_points.items(): # Dann aus d_rack_conn_points alle verbindungspunkte holen und dazu speichern - to_rack = key.split(" + ")[0] - if to_rack in d_rack_to_points: - d_rack_to_points[to_rack].extend(point) - - for key in d_rack_to_points: - unique_points = list({(pt.x, pt.y): pt for pt in d_rack_to_points[key]}.values()) - d_rack_to_points[key] = unique_points - - return rack_segments_pinned - def get_node_positions(self): ''' Daten werden durch generate_graph() befüllt''' return self._node_positions @@ -705,7 +583,7 @@ class TestPlant(unittest.TestCase): # 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() + rack.join_racks_str() self.assertEqual(rack.get_points_from_rack("Rack_1-1"), [Point(0, 0), Point(0, 5), Point (0, 10)]) @@ -722,7 +600,7 @@ class TestPlant(unittest.TestCase): # 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() + 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)]) @@ -841,8 +719,9 @@ class TestPlant(unittest.TestCase): # Extrahiere Farb-Informationen der Kanten edge_colors = [G[u][v].get('color', 'black') for u, v in G.edges()] # Zeiche Graphen und zeige in - nx.draw(G, pos, with_labels=False, node_size=10, font_size=8, edge_color=edge_colors) - plt.show() + 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) @@ -878,84 +757,33 @@ class TestPlant(unittest.TestCase): G1 = nx.Graph() pos = an.generate_graph(G1) - nx.draw(G1, pos, with_labels=False, node_size=10, font_size=8) - plt.show() + 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()] - nx.draw(G2, pos, with_labels=False, node_size=10, font_size=8, edge_color=edge_colors) - plt.show() + 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()] - nx.draw(G3, pos, with_labels=False, node_size=10, font_size=8, edge_color=edge_colors) - plt.show() - + if draw: + nx.draw(G3, pos, with_labels=False, node_size=10, font_size=8, edge_color=edge_colors) + plt.show() - ''' Erstellt Graphen mit Racks, Sensoren und Unterverteilern und sucht kürzeste Wege von Unterverteiler zu zugehörigen Sensoren''' - - rack_segs = {'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)} - - mapping = {'Dist_1': ['Sens_1', 'Sens_2'], - 'Dist_2': ['Sens_3']} - - # Erstelle Anlage - an = Anlage(tol_snap=1) - # 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) - # Extrahiere Farb-Informationen der Kanten - edge_colors = [G3[u][v].get('color', 'black') for u, v in G3.edges()] - # Zeiche Graphen und zeige in - nx.draw(G3, 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(G3) - - 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) 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'))