Compare commits

...

2 Commits

+315 -86
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,21 +214,63 @@ 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()
# 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]]):
return self._racks.add_racks(racks)
@@ -184,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():
@@ -200,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()
@@ -214,6 +307,9 @@ class Anlage():
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()
@@ -232,6 +328,9 @@ class Anlage():
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()
@@ -444,7 +543,7 @@ class Anlage():
points.extend(self.get_points_from_sensors())
nodeids = NodeIDs(points)
self._nodeids.add_points(points)
for p in points:
if self.is_distributor(p):
@@ -453,19 +552,19 @@ class Anlage():
shape = "^"
else:
shape = "o"
nid = nodeids.get_id(p)
nid = self._nodeids.get_id(p)
G.add_node(nid, shape=shape) # Knoten für Startpunkt
pos = dict()
for node in G.nodes:
point = nodeids.get_point(node)
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 = nodeids.get_id(start)
nid_end = nodeids.get_id(end)
nid_start = self._nodeids.get_id(start)
nid_end = self._nodeids.get_id(end)
if re.match("v-.*", rname):
color = "red"
@@ -476,15 +575,56 @@ class Anlage():
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)
pfade[f"{dname}-{sname}"] = {"pfad": pfad_nodes, "laenge": pfad_length}
def show_node_ids(self):
return self._nodeids.show()
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.
'''
@@ -529,104 +669,193 @@ 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):
# 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']}
# 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)
# 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, "")
if __name__ == '__main__':
unittest.main()