import json from shapely.geometry import LineString, Point 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 import re class PointSorter: def __init__(self): self._points_by_x = [] # [(x, y)] self._points_by_y = [] # [(y, x)] def add_point(self, x, y): bisect.insort(self._points_by_x, (x, y)) bisect.insort(self._points_by_y, (y, x)) def add_points(self, points:list[Point]): for p in points: self.add_point(p.x, p.y) 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 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 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"Punkt nicht vorhanden!, {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 nicht vorhanden! {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, racks=dict(), tol_snap = 1): self._point2rack = dict() self._rack2begend = dict() self.add_racks(racks) # Toleranzen zur Rack anbindung aneinander (Rack Snap) self._tol_snap = tol_snap def add_rack(self, beg:Point, end:Point, name:str): #Hier wird Rack nur mit Anfang und Ende hinzugefügt -> wie macht man Zwischenpunkte? if beg in self._point2rack: self._point2rack[beg].append(name) else: self._point2rack[beg] = [name] if end in self._point2rack: self._point2rack[end].append(name) else: self._point2rack[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): if point in self._point2rack: self._point2rack[point].append(name) else: self._point2rack[point] = [name] def get_racks_from_point(self, point:Point) -> list[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 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 combinations(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]) if l1.distance(last) <= self._tol_snap: snap_point = l1.interpolate(l1.project(last)) self.add_point_to_rack(snap_point, rnames[l1]) 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, snap_step=10, tol_connect=2, tol_connect_step=0.5): # Container für alle Racks self._racks = RackIDs() # 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 Wege self._sensor2dist = dict() # Toleranzen zur Rack anbindung aneinander (Rack Snap) self._tol_snap = tol_snap self._snap_step = snap_step # Toleranzen zur Anbindung von Sensoren / Verteilern zu Racks self._tol_connect = tol_connect self._connect_step = tol_connect_step 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): return self._sensors.values() def get_sensor_onpoints(self): 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: return self._sensors[sname] def connect_sensors_to_racks(self): for sname, pos in self._sensors.items(): rack_borders = self._racks.get_racks_borders() onpoint, rack_name = self.find_nearest_rack_from_point(2, 0.5, pos, rack_borders) self._sensor_onpoints[sname] = (onpoint, rack_name) self.add_point_to_rack(onpoint, rack_name) # Füge "virtuelle Racks" von Sensor zu Aufpunkt von Sensor auf Rack hinzu. vrackname = f"v-{sname}-{rack_name}" self._racks.add_rack(pos, onpoint, vrackname) return 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: return self._distributors[dname] def connect_distributor_to_racks(self): for dname, pos in self._distributors.items(): rack_borders = self._racks.get_racks_borders() onpoint, rack_name = self.find_nearest_rack_from_point(self._tol_connect, self._connect_step, pos, rack_borders) self._distributors_onpoints[dname] = (onpoint, rack_name) self.add_point_to_rack(onpoint, rack_name) # Füge "virtuelle Racks" von Sensor zu Aufpunkt von Sensor auf Rack hinzu. drackname = f"d-{dname}-{rack_name}" self._racks.add_rack(pos, onpoint, drackname) return self._distributors_onpoints def rack_segmentation(self, racks:dict) -> list[tuple[str, int, LineString]]: ''' Racks werden zu LineString konvertiert. Racks bestehend aus Polylinine werden in einzelne Segmente zerlegt und in Liste gesammelt. ''' rack_segments = [] for rack_id, nodes in racks.items(): # Sortiere Node_1, Node_2, ... sorted_keys = sorted(nodes.keys(), key=lambda k: int(k.split("_")[1])) coords = [tuple(nodes[k]) for k in sorted_keys] for i in range(len(coords) - 1): p1, p2 = coords[i], coords[i+1] line = LineString([p1, p2]) rack_segments.append((rack_id, i, line)) return(rack_segments) def find_rack_endpoints(self, rack_segments): ''' Endpunkte der Racks-Segmente werden in Points konvertiert und in Liste gesammelt''' segment_endpoints = [] for rack_id, idx, line in rack_segments: for pt in [line.coords[0], line.coords[1]]: segment_endpoints.append((rack_id, idx, Point(pt))) return(segment_endpoints) def join_racks(self): self._racks.join_racks() def increase_circle(self, tol, tol_step, line, pt, rack_id, idx, other_rack_id, other_idx, verbindungen, endpoint_pinned): ''' vergrößere Kreis bis Schnittpunkt mit Rack entsteht. Argumente: tol, tol_step -- Toleranz und Schittweite line -- linestring der entlang gelaufen wird rack_id, idx -- Rack_id und index von dem linestring stammt pt -- Punkt der Überprüft wird other_rack_id, other_idx -- Rack zu welchem der zu untersuchende Punkt gehört verbindungen -- Liste an die angefügt wird und die verbindungspunkte speichert endpoint_pinned -- Liste, die Rack und index von dem untersuchten Punkt und den neuen angepinnten Punkt speichert ''' radius = tol_step while radius <= tol: circle = pt.buffer(radius) if circle.intersects(line): contact = circle.intersection(line) if contact.geom_type == "Point": nearest = contact else: nearest = nearest_points(pt, contact)[1] #print(f" 🟡 Kreisberührung bei {nearest} mit {rack_id}_{idx}") verbindungen.append((rack_id, idx, other_rack_id, other_idx, nearest)) # Füge verschobenen Endpunkt zu Liste hinzu. [Punkt gehört zu Rack_Nr, alter Punkt, neuer Punkt, gepinnt an Target_Rack] endpoint_pinned.append((other_rack_id, other_idx, pt, nearest, rack_id)) break radius += tol_step def find_nearest_rack_from_point(self, max_dist, coarse_step, sensor:Point, racks:dict) -> tuple[Point, str]: # 1. grobe Kandidatensuche candidate_lines = [] radius = coarse_step rack_lines = dict() while radius <= max_dist: circle = sensor.buffer(radius) for r_name, pts in racks.items(): line = LineString([pts[0], pts[-1]]) #Linestring aus erstem und letzten Eintrag in Rack dict erzeugen if circle.intersects(line): candidate_lines.append((r_name, line)) if candidate_lines: break radius += coarse_step if not candidate_lines: return None, None # 2. Feinbestimmung über Distanz candidates_distance = [ (r_name, line, line.distance(sensor)) for r_name, line in candidate_lines ] # Sortieren nach Abstand candidates_distance.sort(key=lambda x: x[2]) '''# Theoretisch könnten mehrere ähnlich naheliegende Racks zurückgegeben werden.''' r_best, line_best, _ = candidates_distance[0] # Hier wird nur das tatsächlich dem Senso nächste Rack gegriffen # Aufpunkt bestimmen nearest_point = line_best.interpolate(line_best.project(sensor)) return (nearest_point, r_best) 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 generate_connected_racks(self, racks_json:dict[str, list[Point]]) -> dict: rack_segments = self.rack_segmentation(racks_json) rack_endpoints = self.find_rack_endpoints(rack_segments) # könnte man hier auch get_racks_borders nehmen? connected_racks = self.search_connections(rack_segments, rack_endpoints, self._tol_snap, self._snap_step) #Kann man diese Ausgabe jetzt nochmal in sowas wie add_Racks aufrufen um "eingelesene Racks" zu überscheiben? self._racks.add_racks(connected_racks) return connected_racks 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 pos = dict() for node in G.nodes: point = self._nodeids.get_point(node) pos[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) if re.match("v-.*", rname): color = "red" elif re.match("d-.*", rname): color = "blue" else: color = "black" G.add_edge(nid_start, nid_end, color=color, weight=start.distance(end)) return pos 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): quelle = self._nodeids.get_id(self.get_distributor_point(dname)) ziel = self._nodeids.get_id(self.get_sensor_point(sname)) print(self.get_distributor_point(dname), dname, quelle) print(self.get_sensor_point(sname), sname, ziel) 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): pfade = dict() for sname, dname in self._sensor2dist.items(): pfad_nodes, pfad_length = self.create_cable_path(G, sname, dname) pfad_coords = self._nodeids.get_points(pfad_nodes) pfade[f"{dname}-{sname}"] = {"path_nodes": pfad_nodes, "path_coords": pfad_coords, "path_length": pfad_length} return pfade def show_node_ids(self): return self._nodeids.show() class TestLinesweep(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_linesweep(self): ''' Prüft ob aus ungeanuen Endpunkten von Racks innerhalb einer Json ein neues Rack-Gerüst mit aufeinander Liegenden Endpunkten auf Racks erzeugt wird. ''' tol = 200 tol_step = 10 racks_json_str= '''{ "Rack_1": { "Node_1": [ 4946.5, 15774.4 ], "Node_2": [ 4946.5, 3879.4 ] }, "Rack_2": { "Node_1": [ 0.1, 57.6 ], "Node_2": [ 0.1, 3777.6 ], "Node_3": [ 14755.1, 3777.6 ] }, "Rack_3": { "Node_1": [ 185.1, 15865.5 ], "Node_2": [ 12450.7, 15865.5 ] }, "Rack_4": { "Node_1": [ 2866.6, 15774.4 ], "Node_2": [ 2866.6, 3880.4 ] }, "Rack_5": { "Node_1": [ 8866.1, 15774.4 ], "Node_2": [ 8866.1, 3878.4 ] }}''' racks_json = json.loads(racks_json_str) an = Anlage() connected_racks = an.generate_connected_racks(racks_json) res_rack_seg = {'Rack_1-0': [Point(4946.5, 15865.5), Point(4946.5, 3777.6)], 'Rack_2-0': [Point(0.1, 57.6), Point(0.1, 3777.6)], 'Rack_2-1': [Point(0.1, 3777.6), Point(14755.1, 3777.6)], 'Rack_3-0': [Point(185.1, 15865.5), Point(12450.7, 15865.5)], 'Rack_4-0': [Point(2866.6, 15865.5), Point(2866.6, 3777.6)], 'Rack_5-0': [Point(8866.1, 15865.5), Point(8866.1, 3777.6)] } self.assertEqual(connected_racks, res_rack_seg) 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 Schniuttpunkten und füge Schnittpunkte (exakt & beinahe) zu jeweiligem Rack hinzu rack.join_racks() 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() # 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 Schniuttpunkten und füge Schnittpunkte (exakt & beinahe) zu jeweiligem Rack hinzu rack.join_racks() 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)]} point2rack = RackIDs(res_rack_seg) self.assertEqual(point2rack.get_racks_from_point(Point(1, 0)), ["Rack_1-0", "Rack_2-0"]) self.assertEqual(point2rack.get_racks_from_point(Point(5, 6)), ["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 inzufügen und einsortieren eines Zwischenpunktes zwische nRack-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(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_generate_graph(self): '''Generiert einen Graphen in 3 unterschiedlichen Ausbauestufen (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) 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() 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() def test_Wegsuche(self): ''' Erstellt Graphen mit Racks, Sensoren und Unterverteilern und sucht kürzeste Wege von Unterverteiler zu zugehörigen Sensoren''' 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)} mapping = {'Dist_1': ['Sens_1', 'Sens_2'], 'Dist_2': ['Sens_3']} # 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) # 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) self.assertEqual(paths['Dist_1-Sens_1']["path_coords"], [Point(-1, 9), Point(0, 9), Point(0, 3), Point(0, 1), Point(1, 1)]) self.assertEqual(paths['Dist_1-Sens_2']["path_coords"], [Point(-1, 9), Point(0, 9), Point(0, 3), Point(2, 3), Point(2, 4)]) self.assertEqual(paths['Dist_2-Sens_3']["path_coords"], [Point(11, 0), Point(10, 0), Point(10, 2), Point(9, 2)]) self.assertEqual(paths['Dist_1-Sens_1']["path_length"], 10) self.assertEqual(paths['Dist_1-Sens_2']["path_length"], 10) self.assertEqual(paths['Dist_2-Sens_3']["path_length"], 4) if __name__ == '__main__': unittest.main()