Methode für STR Tree mit Bounding Box hinzugefügt.

This commit is contained in:
2025-05-26 13:33:15 +02:00
parent b5c1a5f38c
commit e347794373
+69 -45
View File
@@ -1,5 +1,5 @@
import json import json
from shapely.geometry import LineString, Point from shapely.geometry import LineString, Point, box
from shapely.ops import nearest_points from shapely.ops import nearest_points
import unittest import unittest
from collections import defaultdict from collections import defaultdict
@@ -9,7 +9,7 @@ import matplotlib.pyplot as plt
from itertools import pairwise, combinations, permutations from itertools import pairwise, combinations, permutations
import re import re
from shapely.strtree import STRtree from shapely.strtree import STRtree
import shapely
class PointSorter: class PointSorter:
def __init__(self): def __init__(self):
@@ -442,6 +442,32 @@ class Anlage():
nearest_point = nearest_line.interpolate(nearest_line.project(sensor)) nearest_point = nearest_line.interpolate(nearest_line.project(sensor))
return(nearest_point, rack_name) return(nearest_point, rack_name)
def find_nearest_rack_from_point_STR_bbox(self, max_dist, sensor:Point) -> tuple[Point, str]:
if not hasattr(self, "_rack_tree"):
self._build_rack_strtree()
minx, miny, maxx, maxy = sensor.x - max_dist, sensor.y - max_dist, sensor.x + max_dist, sensor.y + max_dist
bbox = box(minx, miny, maxx, maxy)
candidates = self._rack_tree.query(box)
if not candidates:
return None, None
for line in candidates:
dist = sensor.distance(line)
if dist < best_dist:
best_dist = dist
best_line = line
if best_dist > max_dist:
return None, None
rack_name = self._rack_map[best_line]
nearest_point = best_line.interpolate(best_line.project(sensor))
return nearest_point, rack_name
def connect_equipment_to_racks(self, equipment: dict, onpoints: dict) -> list: def connect_equipment_to_racks(self, equipment: dict, onpoints: dict) -> list:
'''Verbindet Peripherie (Sensoren / Aktoren/ Unterverteiler) mit dem nächsten Rack. '''Verbindet Peripherie (Sensoren / Aktoren/ Unterverteiler) mit dem nächsten Rack.
Eingabe: Dict des Equipments (Sensoren o. Dists), Dict der Aufpunkte von Sensoren o. Dists Eingabe: Dict des Equipments (Sensoren o. Dists), Dict der Aufpunkte von Sensoren o. Dists
@@ -449,7 +475,7 @@ class Anlage():
''' '''
errors = [] errors = []
for name, pos in equipment.items(): for name, pos in equipment.items():
onpoint, rackname = self.find_nearest_rack_from_point_tree(self._tol_connect, pos) onpoint, rackname = self.find_nearest_rack_from_point_STR_bbox(self._tol_connect, pos)
if onpoint == None or rackname == None: if onpoint == None or rackname == None:
errors.append((name, pos)) errors.append((name, pos))
continue continue
@@ -496,9 +522,6 @@ class Anlage():
return errors return errors
def find_nearest_rack_from_point(self, max_dist, coarse_step, sensor:Point, racks:dict) -> tuple[Point, str]: def find_nearest_rack_from_point(self, max_dist, coarse_step, sensor:Point, racks:dict) -> tuple[Point, str]:
# 1. grobe Kandidatensuche # 1. grobe Kandidatensuche
@@ -840,7 +863,37 @@ class TestLinesweep(unittest.TestCase):
# self.assertEqual(plist2, [Point(0, 0), Point(0,1), Point(0, 10)]) # self.assertEqual(plist2, [Point(0, 0), Point(0,1), Point(0, 10)])
# def test_add_equipment_w_tree(self): def test_add_equipment_w_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)
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_add_equipment_w_tree_batch(self):
# racks = {'Rack_1': [Point(0, 0), Point(0, 10)], # racks = {'Rack_1': [Point(0, 0), Point(0, 10)],
# 'Rack_2': [Point(10, -2), Point(10, 5)], # 'Rack_2': [Point(10, -2), Point(10, 5)],
@@ -859,51 +912,22 @@ class TestLinesweep(unittest.TestCase):
# an.add_sensors(sensors) # an.add_sensors(sensors)
# an.add_distributors(distributors) # an.add_distributors(distributors)
# an.connect_equipment_to_racks(an._sensors, an._sensor_onpoints) # an.connect_equipment_batch(an._sensors, an._sensor_onpoints)
# an.connect_equipment_to_racks(an._distributors, an._distributors_onpoints) # an.connect_equipment_batch(an._distributors, an._distributors_onpoints)
# plist1 = an.get_points_from_rack("Rack_1") # plist1 = an.get_points_from_rack("Rack_1")
# plist2 = an.get_points_from_rack("Rack_2") # plist2 = an.get_points_from_rack("Rack_2")
# G1 = nx.Graph()
# pos = an.generate_graph(G1)
# nx.draw(G1, pos, with_labels=False, node_size=10, font_size=8)
# plt.show()
# self.assertEqual(plist1, [Point(0, 0), Point(0, 1), Point(0, 3), Point(0, 9), Point(0, 10)]) # 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)]) # self.assertEqual(plist2, [Point(10, -2), Point(10, 0), Point(10, 2), Point(10, 3), Point(10, 5)])
def test_add_equipment_w_tree_batch(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)
an.set_racks(racks)
an.join_racks()
an.add_sensors(sensors)
an.add_distributors(distributors)
an.connect_equipment_batch(an._sensors, an._sensor_onpoints)
an.connect_equipment_batch(an._distributors, an._distributors_onpoints)
plist1 = an.get_points_from_rack("Rack_1")
plist2 = an.get_points_from_rack("Rack_2")
G1 = nx.Graph()
pos = an.generate_graph(G1)
nx.draw(G1, pos, with_labels=False, node_size=10, font_size=8)
plt.show()
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_w_tree(self): # def test_wegsuche_w_tree(self):
# racks = {'Rack_1-0': [Point(0, 0), Point(0, 10)], # racks = {'Rack_1-0': [Point(0, 0), Point(0, 10)],