Methodik zur Aufsplittung von Polylines in einzelne Rack Segmente implementiert. Methodik zur verbindung von Racks an echten Scnittpunkten sowie an nahezu Schnittpunkten implementiert. Unittests zu jeder Methode

This commit is contained in:
2025-05-16 12:34:59 +02:00
parent e4d7902f86
commit 55459c73b1
+266 -110
View File
@@ -6,7 +6,7 @@ from collections import defaultdict
import bisect
import networkx as nx
import matplotlib.pyplot as plt
from itertools import pairwise
from itertools import pairwise, combinations
import re
class PointSorter:
@@ -60,19 +60,32 @@ class NodeIDs():
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[f"{self._counter}"] = point
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:
return self._id2cord[f"{nid}"]
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()
@@ -81,17 +94,26 @@ class NodeIDs():
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()):
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:
@@ -103,7 +125,17 @@ class RackIDs():
else:
self._point2rack[end] = [name]
self._rack2begend[name] = [beg, end] # Anfangs und Endpunkte zu Rack Namen merken
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.
@@ -118,13 +150,9 @@ class RackIDs():
return self._point2rack
def get_rack_names(self) -> list:
return self._rack2begend.keys()
return list(self._rack2begend.keys())
def add_racks(self, racks:dict):
for name,v in racks.items():
if len(v) != 2:
raise AttributeError
self.add_rack(v[0], v[1], name)
def add_point_to_rack(self, point:Point, name:str):
if point in self._point2rack:
@@ -144,13 +172,40 @@ class RackIDs():
ret.append(p)
pin.add_points(ret)
ret_sorted = list()
[pa, pe] = self._rack2begend[name]
#(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))
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:
@@ -159,6 +214,43 @@ class RackIDs():
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
----------
G : NetworkX graph
weight : string or function
If this is a string, then edge weights will be accessed via the
edge attribute with this key (that is, the weight of the edge
joining `u` to `v` will be ``G.edges[u, v][weight]``). If no
such edge attribute exists, the weight of the edge is assumed to
be one.
If this is a function, the weight of an edge is the value
returned by the function. The function must accept exactly three
positional arguments: the two endpoints of an edge and the
dictionary of edge attributes for that edge. The function must
return a number.
Returns
-------
distance : dictionary
Dictionary, keyed by source and target, of shortest paths.
Examples
--------
>>> graph = nx.DiGraph()
>>> graph.add_weighted_edges_from(
... [("0", "3", 3), ("0", "1", -5), ("0", "2", 2), ("1", "2", 4), ("2", "3", 1)]
... )
>>> paths = nx.johnson(graph, weight="weight")
>>> paths["0"]["2"]
['0', '1', '2']
"""
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()
@@ -189,9 +281,6 @@ class Anlage():
def add_point_to_rack(self, point:Point, rname:str):
return self._racks.add_point_to_rack(point, rname)
def get_points_from_rack(self, rname:str):
return self._racks.get_points_from_rack(rname)
def get_all_rack_points(self):
ret = list()
for rname in self._racks.get_rack_names():
@@ -205,7 +294,6 @@ class Anlage():
''' 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()
@@ -502,18 +590,24 @@ class Anlage():
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=1, target=5)
pfad_length = nx.shortest_path_length(G, source=quelle, target=ziel, weight='weight')
return pfad_nodes, pfad_length
def create_cable_paths(self, G):
for sname, dname in self._sensor2dist:
self.create_cable_path(G, sname, dname)
pfade = dict()
for sname, dname in self._sensor2dist.items():
pfad_nodes, pfad_length = self.create_cable_path(G, sname, dname)
pfade[f"{dname}-{sname}"] = {"pfad": pfad_nodes, "laenge": pfad_length}
def show_node_ids(self):
return self._nodeids.show()
@@ -522,7 +616,15 @@ class Anlage():
class TestLinesweep(unittest.TestCase):
def test_linesweep(self):
def test_duplicate_points(self):
nodeids = NodeIDs()
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.
'''
@@ -567,136 +669,190 @@ class TestLinesweep(unittest.TestCase):
self.assertEqual(connected_racks, res_rack_seg)
def test_ids_to_point(self):
''' Testet, ob gefragter Punkt auf Racks a, b, c liegt'''
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)]
}
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)]}
rack = RackIDs()
rack.add_racks(racks_data)
self.assertEqual(rack.get_rack_names(), ['Rack_1-1', 'Rack_1-2', 'Rack_2'])
point2rack = RackIDs(res_rack_seg)
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)],
}
rack = RackIDs()
rack.add_racks(racks_data)
rack.join_racks()
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)])
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)]
}
rack = RackIDs()
rack.add_racks(racks_data)
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'''
# 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)]}
# 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")
# 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)])
# 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).'''
# 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)]}
# 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)}
# 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)
# an = Anlage()
# point2rack = an.set_racks(rack_segs)
# an.add_sensors(sensors)
plist1 = an.get_points_from_rack("Rack_1-0")
# plist1 = an.get_points_from_rack("Rack_1-0")
an.connect_sensors_to_racks()
plist2 = 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)])
# 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):
# def test_generate_graph(self):
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)]}
# 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)}
# 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)}
# distributors = {'Dist_1': Point(-1, 9),
# 'Dist_2': Point(11, 0)}
an = Anlage()
an.set_racks(rack_segs)
# an = Anlage()
# an.set_racks(rack_segs)
G1 = nx.Graph()
pos = an.generate_graph(G1)
nx.draw(G1, pos, with_labels=False, node_size=10, font_size=8)
plt.show()
# 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()
# an.add_sensors(sensors)
# an.connect_sensors_to_racks()
G2 = nx.Graph()
pos = an.generate_graph(G2)
# 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()
# 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()
# an.add_distributors(distributors)
# an.connect_distributor_to_racks()
G3 = nx.Graph()
pos = an.generate_graph(G3)
# G3 = nx.Graph()
# pos = an.generate_graph(G3)
edge_colors = [G3[u][v].get('color', 'black') for u, v in G3.edges()]
# 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()
# nx.draw(G3, pos, with_labels=False, node_size=10, font_size=8, edge_color=edge_colors)
# plt.show()
def test_Wegsuche(self):
# def test_Wegsuche(self):
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)]}
# 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)}
# 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)}
# distributors = {'Dist_1': Point(-1, 9),
# 'Dist_2': Point(11, 0)}
mapping = {'Dist_1': ["Sens_1", "Sens_2"],
'Dist_2': ["Sens_3"]}
# mapping = {'Dist_1': ['Sens_1', 'Sens_2'],
# 'Dist_2': ['Sens_3']}
an = Anlage()
an.set_racks(rack_segs)
an.add_sensors(sensors)
an.connect_sensors_to_racks()
an.add_distributors(distributors)
an.connect_distributor_to_racks()
an.map_distributors_to_sensors(mapping)
G3 = nx.Graph()
pos = an.generate_graph(G3)
edge_colors = [G3[u][v].get('color', 'black') for u, v in G3.edges()]
# an = Anlage()
# an.set_racks(rack_segs)
# an.add_sensors(sensors)
# an.connect_sensors_to_racks()
# an.add_distributors(distributors)
# an.connect_distributor_to_racks()
# an.map_distributors_to_sensors(mapping)
nx.draw(G3, pos, with_labels=False, node_size=10, font_size=8, edge_color=edge_colors)
plt.show()
# G3 = nx.Graph()
# pos = an.generate_graph(G3)
# print(G3.nodes)
# print(G3.edges)
# print([(n, nbrdict) for n, nbrdict in G3.adjacency()])
# print(an.show_node_ids())
# 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()
# paths = an.create_cable_paths(G3)
# self.assertEqual(paths, "")