import os import json import argparse import heapq import math 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 import configparser # Funktionen def load_json(jsonfilename): with open(jsonfilename, encoding='utf-8') as fh: return json.load(fh) def write_results(jsnResults, outdir, filename): """ write results to a json file """ print("writing results to a json file ...") outfile = os.path.join(outdir, filename) with open(outfile, 'w', encoding='utf-8') as fh: fh.write(jsnResults) print("done") def to_json(d, pretty: bool = True) -> str: return json.dumps(d, indent=2 if pretty else None, ensure_ascii=False, default=str) #ensure_ascii false für darstellung von "ue" 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","") } 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: ltemp.append(Point(p)) 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 = rawdata["tunnels"]["length"] # Fehler, welche im getpositions auftreten weiterführen: im Layout fehlende Dists / Sensoren / fehlende Attribute errors_dists = rawdata["not_found"]["missing_distributors"] errors_sensors = rawdata["not_found"]["missing_sensors"] errors_attributes = rawdata["not_found"]["missing_attributes"] return (dracks, dsensors, dsubdists, mapping, dtunnels, dtunlength, errors_dists, errors_sensors, errors_attributes) 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 = os.environ.get("PROJECT_WORK") config_dir = os.environ.get("PROJECT_CFG") # Pfade zu JSON-Dateien jsonfilename = args.filename sensors_path = os.path.join(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) = prepare_data(rawdata) config = configparser.ConfigParser(allow_no_value=True, delimiters=("=")) config.optionxform = lambda option: option # preserve case for letters config.read(os.path.join(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 = os.path.splitext(args.write)[0] 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 write_results(to_json(cable_paths), work_dir, f"{basename}.json")