import os import json import argparse import heapq import math import matplotlib.pyplot as plt import networkx as nx # Hilfsfunktionen def load_json(filepath): with open(filepath, 'r', encoding='utf-8') as f: return json.load(f) def parse_pos(pos_str): """ Konvertiert '(x, y)' oder '(x, y, z)' in ein Tuple """ try: return tuple(map(float, pos_str.strip('[]').split(','))) except Exception: raise ValueError(f"Ungültiges Positionsformat: {pos_str}") def distance(p1, p2): """ Euklidische Distanz in 2D """ return math.sqrt((p1[0] - p2[0])**2 + (p1[1] - p2[1])**2) def add_edge(graph, node1, node2, dist): """ Fügt eine ungerichtete Kante zwischen zwei Knoten hinzu, aber nur einmal """ if node1 not in graph: graph[node1] = [] if node2 not in graph: graph[node2] = [] # Nur hinzufügen, wenn Kante noch nicht existiert (ungerichtet) if not any(n == node2 for n, _ in graph[node1]): graph[node1].append((node2, dist)) if not any(n == node1 for n, _ in graph[node2]): graph[node2].append((node1, dist)) def project_point_on_segment(p, a, b): """ Projektion eines Punktes p auf ein Liniensegment a-b """ ax, ay = a bx, by = b px, py = p dx = bx - ax dy = by - ay if dx == dy == 0: return a t = ((px - ax) * dx + (py - ay) * dy) / (dx * dx + dy * dy) t = max(0, min(1, t)) # Begrenze t auf [0,1] return (ax + t * dx, ay + t * dy) def dijkstra(graph, start): """ Dijkstra-Algorithmus, um die kürzesten Wege im Graphen zu berechnen """ distances = {node: float('inf') for node in graph} distances[start] = 0 priority_queue = [(0, start)] # (Distanz, Knoten) while priority_queue: current_distance, current_node = heapq.heappop(priority_queue) if current_distance > distances[current_node]: continue for neighbor, weight in graph[current_node]: distance = current_distance + weight if distance < distances[neighbor]: distances[neighbor] = distance heapq.heappush(priority_queue, (distance, neighbor)) return distances def print_graph(graph): printed = set() for node, edges in graph.items(): for neighbor, dist in edges: edge_id = tuple(sorted((node, neighbor))) if edge_id not in printed: printed.add(edge_id) print(f"{edge_id[0]} --> {edge_id[1]} (Distanz: {dist})") def visualize_graph(graph, racks): G = nx.Graph() pos = {} for node, edges in graph.items(): pos[node] = node_to_coords(node, racks) for neighbor, distance in edges: if not G.has_edge(node, neighbor): # Doppelte Kanten vermeiden G.add_edge(node, neighbor, weight=round(distance, 1)) plt.figure(figsize=(10, 10)) nx.draw( G, pos, with_labels=True, node_size=100, font_size=8, node_color='skyblue', edge_color='gray' ) edge_labels = nx.get_edge_attributes(G, 'weight') nx.draw_networkx_edge_labels(G, pos, edge_labels=edge_labels, font_size=6) plt.title("Rack-Graph (aus racks.json)") plt.axis("equal") plt.tight_layout() plt.show() def node_to_coords(node_name, racks): """Extrahiert die Koordinaten aus dem Knotennamen wie 'Rack_1_Node_2'""" parts = node_name.split("_") rack = f"{parts[0]}_{parts[1]}" node = f"{parts[2]}_{parts[3]}" coords = racks[rack][node] return tuple(coords) if __name__ == "__main__": parser = argparse.ArgumentParser(description='Berechne Wege von Sensoren zu Verteilern über Kabeltrassen') parser.add_argument('-c', '--console', action='store_true', help='Ausgabe auf Konsole') args = parser.parse_args() # Umgebungsvariablen work_dir = os.environ.get("PROJECT_WORK") config_dir = os.environ.get("PROJECT_CFG") # Pfade zu JSON-Dateien sensors_path = os.path.join(work_dir, "sensors.json") subdist_path = os.path.join(work_dir, "subdistributors.json") racks_path = os.path.join(work_dir, "racks.json") # Einlesen sensors = load_json(sensors_path) subdists = load_json(subdist_path) racks = load_json(racks_path) # Graph erstellen graph = {} for rack_id, rack in racks.items(): nodes = list(rack.values()) # Liste aller Knoten im Rack for i in range(len(nodes) - 1): segment_start = tuple(nodes[i]) segment_end = tuple(nodes[i + 1]) dist = distance(segment_start, segment_end) # Erstelle Kanten zwischen den benachbarten Knoten add_edge(graph, f"{rack_id}_Node_{i+1}", f"{rack_id}_Node_{i+2}", dist) # Graph in Kommandozeile beschreiben und mittels matplotlib ausgeben print("\nGraph basierend auf den Racks (ungerichtet, eindeutige Kanten):") print_graph(graph) visualize_graph(graph, racks) """# 1. Vom Sensor zum Rack laufen und Knoten einfügen for sensor_id, sensor_info in sensors.items(): sensor_pos = tuple(sensor_info['pos']) for rack in racks: for segment_start, segment_end in zip(rack[:-1], rack[1:]): # Berechne Distanz von Sensor zur Kabeltrasse px, py = project_point_on_segment(sensor_pos, segment_start, segment_end) dist = distance(sensor_pos, (px, py)) rack_id = f"rack_{rack}" # Sensor zum Rack Knoten verbinden add_edge(graph, sensor_id, rack_id, dist) # 2. Vom Unterverteiler (UV) zum Rack laufen und Knoten einfügen for uc_id, uc_pos in subdists.items(): for rack in racks: for segment_start, segment_end in zip(rack[:-1], rack[1:]): # Berechne Distanz von UV zur Kabeltrasse px, py = project_point_on_segment(uc_pos, segment_start, segment_end) dist = distance(uc_pos, (px, py)) rack_id = f"rack_{rack}" # UV zum Rack Knoten verbinden add_edge(graph, uc_id, rack_id, dist) # 3. Vom Sensor Knoten zum zugehörigen Unterverteiler Knoten entlang der Racks for sensor_id, sensor_info in sensors.items(): subdist_id = None if 'KENNZEICHNUNG' in sensor_info: for uc_id in subdists: if uc_id in sensor_info['KENNZEICHNUNG']: subdist_id = uc_id break if subdist_id: # Verbinde den Sensor mit dem zugehörigen Unterverteiler sensor_pos = tuple(sensor_info['pos']) uc_pos = subdists[subdist_id] # Berechne Distanz von Sensor zum Unterverteiler (über Trassen) dist = distance(sensor_pos, uc_pos) add_edge(graph, sensor_id, subdist_id, dist) # 4. Berechnung der kürzesten Wege mit Dijkstra routing_result = {} for sensor_id in sensors: distances = dijkstra(graph, sensor_id) routing_result[sensor_id] = distances if args.console: print(json.dumps(routing_result, indent=2))"""