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

Klassendeklaration in linesweep_circle. Umbau des graphbuild auf Stuktur mit unittests

Browse Source
This commit is contained in:
lertlmaier
2025-05-13 13:48:49 +02:00
parent 58e121bdcf
commit 559a4690c8
2 changed files with 300 additions and 89 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/graphbuild.py
+100
View File
@@ -0,0 +1,100 @@
import networkx as nx
from shapely.geometry import Point
import matplotlib.pyplot as plt
import unittest
from linesweep_circle import NodeIDs
def graphbuild(d_racks_segments, d_rack_conn_points):
# 2. Graph erstellen
G = nx.Graph()
# 3. Knoten und Kanten aus d_racts_segments hinzufügen
for rack, points in d_racks_segments.items():
start = points[0]
end = points[1]
G.add_node(start) # Knoten für Startpunkt
G.add_node(end) # Knoten für Endpunkt
G.add_edge(start, end) # Kante zwischen Start und Endpunkt
# 4. Verbindungen aus d_rack_conn_points hinzufügen
for connection, points in d_rack_conn_points.items():
for point in points:
# Wir fügen die Kante zwischen den Punkten der Verbindung hinzu
# Hinweis: Wir nehmen hier an, dass der Punkt von beiden Racks als Verbindungspunkt betrachtet wird
G.add_edge(point, point) # Verbindungspunkt als Kante
return G
class TestShapely(unittest.TestCase):
def test_graph(self):
d_racks_segments = {
'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)]
}
d_rack_conn_points = {
'Rack_2-0 + Rack_2-1': [Point(0.1, 3777.6)],
'Rack_2-1 + Rack_1-0': [Point(4946.5, 3777.6)],
'Rack_2-1 + Rack_4-0': [Point(2866.6, 3777.6)],
'Rack_2-1 + Rack_5-0': [Point(8866.1, 3777.6)],
'Rack_3-0 + Rack_1-0': [Point(4946.5, 15865.5)],
'Rack_3-0 + Rack_4-0': [Point(2866.6, 15865.5)],
'Rack_3-0 + Rack_5-0': [Point(8866.1, 15865.5)]
}
G = graphbuild(d_racks_segments, d_rack_conn_points)
#print(nx.has_path(G, Point(4946.5, 15865.5), Point(0.1, 57.6)))
# 5. Graph anzeigen
nodes_str = "[<POINT (4946.5 15865.5)>, <POINT (4946.5 3777.6)>, <POINT (0.1 57.6)>, <POINT (0.1 3777.6)>," \
" <POINT (14755.1 3777.6)>, <POINT (185.1 15865.5)>, <POINT (12450.7 15865.5)>, <POINT (2866.6 15865.5)>," \
" <POINT (2866.6 3777.6)>, <POINT (8866.1 15865.5)>, <POINT (8866.1 3777.6)>]"
edges_str = "[(<POINT (4946.5 15865.5)>, <POINT (4946.5 3777.6)>), (<POINT (4946.5 15865.5)>," \
" <POINT (4946.5 15865.5)>), (<POINT (4946.5 3777.6)>, <POINT (4946.5 3777.6)>)," \
" (<POINT (0.1 57.6)>, <POINT (0.1 3777.6)>), (<POINT (0.1 3777.6)>, <POINT (14755.1 3777.6)>)," \
" (<POINT (0.1 3777.6)>, <POINT (0.1 3777.6)>), (<POINT (185.1 15865.5)>, <POINT (12450.7 15865.5)>)," \
" (<POINT (2866.6 15865.5)>, <POINT (2866.6 3777.6)>), (<POINT (2866.6 15865.5)>, <POINT (2866.6 15865.5)>)," \
" (<POINT (2866.6 3777.6)>, <POINT (2866.6 3777.6)>), (<POINT (8866.1 15865.5)>, <POINT (8866.1 3777.6)>)," \
" (<POINT (8866.1 15865.5)>, <POINT (8866.1 15865.5)>), (<POINT (8866.1 3777.6)>, <POINT (8866.1 3777.6)>)]"
nodes = G.nodes
self.assertEqual(str(G.nodes), nodes_str)
self.assertEqual(str(G.edges), edges_str)
allids = NodeIDs(nodes)
ids = allids.get_ids(nodes)
self.assertEqual(len(ids), len(nodes))
# 6. Optional: Visualisierung des Graphen
# Dies ist ein sehr grundlegendes Plotten. Bei Bedarf könnte man es weiter anpassen.
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()
if __name__ == '__main__':
unittest.main()
lib/linesweep_circle.py
+200 -89
View File
@@ -1,115 +1,226 @@
import json import json
from shapely.geometry import LineString, Point from shapely.geometry import LineString, Point
from shapely.ops import nearest_points from shapely.ops import nearest_points
import unittest
# === Konfiguration === class NodeIDs():
tol = 200 def __init__(self, points=[]):
tol_step = 10 self._counter = 0
self._cord2id = dict()
self._id2cord = dict()
self.add_points(points)
# === Lade JSON-Daten === def add_point(self, point:Point):
with open("C:/10-Develop/kabellaengen/work/easy_positions.json", "r") as f: self._counter += 1
data = json.load(f) self._cord2id[f"{point.x} {point.y}"] = self._counter
self._id2cord[f"{self._counter}"] = point
racks_json = data["racks"] #Suchen nach Racks in gesamter Json-Übergabe def add_points(self, points):
for p in points:
self.add_point(p)
# === 1. Racks in Segmente zerlegen === def get_id(self, point:Point) -> int:
''' Hier werden Racks, die aus "echter" Polylinie bestehen (mehrere Nodes, z.B. Rack 2 in easy.dxf) in einzelne Segmente zerlegt (Node1 -> Node2, Node2 -> Node3)''' return self._cord2id[f"{point.x} {point.y}"]
rack_segments = []
for rack_id, nodes in racks_json.items(): def get_point(self, nid:int) -> Point:
# Sortiere Node_1, Node_2, ... return self._id2cord[f"{nid}"]
sorted_keys = sorted(nodes.keys(), key=lambda k: int(k.split("_")[1]))
coords = [tuple(nodes[k]) for k in sorted_keys] 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 get_points(self, nids:list[int]) -> list[Point]:
ret = list()
for n in nids:
c = self.get_point(n)
ret.append(c)
return ret
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))
def to_json(d, pretty: bool = True) -> str:
return json.dumps(d, indent=2 if pretty else None, default=str) #ensure_ascii false für darstellung von "ue"
# === 2. Alle Endpunkte sammeln === def rack_segmentation(racks):
''' Alle Endpunkte aller Racks als Point gespeichert, um shapely funktionen verwenden zu können''' ''' Racks werden zu LineString konvertiert. Racks bestehend aus Polylinine werden in einzelne Segmente zerlegt.'''
segment_endpoints = [] rack_segments = []
for rack_id, idx, line in rack_segments: for rack_id, nodes in racks.items():
for pt in [line.coords[0], line.coords[1]]: # Sortiere Node_1, Node_2, ...
segment_endpoints.append((rack_id, idx, Point(pt))) 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 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)))
return(segment_endpoints)
def increase_circle(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
# === 3. Verbindungen suchen === # === 3. Verbindungen suchen ===
verbindungen = [] def search_connections(rack_segments, segment_endpoints, tol, tol_step):
endpoint_pinned = [] verbindungen = []
endpoint_pinned = []
# === A: Echte Schnittpunkte zwischen Linien finden === # === A: Echte Schnittpunkte zwischen Linien finden ===
''' Alle Segmente mit allen überprüfen, um echte SP zu finden''' ''' Alle Segmente mit allen überprüfen, um echte SP zu finden'''
for i, (rack_id1, idx1, line1) in enumerate(rack_segments): for i, (rack_id1, idx1, line1) in enumerate(rack_segments):
print(f"\n=== Prüfe {rack_id1}_{idx1} auf echte Schnittpunkte") #print(f"\n=== Prüfe {rack_id1}_{idx1} auf echte Schnittpunkte")
for j, (rack_id2, idx2, line2) in enumerate(rack_segments): for j, (rack_id2, idx2, line2) in enumerate(rack_segments):
if i >= j: if i >= j:
continue # keine Duplikate / sich selbst continue # keine Duplikate / sich selbst
if line1.intersects(line2): if line1.intersects(line2):
inter = line1.intersection(line2) inter = line1.intersection(line2)
if inter.geom_type == "Point": if inter.geom_type == "Point":
print(f"✅ Exakter Schnittpunkt {inter} zwischen {rack_id1}_{idx1} und {rack_id2}_{idx2}") #print(f"✅ Exakter Schnittpunkt {inter} zwischen {rack_id1}_{idx1} und {rack_id2}_{idx2}")
verbindungen.append((rack_id1, idx1, rack_id2, idx2, inter)) verbindungen.append((rack_id1, idx1, rack_id2, idx2, inter))
# === B: Näherungsweise Verbindung durch Toleranz-Kreise === # === B: Näherungsweise Verbindung durch Toleranz-Kreise ===
''' Entlanglaufen der Racks und Scan nach Endpunkten im Toleranzbereich''' ''' Entlanglaufen der Racks und Scan nach Endpunkten im Toleranzbereich'''
for rack_id, idx, line in rack_segments: for rack_id, idx, line in rack_segments:
print(f"\n=== Prüfe {rack_id}_{idx1} auf Punkte im Toleranzbereich") #print(f"\n=== Prüfe {rack_id}_{idx1} auf Punkte im Toleranzbereich")
for other_rack_id, other_idx, pt in segment_endpoints: for other_rack_id, other_idx, pt in segment_endpoints:
if rack_id == other_rack_id: if rack_id == other_rack_id:
continue # ignoriere eigene Endpunkte continue # ignoriere eigene Endpunkte
# Exakte Schnittpunkte ignorieren # Exakte Schnittpunkte ignorieren
if line.intersects(pt): if line.intersects(pt):
continue continue
dist = line.distance(pt) dist = line.distance(pt)
if dist < tol: if dist < tol:
print(f"🔍 Punkt {pt} liegt {dist:.2f} von Linie {rack_id}_{idx} entfernt") increase_circle(tol, tol_step, line, pt, rack_id, idx, other_rack_id, other_idx, verbindungen, endpoint_pinned)
#print(f"🔍 Punkt {pt} liegt {dist:.2f} von Linie {rack_id}_{idx} entfernt")
radius = tol_step # radius = tol_step
while radius <= tol: # while radius <= tol:
circle = pt.buffer(radius) # circle = pt.buffer(radius)
if circle.intersects(line): # if circle.intersects(line):
contact = circle.intersection(line) # contact = circle.intersection(line)
if contact.geom_type == "Point": # if contact.geom_type == "Point":
nearest = contact # nearest = contact
else: # else:
nearest = nearest_points(pt, contact)[1] # nearest = nearest_points(pt, contact)[1]
print(f" 🟡 Kreisberührung bei {nearest} mit {rack_id}_{idx}") # #print(f" 🟡 Kreisberührung bei {nearest} mit {rack_id}_{idx}")
verbindungen.append((rack_id, idx, other_rack_id, other_idx, nearest)) # 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] # # 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)) # endpoint_pinned.append((other_rack_id, other_idx, pt, nearest, rack_id))
break # break
radius += tol_step # radius += tol_step
# === Endpunkte aktualisieren === # === Endpunkte aktualisieren ===
# Dict erstellen, dass mit dem Key "Rack_id - index" dahinter die Koordinaten von Anfang und Endpunkt speichert # Dict erstellen, dass mit dem Key "Rack_id - index" dahinter die Koordinaten von Anfang und Endpunkt speichert
d_racks_segments = dict() d_racks_segments = dict()
for rack_id, idx, linestring in rack_segments: for rack_id, idx, linestring in rack_segments:
key = f"{rack_id}-{idx}" key = f"{rack_id}-{idx}"
d_racks_segments[key] = [Point(linestring.coords[0]), Point(linestring.coords[1])] #Alle Racks in ihrer eingelesenen Form zum Dict hinzufügen d_racks_segments[key] = [Point(linestring.coords[0]), Point(linestring.coords[1])] #Alle Racks in ihrer eingelesenen Form zum Dict hinzufügen
for rack_id, idx, old_pt, new_pt, taget_rack in endpoint_pinned: #Durch verschobene Endpunkte laufen... for rack_id, idx, old_pt, new_pt, taget_rack in endpoint_pinned: #Durch verschobene Endpunkte laufen...
key = f"{rack_id}-{idx}" key = f"{rack_id}-{idx}"
coords = d_racks_segments.get(key) coords = d_racks_segments.get(key)
if coords: #...und bei Übereinstimmung von Start oder Endkoordinate die ursprüngliche (eingelesene) mit der gepinnten überschreiben if coords: #...und bei Übereinstimmung von Start oder Endkoordinate die ursprüngliche (eingelesene) mit der gepinnten überschreiben
# Vergleich mit Startpunkt # Vergleich mit Startpunkt
if Point(coords[0]).equals(old_pt): if Point(coords[0]).equals(old_pt):
coords[0] = Point(new_pt.x, new_pt.y) #.x bzw .y übergibt x bzw y Koordinate von Objekt POINT coords[0] = Point(new_pt.x, new_pt.y) #.x bzw .y übergibt x bzw y Koordinate von Objekt POINT
# Vergleich mit Endpunkt # Vergleich mit Endpunkt
elif Point(coords[1]).equals(old_pt): elif Point(coords[1]).equals(old_pt):
coords[1] = Point(new_pt.x, new_pt.y) coords[1] = Point(new_pt.x, new_pt.y)
d_racks_segments[key] = coords # aktualisieren d_racks_segments[key] = coords # aktualisieren
# === Ausgabe-Verbindungen === #Dict erstellen, dass alle Punkte die an einem Rack anschließen
print("\n=== Gefundene Verbindungen ===") d_rack_conn_points = dict()
for v in verbindungen:
print(v) for conn_to_rack, conn_to_idx, conn_from_rack, conn_from_idx, conn_point in verbindungen:
key = f"{conn_to_rack}-{conn_to_idx} + {conn_from_rack}-{conn_from_idx}"
d_rack_conn_points[key] = [conn_point]
return [d_racks_segments, d_rack_conn_points]
class TestLinesweep(unittest.TestCase):
def setUp(self):
# === Lade JSON-Daten ===
with open("C:/10-Develop/kabellaengen/work/easy_positions.json", "r") as f:
self.data = json.load(f)
def test_linesweep(self):
# === Konfiguration ===
tol = 200
tol_step = 10
racks_json = self.data["racks"] #Suchen nach Racks in gesamter Json-Übergabe
# === 1. Racks in Segmente zerlegen ===
''' Hier werden Racks, die aus "echter" Polylinie bestehen (mehrere Nodes, z.B. Rack 2 in easy.dxf) in einzelne Segmente zerlegt (Node1 -> Node2, Node2 -> Node3)'''
rack_segments = rack_segmentation(racks_json)
# === 2. Alle Endpunkte sammeln ===
''' Alle Endpunkte aller Racks als Point gespeichert, um shapely funktionen verwenden zu können'''
segment_endpoints = rack_endpoints(rack_segments)
d_racks_segments, d_rack_conn_points = search_connections(rack_segments, segment_endpoints, tol, tol_step)
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)]
}
log_res = to_json(res_rack_seg)
self.assertEqual(d_racks_segments, res_rack_seg)
if __name__ == '__main__':
unittest.main()