diff --git a/lib/graphbuild.py b/lib/graphbuild.py index b525af7..1ae6108 100644 --- a/lib/graphbuild.py +++ b/lib/graphbuild.py @@ -96,8 +96,8 @@ class TestShapely(unittest.TestCase): pos = {node: (node.x, node.y) for node in G.nodes()} # Positionen der Knoten aus den Punkten # Plotten des Graphen - #nx.draw(G, pos, with_labels=False, node_size=10, font_size=8) - #plt.show() + nx.draw(G, pos, with_labels=False, node_size=10, font_size=8) + plt.show() def test_easy(self): pass diff --git a/lib/linesweep_circle.py b/lib/linesweep_circle.py index 49d2484..9bf07ee 100644 --- a/lib/linesweep_circle.py +++ b/lib/linesweep_circle.py @@ -5,6 +5,9 @@ import unittest from collections import defaultdict import bisect import networkx as nx +import matplotlib.pyplot as plt +from itertools import pairwise +import re class PointSorter: def __init__(self): @@ -61,7 +64,7 @@ class NodeIDs(): self._cord2id[f"{point.x} {point.y}"] = self._counter self._id2cord[f"{self._counter}"] = point - def add_points(self, points): + def add_points(self, points:list[Point]): for p in points: self.add_point(p) @@ -114,7 +117,7 @@ class RackIDs(): ''' return self._point2rack - def get_rack_names(self): + def get_rack_names(self) -> list: return self._rack2begend.keys() def add_racks(self, racks:dict): @@ -156,14 +159,21 @@ class RackIDs(): return False class Anlage(): - def __init__(self, tol=200, tol_step=10): - #self._points = PointSorter() + 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() - #self._nodeids = NodeIDs() + # Container für alle Sensoren self._sensors = dict() self._sensor_onpoints = dict() - self._tol = tol - self._tol_step = tol_step + # Container für alle Unterverteiler + self._distributors = dict() + self._distributors_onpoints = 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]]): return self._racks.add_racks(racks) @@ -180,11 +190,16 @@ class Anlage(): def get_all_rack_points(self): ret = list() for rname in self._racks.get_rack_names(): - ret.append(self.get_points_from_rack(rname)) - s = set() - for r in ret: - s.add(r) - return s + 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() @@ -202,10 +217,31 @@ class Anlage(): 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_sensor(2, 0.5, pos, rack_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 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. @@ -262,7 +298,7 @@ class Anlage(): break radius += tol_step - def find_nearest_rack_from_sensor(self, max_dist, coarse_step, sensor:Point, racks:dict) -> tuple[Point, str]: + 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 @@ -385,33 +421,64 @@ class Anlage(): 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, self._tol_step) #Kann man diese Ausgabe jetzt nochmal in sowas wie add_Racks aufrufen um "eingelesene Racks" zu überscheiben? + 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 generate_graph(self): + 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() - G = nx.Graph() + + points.extend(self.get_all_rack_points()) - points.append(self.get_all_rack_points()) - - points.append(self.get_points_from_sensors()) + points.extend(self.get_points_from_sensors()) nodeids = NodeIDs(points) for p in points: + if self.is_distributor(p): + shape = "s" + elif self.is_sensor(p): + shape = "^" + else: + shape = "o" nid = nodeids.get_id(p) - G.add_node(nid) # Knoten für Startpunkt + G.add_node(nid, shape=shape) # Knoten für Startpunkt - # for p in points: - # pos = {nid: (p.x, p.y) for nid in G.nodes()} + pos = dict() + for node in G.nodes: + point = 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 = nodeids.get_id(start) + nid_end = 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 - pos = {node: (node.x, node.y) for node in G.nodes()} - #nx.draw(G, pos, with_labels=False, node_size=10, font_size=8) - return G + + @@ -527,19 +594,39 @@ class TestLinesweep(unittest.TestCase): 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) - graph_racks = an.generate_graph() + 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() - graph_racks_sensors = an.generate_graph() + 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() + if __name__ == '__main__': unittest.main() \ No newline at end of file