m.stangl/kabellaengen
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Einlesen der Dateien in routing.py angepasst, sodass plant.py Daten verarbeiten kann.

Browse Source
This commit is contained in:
lertlmaier
2025-05-16 17:53:34 +02:00
parent b8b5db45eb
commit b4149870f6
6 changed files with 126 additions and 212 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.vscode/launch.json
Vendored
+27 -29
View File
@@ -1,10 +1,13 @@
{
// Verwendet IntelliSense zum Ermitteln möglicher Attribute.
// Zeigen Sie auf vorhandene Attribute, um die zugehörigen Beschreibungen anzuzeigen.
// Weitere Informationen finden Sie unter https://go.microsoft.com/fwlink/?linkid=830387
"version": "0.2.0",
"configurations": [
{
"name": "Python Debugger: Current File",
"type": "debugpy",
"request": "launch",
"program": "${file}",
"console": "integratedTerminal"
},
{
"name": "Python-Debugger: Aktuelle Datei mit Argumenten",
"type": "debugpy",
@@ -18,24 +21,22 @@
"type": "debugpy",
"request": "launch",
"program": "${file}",
"console": "integratedTerminal",
"justMyCode": true
"console": "integratedTerminal"
},
{
"name": "use easy.dxf",
"name": "getpositions with easy.dxf",
"type": "debugpy",
"request": "launch",
"program": "${file}",
"console": "integratedTerminal",
"justMyCode": true,
"args": [
"--filename",
"easy.dxf",
"-s",
"-d",
"-r",
"-c",
"-w"
"--sensors",
"--dists",
"--rack",
"--console",
"--write"
]
},
{
@@ -44,12 +45,12 @@
"request": "launch",
"program": "${file}",
"console": "integratedTerminal",
"justMyCode": true,
"args": [
"--filename",
"easy.dxf",
"-r",
"-c"
"-c",
"-w"
]
},
{
@@ -58,7 +59,6 @@
"request": "launch",
"program": "${file}",
"console": "integratedTerminal",
"justMyCode": true,
"args": [
"--filename",
"ST_6300_Steuerungstestlayout1_neueBloecke.dxf",
@@ -67,31 +67,29 @@
"-r"
]
},
{
"name": "run routing for easy",
"type": "debugpy",
"request": "launch",
"program": "${file}",
"console": "integratedTerminal",
"justMyCode": true,
"args": [
"--inputfile",
"easy_positions.json"
]
},
{
"name": "draw cable dxf from easy.json",
"type": "debugpy",
"request": "launch",
"program": "${file}",
"console": "integratedTerminal",
"justMyCode": true,
"args": [
"--json",
"easy.json",
"--dxf",
"easy.dxf"
]
},
{
"name": "routing for easy_positions.json",
"type": "debugpy",
"request": "launch",
"program": "${file}",
"console": "integratedTerminal",
"args": [
"--filename",
"easy_positions.json"
]
}
]
}
bin/draw_dxf.bat
+1 -1
View File
@@ -1,4 +1,4 @@
@echo off
CALL manage_interpreter.bat activate_interpreter
python %PROJECT_LIB%\drawdxf.py
python %PROJECT_LIB%\drawdxf.py %*
CALL manage_interpreter.bat deactivate_interpreter
bin/routing.bat
+4
View File
@@ -0,0 +1,4 @@
@echo off
CALL manage_interpreter.bat activate_interpreter
python %PROJECT_LIB%\routing.py %*
CALL manage_interpreter.bat deactivate_interpreter
lib/getpositions.py
+4 -6
View File
@@ -7,6 +7,7 @@ import sys
import math
import json
import re
from shapely import Point
"""
@@ -132,13 +133,10 @@ def get_rack_positions(msp):
for e in msp.query('LWPOLYLINE[layer=="PRITSCHE_200-60"]'):
#print_polyline(e)
rack_key = f"Rack_{rack_counter}"
ret[rack_key] = dict()
node_counter = 1
ret[rack_key] = list()
for x, y, start_width, end_width, bulge in e.get_points(): # Gibt Tuple (x, y, start_width, end_width, bulge)
node_key = f"Node_{node_counter}"
ret[rack_key][node_key] = [round(x,1), round(y,1)]
node_counter +=1
p = [round(x,1), round(y,1)]
ret[rack_key].append(p)
rack_counter +=1
# iterate over all entities in modelspace
# for e in msp:
lib/linesweep_circle.py → lib/plant.py
+10 -116
View File
@@ -279,6 +279,8 @@ class Anlage():
# Toleranzen zur Anbindung von Sensoren / Verteilern zu Racks
self._tol_connect = tol_connect
self._connect_step = tol_connect_step
# Infos zum zeichnen des Graphen
self._node_positions = dict()
def set_racks(self, racks:dict[str, list[Point]]):
@@ -375,64 +377,9 @@ class Anlage():
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 = []
@@ -551,14 +498,10 @@ class Anlage():
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 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():
@@ -574,11 +517,8 @@ class Anlage():
def generate_graph(self, G:nx.Graph):
points = list()
points.extend(self.get_all_rack_points())
points.extend(self.get_points_from_sensors())
points.extend(self.get_points_from_sensors())
self._nodeids.add_points(points)
for p in points:
@@ -591,10 +531,10 @@ class Anlage():
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)
self._node_positions[node] = (point.x, point.y)
for rname in self.get_rack_names():
plist = self.get_points_from_rack(rname)
@@ -609,7 +549,7 @@ class Anlage():
else:
color = "black"
G.add_edge(nid_start, nid_end, color=color, weight=start.distance(end))
return pos
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.
@@ -629,8 +569,6 @@ class Anlage():
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')
@@ -668,50 +606,6 @@ class TestLinesweep(unittest.TestCase):
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.'''
lib/routing.py
+80 -60
View File
@@ -7,42 +7,92 @@ import matplotlib.pyplot as plt
import networkx as nx
from shapely.geometry import LineString, Point
from shapely.ops import nearest_points
from plant import Anlage
# Funktionen
def load_json(jsonfilename):
with open(jsonfilename, encoding='utf-8') as fh:
return json.load(fh)
def rack_segmentation(racks):
''' Racks werden zu LineString konvertiert. Racks bestehend aus Polylinine werden in einzelne Segmente zerlegt.'''
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]
def create_plant(racks:dict, sensors:dict, distributors:dict, mapping:dict):
# 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)]}
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))
# sensors = {'Sens_1': Point(1, 1),
# 'Sens_2': Point(2, 4),
# 'Sens_3': Point(9, 2)}
return(rack_segments)
# distributors = {'Dist_1': Point(-1, 9),
# 'Dist_2': Point(11, 0)}
# mapping = {'Dist_1': ['Sens_1', 'Sens_2'],
# 'Dist_2': ['Sens_3']}
def rack_endpoints(rack_segments)
''' Endpunkte der Racks werden in Points konvertiert'''
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)))
# "mapping": {
# "UC0101": [
# "BG3241",
# "BG3240",
# "MA0062",
# "FC0062"
# ]
# }
an = Anlage()
# Füge racks aus Daten hinzu
an.set_racks(racks)
# 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
G = nx.Graph()
# Fülle eben erstellten Graphen mit Daten
an.generate_graph(G)
# Ermittle kürzeste Wege von Unterverteilern zu zugehörigen Sensoren
paths = an.create_cable_paths(G)
if args.graph:
draw_graph(G,an)
return(segment_endpoints)
def pin_rack_endpoints(rack_segments, segment_endpoints)
'''Erstellung eines Dicts, welches unter dem Key "Rack_id - Index" die Endpunkte der Racks speichert. Endpunkte von Racks innerhalb der Toleranz werden an nahegelegenes Rack gepinnt.'''
def draw_graph(G:nx.Graph, an:Anlage):
pos = an.get_node_positions()
edge_colors = [G[u][v].get('color', 'black') for u, v in G.edges()]
nx.draw(G, pos, with_labels=False, node_size=10, font_size=8, edge_color=edge_colors)
plt.show()
def prepare_data(rawdata:dict):
sensors = data["sensors"]
subdists = data["distributors"]
racks = data["racks"]
mapping = data["mapping"]
dracks = dict()
for rname,lp in racks:
if rname not in dracks:
dracks[rname] = list()
dracks[rname].append(Point(lp))
return (sensors, subdists, dracks, mapping)
if __name__ == "__main__":
parser = argparse.ArgumentParser(description='Berechne Wege von Sensoren zu Verteilern über Kabeltrassen')
parser.add_argument('-i', '--inputfile', action='store', required=True, default="easy_position.json", help='file with all informations about positions gathered from getpositions', metavar='my_positions.json')
parser.add_argument('-f', '--filename', action='store', required=True, default="easy_position.json", help='file with all informations about positions gathered from getpositions', metavar='my_positions.json')
parser.add_argument('-c', '--console', action='store_true', help='Ausgabe auf Konsole')
parser.add_argument('-g', '--graph', action='store_true', help='Zeichnet den Graphen der Anlage')
args = parser.parse_args()
@@ -51,50 +101,20 @@ if __name__ == "__main__":
config_dir = os.environ.get("PROJECT_CFG")
# Pfade zu JSON-Dateien
jsonfilename = args.inputfile
jsonfilename = args.filename
sensors_path = os.path.join(work_dir, jsonfilename)
# Einlesen
rawdata = load_json(sensors_path)
(racks, sensors, subdists, mapping) = prepare_data(rawdata)
data = load_json(sensors_path)
sensors = data["sensors"]
subdists = data["distributors"]
racks = data["racks"]
# 1. Alle Kreuzungspunkte der Racks ermitteln
''' Funktion linesweep_circle wird aufgerufen.
Eingabe für Funktion: Nummerierte Racks aus getpositions.py. Reine Liste aus Koordinaten [((Anfang), (Ende)), ((Anfang), (Ende)), ... ] wird in Funktion erstellt
Racks werden abgelaufen und tatsächliche Schnittpunkte sowie Nahezu-Schnitte erfasst.
Ausgabe von Linesweep Circle: Liste von Endpunkten der Racks aus .dxf & Liste von SP von Rack_a mit Rack_b
plant = create_plant(racks, sensors, subdists, mapping)
# Erstelle Anlage
'''
# 2. Graph aus Racks erstellen
''' Zunächst leeren Graph erstellen.
Endpunkte der Racks wie in .dxf als Knoten hinzufügen. Kanten zwischen den Knoten ebenso hinzufügen.
Mittels ausgabe von linesweep_cirle: weitere Knoten auf Kanten hinzufügen und graph vollständig verknüpfen
Frage: Problematisch, wenn Knoten sehr nahe beieinander liegen also teoretischen "Doppelt" sind? -> lieber Endpunkte von Racks, die über linesweep_circle gefunden werden überschreiben?
Zusätzlich: über shapely und distance funktion sämtliche Distanzen bestimmen und Graph gewichten
'''
# 3. Sensoren mit nächstgelegenen Racks verknüpfen
''' Für jeden Sensor: Sensor pos aus json zu Graph hinzufügen
Für jeden Sensor: shapely.distance aufrufen und distance zu allen Racks bzw. Edges ermitteln. (if distance(a) > distance(b) -> Circle_methode um Sensor und Schnittpunkt zu Rack_b ermitteln)
SP von Sensor mit Rack als Node hinzufügen und Connection zwischen Sensor, Node, Edge herstellen
'''
# 4. Unterverteiler mit nächstegelegenen Racks verknüpfen
''' siehe 3.
'''
# 5. Bestimmung des küzesten Wegs
''' über Graphen-Toolbox z.B. nx wird für jeden Knoten zu dessen zugehörigen Unterverteilung der kürzeste Weg gefunden.
Schreiben der Ergebnisse in eine seperate kabel.json z.B.
'''