import json import networkx as nx from shapely.geometry import Point from plant import Anlage from pathlib import Path import argparse import os import configparser import matplotlib.pyplot as plt from utils import load_json, to_json, write_results # Funktionen def create_plant(racks:dict, sensors:dict, distributors:dict, mapping:dict, tunnels: dict, tunlength:dict ) -> 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)]} # 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']} # "mapping": { # "UC0101": [ # "BG3241", # "BG3240", # "MA0062", # "FC0062" # ] # } # Einlesen der Toleranzen zur Verbindung von Rack zueinander und Peripherie zu Racks aus Config tol_snap = config.getfloat("Racks", "SnapTolerances") tol_connect = config.getfloat("Sensoren", "ConnectionTolerances") G = nx.Graph() an = Anlage(tol_snap=tol_snap, tol_connect=tol_connect) # 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 und speichere Sensoren mit deren Artikelnummern an.add_sensors({sname: sdata["point"] for sname, sdata in sensors.items()}) # nur Punkte zu jeweiligem sname übergeben an.set_sensor_artnrs({sname: sdata["artinr"] for sname, sdata in sensors.items()}) # nur Artikelnummern zu jeweiligem sname übergeben # Verbinde Sensoren mit deren naheliegendsten Racks errors_sensors = an.connect_sensors_to_racks() # Füge UV hinzu an.add_distributors(distributors) # Verbinde UV mit deren naheliegendsten Racks errors_dists = an.connect_distributor_to_racks() # Füge Tunnel hinzu und speichere Länge des Tunnels an.add_tunnels(tunnels) an.set_tunnel_length(tunlength) # Verbinde Tunnel mit deren naheliegendsten Racks und Tunnel zu sich selbst errors_tunnels = an.connect_tunnels() # 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: print("Displaying Graph in seperate window. To continue please close that window.") draw_graph(G,an) paths['errors_sensors'] = errors_sensors # Sensoren die nicht zu Racks verbunden werden konnten paths['errors_dists'] = errors_dists # Distributoren, die nicht zu Racks verbunden werden konnten paths['errors_tunnels'] = errors_tunnels # Tunnel, die nicht zu Racks verbunden werden konnten # paths['errors_routing'] # Knoten des Graphen, die nicht verbunden werden konnten (Ausgabe wird in create_cable_paths() erstellt) return paths 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=True, node_size=10, font_size=8, edge_color=edge_colors, node_color='none') nx.draw_networkx_nodes(G, pos, linewidths= 0.5, edgecolors = 'red', node_color = 'none') plt.show() def prepare_data(rawdata:dict): sensors = rawdata["sensors"] dsensors = dict() for sname, sdata in sensors.items(): dsensors[sname] = { "point": Point(sdata["pos"]), "artinr": sdata.get("ARTINR","").strip() } subdists = rawdata["distributors"] dsubdists = dict() for dname, pos in subdists.items(): dsubdists[dname] = Point(pos) racks = rawdata["racks"] dracks = dict() for rname,lp in racks.items(): ltemp = list() for p in lp: if len(p) == 3: pt = Point(p[0], p[1], p[2]) else: pt = Point(p[0], p[1], 0.0) ltemp.append(pt) dracks[rname] = ltemp mapping = rawdata["mappings"] tunnels = rawdata["tunnels"] dtunnels = dict() for tname,lp in tunnels.items(): if tname == "length": continue ltemp = list() for p in lp: ltemp.append(Point(p)) dtunnels[tname] = ltemp dtunlength = {} if "length" in rawdata["tunnels"]: dtunlength = rawdata["tunnels"]["length"] # Fehler, welche im getpositions auftreten weiterführen: im Layout fehlende Dists / Sensoren / fehlende Attribute errors_dists = list() errors_sensors = list() errors_attributes = dict() if "not_found" in rawdata: if "missing_distributors" in rawdata["not_found"]: errors_dists = rawdata["not_found"]["missing_distributors"] if "missing_sensors" in rawdata["not_found"]: errors_sensors = rawdata["not_found"]["missing_sensors"] if "missing_attributes" in rawdata["not_found"]: errors_attributes = rawdata["not_found"]["missing_attributes"] # Warnungen aus getpositions übernehmen warnings = dict() if "warnings" in rawdata: warnings = rawdata["warnings"] return (dracks, dsensors, dsubdists, mapping, dtunnels, dtunlength, errors_dists, errors_sensors, errors_attributes, warnings) if __name__ == "__main__": parser = argparse.ArgumentParser(description='Calculate cable-routing from Subdistributors zu sensors / actuators alon cable-racks') parser.add_argument('-f', '--filename', action='store', required=True, default="file_positions.json", help='file with all informations about positions gathered from getpositions', metavar='my_positions.json') parser.add_argument('-c', '--console', action='store_true', help='print to console') parser.add_argument('-g', '--graph', action='store_true', help='draw and show generated graph') parser.add_argument('-w', '--write', action='store', help='create .json file to pass into drawing module to visualize results') args = parser.parse_args() # Umgebungsvariablen work_dir_env = os.environ.get("PROJECT_WORK") config_dir_env = os.environ.get("PROJECT_CFG") if work_dir_env is None or config_dir_env is None: raise RuntimeError("PROJECT_WORK or PROJECT_CFG environment variable not set.") work_dir = Path(work_dir_env) config_dir = Path(config_dir_env) # Pfade zu JSON-Dateien jsonfilename = args.filename sensors_path = work_dir / jsonfilename # Einlesen und Vorbereiten der Daten rawdata = load_json(sensors_path) (racks, sensors, subdists, mapping, tunnels, tunlength, errors_dists, errors_sensors, errors_attributes, warnings) = prepare_data(rawdata) config = configparser.ConfigParser(allow_no_value=True, delimiters=("=")) config.optionxform = lambda optionstr: optionstr # preserve case for letters config.read(str(config_dir / "allgemein.cfg")) # virtuelle Anlage erstellen cable_paths = create_plant(racks, sensors, subdists, mapping, tunnels, tunlength) if args.console: print(to_json(cable_paths)) print("failed sensor:") print(to_json(cable_paths['errors_sensors'])) print("failed dists:") print(to_json(cable_paths['errors_dists'])) print("failed tunnels:") print(to_json(cable_paths['errors_tunnels'])) # Ausgabe schreiben if args.write: basename = Path(args.write).stem cable_paths["errors_dists_not_in_layout"] = errors_dists cable_paths["errors_sensors_not_in_layout"] = errors_sensors cable_paths["errors_missing_attributes"] = errors_attributes # Warnungen aus getpositions übertragen cable_paths["warnings"] = warnings write_results(to_json(cable_paths), work_dir, f"{basename}.json")