Michael Stangl
943 lines
35 KiB
Python
943 lines
35 KiB
Python
import json
|
|
from shapely.geometry import LineString, Point, box
|
|
from shapely.ops import nearest_points
|
|
import unittest
|
|
from collections import defaultdict
|
|
import networkx as nx
|
|
import matplotlib.pyplot as plt
|
|
from itertools import pairwise
|
|
import re
|
|
from shapely.strtree import STRtree
|
|
import math
|
|
import shapely
|
|
|
|
from utils import to_json
|
|
|
|
# Globale Variable, die in main aufgerufen wird und steuert ob Graphen in unittests gezeichnet werden
|
|
draw = False
|
|
class PointSorter:
|
|
''' Klasse, die Punkte sortiert.
|
|
Die Punkte werden in der Liste self.points gespeichert.
|
|
Die Punkte werden sortiert nach x- und y-Koordinate.
|
|
'''
|
|
def __init__(self):
|
|
self.points = []
|
|
|
|
def add_point(self, point:Point):
|
|
self.points.append(point)
|
|
|
|
def add_points(self, points:list[Point]):
|
|
for p in points:
|
|
self.add_point(p)
|
|
|
|
def get_sorted_by_x(self):
|
|
return sorted(self.points, key = lambda p: p.x)
|
|
|
|
def get_sorted_by_y(self):
|
|
return sorted(self.points, key = lambda p: p.y)
|
|
|
|
class NodeIDs():
|
|
''' Klasse, die Punkte verwaltet und NodeIDs zu Punkten zuordnet.
|
|
Die NodeIDs sind ganze Zahlen, die die Position der Punkte in der Liste self.points repräsentieren.
|
|
'''
|
|
def __init__(self, points=[]):
|
|
self._counter = 0
|
|
self._cord2id = dict()
|
|
self._id2cord = dict()
|
|
self.add_points(points)
|
|
|
|
def add_point(self, point:Point):
|
|
''' Fügt den Punkt unter einer neuen NodeId hinzu.
|
|
'''
|
|
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"Point not found!, {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 nnot found! {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]:
|
|
''' Gibt zu einer Liste von NodeIDs die zugehörigen Punkte zurück.
|
|
Die Punkte werden in der gleichen Reihenfolge wie die NodeIDs zurückgegeben.
|
|
'''
|
|
ret = list()
|
|
for n in nids:
|
|
c = self.get_point(n)
|
|
ret.append(c)
|
|
return ret
|
|
|
|
def show(self):
|
|
return self._id2cord
|
|
class RackIDs():
|
|
''' Klasse, die Racks verwaltet und Rack-Racks und Rack-Equipment miteinander verbindet.
|
|
Die Racks werden als Linien in den STR-Baum eingefügt.
|
|
'''
|
|
def __init__(self, tol_snap = 200.0):
|
|
self._point2rack = dict()
|
|
self._rack2begend = dict()
|
|
# Toleranzen zur Rack anbindung aneinander (Rack Snap)
|
|
self._tol_snap = tol_snap
|
|
# 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):
|
|
''' Fügt einen Rack hinzu.
|
|
Fügt die Anfangs- und Endpunkte des Racks zu den Racks hinzu.
|
|
'''
|
|
self.add_point_to_rack(beg, name)
|
|
self.add_point_to_rack(end, name)
|
|
|
|
self._rack2begend[name] = (beg, end) # Anfangs und Endpunkte zu Rack Namen merken
|
|
|
|
def add_racks(self, racks:dict):
|
|
''' Fügt Racks hinzu.
|
|
Fügt die Anfangs- und Endpunkte der Racks zu den Racks hinzu.
|
|
'''
|
|
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):
|
|
''' Fügt einen Punkt zu genanntem Rack hinzu. (z.B Zwischenpunkt in der Mitte)'''
|
|
if point not in self._point2rack:
|
|
self._point2rack[point] = set()
|
|
self._point2rack[point].add(name)
|
|
|
|
def get_racks_from_point(self, point:Point) -> set[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()
|
|
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 _build_rack_strtree(self):
|
|
''' Erzeugt einen STR-Baum aus den Racks.
|
|
Die Racks werden als Linien in den STR-Baum eingefügt.
|
|
'''
|
|
self._rack_lines = []
|
|
self._rack_map = {}
|
|
for r_name, pts in self.get_racks_borders().items():
|
|
line = LineString([pts[0], pts[-1]])
|
|
self._rack_lines.append(line)
|
|
self._rack_map[line] = r_name
|
|
self._rack_tree = STRtree(self._rack_lines)
|
|
|
|
def join_racks_str(self):
|
|
''' Verbindet Racks miteinander.
|
|
Die Racks werden als Linien in den STR-Baum eingefügt.
|
|
'''
|
|
if self._rack_tree is None:
|
|
self._build_rack_strtree()
|
|
|
|
rack_tree = self._rack_tree
|
|
rnames = self._rack_map
|
|
allracks = self._rack_lines
|
|
|
|
# Erzeugung von BoundingBox
|
|
for i, l1 in enumerate(allracks):
|
|
bbox = box(*l1.bounds).buffer(self._tol_snap)
|
|
candidates = rack_tree.query(bbox)
|
|
candidates = [self._rack_lines[idx] for idx in candidates]
|
|
|
|
for l2 in candidates:
|
|
if l1.equals(l2):
|
|
continue
|
|
# Echte Schnittpunkte
|
|
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])
|
|
continue
|
|
|
|
# Beinahe Schnittpunkte -> Snapping
|
|
for pt in [Point(l2.coords[0]), Point(l2.coords[-1])]:
|
|
if l1.distance(pt) <= self._tol_snap:
|
|
snap_point = l1.interpolate(l1.project(pt))
|
|
self.add_point_to_rack(snap_point, rnames[l1])
|
|
connrackname = f"c-{rnames[l2]}-{rnames[l1]}"
|
|
self.add_rack(pt, snap_point, connrackname)
|
|
|
|
def rack_is_horizontal(self, name):
|
|
''' Gibt True zurück, wenn der Rack horizontal ist.
|
|
False, wenn der Rack vertikal ist.
|
|
'''
|
|
[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.0, tol_connect=5000.0):
|
|
# Container für alle Racks
|
|
self._racks = RackIDs(tol_snap=tol_snap)
|
|
self._rack_lengths = defaultdict(float)
|
|
# 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 Tunnel
|
|
self._tunnels = dict()
|
|
self._tunnel_onpoints = dict()
|
|
self._tunnel_lengths = dict()
|
|
#Container für alle Wege
|
|
self._sensor2dist = dict()
|
|
# Container für Sensor Artikelnummern
|
|
self._sensor_artnrs = dict()
|
|
# Toleranzen zur Rack anbindung aneinander (Rack Snap)
|
|
self._tol_snap = tol_snap
|
|
# Toleranzen zur Anbindung von Sensoren / Verteilern zu Racks
|
|
self._tol_connect = tol_connect
|
|
# 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.
|
|
|
|
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):
|
|
'''gibt die Aufpunkte aller Sensoren auf den Racks zurück'''
|
|
return self._sensors.values()
|
|
|
|
def get_distributor_onpoints(self):
|
|
'''gibt die Aufpunkte aller Unterverteiler auf den Racks zurück'''
|
|
return self._distributors_onpoints.values()
|
|
|
|
def get_sensor_onpoints(self):
|
|
'''gibt die Aufpunkte aller Sensoren auf den Racks zurück'''
|
|
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:
|
|
if sname in self._sensors:
|
|
return self._sensors[sname]
|
|
raise Exception("Sensor not found")
|
|
|
|
def set_sensor_artnrs(self, artnr_dict: dict):
|
|
self._sensor_artnrs = artnr_dict
|
|
|
|
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
|
|
'''
|
|
return self.connect_equipment_to_racks(self._sensors, 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:
|
|
if dname in self._distributors:
|
|
return self._distributors[dname]
|
|
raise Exception("Distributor not found")
|
|
|
|
def connect_distributor_to_racks(self) -> list:
|
|
'''verbindet die Unterverteiler mit den Racks
|
|
die Rückage enthält ein Tuple, welche Unterverteiler keinem Rack zugeordnet werden konnten
|
|
'''
|
|
return self.connect_equipment_to_racks(self._distributors, self._distributors_onpoints)
|
|
|
|
def get_tunnel_onpoints(self):
|
|
return self._tunnel_onpoints.values()
|
|
|
|
|
|
def add_tunnel(self, tname: str, pos:Point):
|
|
self._tunnels[tname] = pos
|
|
|
|
def set_tunnel_length(self, tunlength: dict):
|
|
self._tunnel_lengths = tunlength
|
|
|
|
def get_tunnel_length(self, tname: str):
|
|
if tname in self._tunnels:
|
|
return float(self._tunnel_lengths[tname])
|
|
raise Exception("Tunnel-Length not found")
|
|
|
|
def add_tunnels(self, tunnels:dict):
|
|
for tname,pos in tunnels.items():
|
|
self.add_tunnel(tname, pos)
|
|
|
|
def get_tunnel_point(self, tname:str) -> Point:
|
|
if tname in self._tunnels:
|
|
return self._tunnels[tname]
|
|
raise Exception("Tunnel not found")
|
|
|
|
def connect_tunnels(self) -> list:
|
|
'''verbindet die Anfangs und Endpunkte der Tunnel miteiandner als t-Rack
|
|
Verbindet sowohl Tunnel anfang als auch Tunnel ende jeweils mit nahegelgenem Rack (standard Aufruf wie Sensoren / Distributoren)
|
|
die Rückage enthält ein Tuple, welche Tunnel keinem Rack zugeordnet werden konnten
|
|
'''
|
|
for tname, pos in self._tunnels.items():
|
|
if len(pos) == 2:
|
|
tunbeg = pos[0]
|
|
tunend = pos[1]
|
|
trackname = f"t-Rack-{tname}"
|
|
self._racks.add_rack(tunbeg, tunend, trackname)
|
|
else:
|
|
raise Exception("Tunnel has more than one begin and end/ just one begin")
|
|
errors = list()
|
|
for name, pos in self._tunnels.items():
|
|
p1 = self._tunnels[name][0]
|
|
p2 = self._tunnels[name][1]
|
|
|
|
ret = self.connect_equipment_to_racks({f"{name}-A": p1}, self._tunnel_onpoints)
|
|
if ret:
|
|
errors.append(ret)
|
|
ret = self.connect_equipment_to_racks({f"{name}-E": p2}, self._tunnel_onpoints)
|
|
if ret:
|
|
errors.append(ret)
|
|
|
|
return errors
|
|
|
|
|
|
def coord_is_in_tunnel(self, coord):
|
|
for tname, pos in self._tunnels.items():
|
|
if coord == pos[0]:
|
|
return True
|
|
elif coord == pos[1]:
|
|
return True
|
|
return False
|
|
|
|
def join_racks(self):
|
|
self._racks.join_racks_str()
|
|
|
|
def get_rack_lengths(self) -> dict[str, float]:
|
|
return self._rack_lengths
|
|
|
|
def find_nearest_rack_from_point_STR_bbox(self, max_dist, pt:Point) -> tuple[Point, str]:
|
|
if self._racks._rack_tree is None:
|
|
self._racks._build_rack_strtree()
|
|
|
|
minx, miny, maxx, maxy = pt.x - max_dist, pt.y - max_dist, pt.x + max_dist, pt.y + max_dist
|
|
bbox = box(minx, miny, maxx, maxy)
|
|
|
|
candidates = self._racks._rack_tree.query(bbox)
|
|
if len(candidates) == 0:
|
|
raise LookupError("no candidates in box found")
|
|
|
|
candidates = [self._racks._rack_lines[idx] for idx in candidates]
|
|
best_dist = max_dist
|
|
best_line = candidates[0]
|
|
for line in candidates:
|
|
dist = pt.distance(line)
|
|
if dist < best_dist:
|
|
best_dist = dist
|
|
best_line = line
|
|
|
|
rack_name = self._racks._rack_map[best_line]
|
|
nearest_point = best_line.interpolate(best_line.project(pt))
|
|
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 ({'Name': Point}), Dict der Aufpunkte von Sensoren o. Dists (zu Beginn leer)
|
|
Rückgabe: Liste der nicht mit Racks verbundenen Geräte (z.B. Entfernung zu groß)
|
|
'''
|
|
errors = list()
|
|
for name, pos in equipment.items():
|
|
try:
|
|
onpoint, rackname = self.find_nearest_rack_from_point_STR_bbox(self._tol_connect, pos)
|
|
onpoints[name] = (onpoint, rackname)
|
|
except LookupError:
|
|
errors.append({"name": name, "coords": {"x":pos.x, "y":pos.y}}) # Name des fehlgeschlagenenen und Position als Koodinaten zurückgeben
|
|
continue
|
|
self.add_point_to_rack(onpoint, rackname)
|
|
|
|
virtual_rackname = f"v-{name}-{rackname}"
|
|
self._racks.add_rack(pos, onpoint, virtual_rackname)
|
|
|
|
return errors
|
|
|
|
def get_node_positions(self):
|
|
''' Daten werden durch generate_graph() befüllt'''
|
|
return self._node_positions
|
|
|
|
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
|
|
|
|
|
|
for node in G.nodes:
|
|
point = self._nodeids.get_point(node)
|
|
self._node_positions[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)
|
|
|
|
dx = start.x - end.x
|
|
dy = start.y - end.y
|
|
|
|
if shapely.has_z(start) and shapely.has_z(end):
|
|
dz = (getattr(start, "z", 0) - getattr(end, "z", 0))
|
|
else:
|
|
dz = 0.0
|
|
|
|
weight = math.sqrt(dx**2 + dy**2 + dz**2)
|
|
|
|
if re.match("v-.*", rname):
|
|
color = "red"
|
|
elif re.match("d-.*", rname):
|
|
color = "blue"
|
|
elif re.match("t-.*", rname):
|
|
color = "orange"
|
|
tname = rname.split("-", 2)[2]
|
|
weight = self.get_tunnel_length(tname)
|
|
elif re.match("c-.*", rname):
|
|
color = "green"
|
|
else:
|
|
color = "black"
|
|
|
|
G.add_edge(nid_start, nid_end, color=color, weight=weight)
|
|
|
|
self._rack_lengths[rname] += round(weight/1000, 1)
|
|
|
|
return self._node_positions
|
|
|
|
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) -> tuple:
|
|
quelle = self._nodeids.get_id(self.get_distributor_point(dname))
|
|
ziel = self._nodeids.get_id(self.get_sensor_point(sname))
|
|
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) -> dict:
|
|
pfade = dict()
|
|
pfade["kabel"] = list()
|
|
|
|
routing_errors = dict() # Fehler Liste für fehlgeschlagene Kabelverbindungen
|
|
for sname, dname in self._sensor2dist.items():
|
|
try:
|
|
pfad_nodes, pfad_length = self.create_cable_path(G, sname, dname)
|
|
except:
|
|
if dname not in routing_errors:
|
|
routing_errors[dname] = list()
|
|
routing_errors[dname].append(sname)
|
|
continue
|
|
pfad_coords = self._nodeids.get_points(pfad_nodes)
|
|
|
|
ld = dict()
|
|
|
|
ld['id'] = f"{dname}-{sname}"
|
|
if len(self._sensor_artnrs) > 0:
|
|
ld['s_artinr'] = self._sensor_artnrs.get(sname, "")
|
|
ld["coords"]= [{"x":round(p.x,1), "y":round(p.y,1)} for p in pfad_coords]
|
|
ld["length"]= round(pfad_length,1)
|
|
#ld["nodes"]=pfad_nodes
|
|
|
|
ld['tunnel_indices'] = list()
|
|
for i, p in enumerate(pfad_coords):
|
|
if self.coord_is_in_tunnel(p):
|
|
ld['tunnel_indices'].append(i)
|
|
|
|
pfade["kabel"].append(ld)
|
|
|
|
pfade["rack_geometry"] = {
|
|
rname: {
|
|
"length": round(length, 1),
|
|
"coordinates": [
|
|
{
|
|
"x": round(p.x, 1),
|
|
"y": round(p.y, 1),
|
|
"z": round(p.z if shapely.has_z(p) else 0.0, 1)
|
|
}
|
|
for p in self.get_points_from_rack(rname)
|
|
]
|
|
}
|
|
for rname, length in self._rack_lengths.items()
|
|
}
|
|
|
|
pfade["errors_routing"] = dict()
|
|
lOfDistErrors = list() # sammeln wo keine Verbindung zwischen Dist und Senor möglich war
|
|
for dname,lofSensornamen in routing_errors.items():
|
|
lOfDistErrors.append({"unterverteiler":dname, "sensoren":lofSensornamen})
|
|
|
|
pfade["errors_routing"] = lOfDistErrors
|
|
|
|
return pfade
|
|
|
|
def show_node_ids(self):
|
|
return self._nodeids.show()
|
|
|
|
|
|
|
|
class TestPlant(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_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 Schnittpunkten und füge Schnittpunkte (exakt & beinahe) zu jeweiligem Rack hinzu
|
|
rack.join_racks_str()
|
|
|
|
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(tol_snap=1)
|
|
# 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_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)])
|
|
|
|
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)]}
|
|
|
|
# 10 |R2-1: ####################
|
|
# 9 |
|
|
# 8 | |
|
|
# 7 | |
|
|
# 6 | | #### R1-0
|
|
# 5 | | #
|
|
# 4 | | #
|
|
# 3 | | #
|
|
# 2 | | #
|
|
# 1 | |#
|
|
# 0 |-----#--------------------
|
|
# 0 1 2 3 4 5
|
|
|
|
|
|
point2rack = RackIDs()
|
|
point2rack.add_racks(res_rack_seg)
|
|
|
|
t1 = set(point2rack.get_racks_from_point(Point(1, 0)))
|
|
self.assertEqual(t1, set(["Rack_1-0", "Rack_2-0"]))
|
|
|
|
t2 = point2rack.get_racks_from_point(Point(5, 6))
|
|
self.assertEqual(t2, set(["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 hinzufügen und einsortieren eines Zwischenpunktes zwischen Rack-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()
|
|
point2rack.add_racks(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_add_equipment_str_tree(self):
|
|
|
|
racks = {'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)}
|
|
|
|
an = Anlage(tol_snap=1.5, tol_connect=1.5)
|
|
an.set_racks(racks)
|
|
an.join_racks()
|
|
|
|
an.add_sensors(sensors)
|
|
an.add_distributors(distributors)
|
|
an.connect_equipment_to_racks(an._sensors, an._sensor_onpoints)
|
|
an.connect_equipment_to_racks(an._distributors, an._distributors_onpoints)
|
|
|
|
plist1 = an.get_points_from_rack("Rack_1")
|
|
plist2 = an.get_points_from_rack("Rack_2")
|
|
|
|
|
|
self.assertEqual(plist1, [Point(0, 0), Point(0, 1), Point(0, 3), Point(0, 9), Point(0, 10)])
|
|
self.assertEqual(plist2, [Point(10, -2), Point(10, 0), Point(10, 2), Point(10, 3), Point(10, 5)])
|
|
|
|
def test_wegsuche_str_tree(self):
|
|
racks = {'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(tol_snap=1)
|
|
an.set_racks(racks)
|
|
an.join_racks()
|
|
|
|
an.add_sensors(sensors)
|
|
an.add_distributors(distributors)
|
|
an.connect_equipment_to_racks(an._sensors, an._sensor_onpoints)
|
|
an.connect_equipment_to_racks(an._distributors, an._distributors_onpoints)
|
|
|
|
an.map_distributors_to_sensors(mapping)
|
|
|
|
G = nx.Graph()
|
|
# Fülle eben erstellten Graphen mit Daten
|
|
pos = an.generate_graph(G)
|
|
# Extrahiere Farb-Informationen der Kanten
|
|
edge_colors = [G[u][v].get('color', 'black') for u, v in G.edges()]
|
|
# Zeiche Graphen und zeige in
|
|
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)
|
|
|
|
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)
|
|
|
|
def test_generate_graph(self):
|
|
'''Generiert einen Graphen in 3 unterschiedlichen Ausbaustufen (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)
|
|
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()]
|
|
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()]
|
|
if draw:
|
|
nx.draw(G3, pos, with_labels=False, node_size=10, font_size=8, edge_color=edge_colors)
|
|
plt.show()
|
|
|
|
|
|
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'))
|
|
suite.addTest(TestPlant('test_intersect_segments'))
|
|
suite.addTest(TestPlant('test_snap_segments'))
|
|
suite.addTest(TestPlant('test_ids_to_point'))
|
|
suite.addTest(TestPlant('test_add_point_interim'))
|
|
suite.addTest(TestPlant('test_add_sensor'))
|
|
suite.addTest(TestPlant('test_add_equipment_str_tree'))
|
|
suite.addTest(TestPlant('test_wegsuche_str_tree'))
|
|
suite.addTest(TestPlant('test_generate_graph'))
|
|
runner = unittest.TextTestRunner()
|
|
runner.run(suite)
|
|
#unittest.main()
|