Löschen von nicht benötigten Funktionen. Hinzufügen von einem Schalter draw = T/F, der das Plotting innerhalb der unittests steuert
This commit is contained in:
+22
-194
@@ -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'))
|
||||
|
||||
Reference in New Issue
Block a user