210 lines
7.8 KiB
Python
210 lines
7.8 KiB
Python
import os
|
|
import json
|
|
import argparse
|
|
import heapq
|
|
import math
|
|
import matplotlib.pyplot as plt
|
|
import networkx as nx
|
|
import shapely
|
|
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:
|
|
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 = 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")
|
|
|
|
|
|
|
|
|
|
|