kabellaengen/lib/routing.py

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"
    # ]
    # }

    tol_snap = config.getfloat("Racks", "SnapTolerances")
    snap_step = config.getfloat("Racks", "SnapStep")
    tol_connect = config.getfloat("Sensoren", "ConnectionTolerances")
    tol_connect_step = config.getfloat("Sensoren", "ConnectSnap")

    an = Anlage(tol_snap=tol_snap, snap_step=snap_step, tol_connect=tol_connect, tol_connect_step=tol_connect_step)
    # 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
    an.add_sensors(sensors)
    # 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
    an.add_tunnels(tunnels)
    an.set_tunnel_length(tunlength)
    # Verbinde Tunnel mit deren naheliegendsten Racks
    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:
        draw_graph(G,an)

    paths['errors_sensors'] = errors_sensors
    paths['errors_dists'] = errors_dists
    paths['errors_tunnels'] = errors_tunnels
    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(sdata["pos"])

    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():
    #     if rname not in dracks:
    #         dracks[rname] = list()
        # dracks[rname].append(Point(lp))
        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"]
    
    
    return (dracks, dsensors, dsubdists, mapping, dtunnels, dtunlength)


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description='Berechne Wege von Sensoren zu Verteilern über Kabeltrassen')
    parser.add_argument('-f', '--filename', action='store', required=True, default="easy_position.json", help='file with all informations about positions gathered from getpositions', metavar='my_positions.json')
    parser.add_argument('-c', '--console', action='store_true', help='Ausgabe auf Konsole')
    parser.add_argument('-g', '--graph', action='store_true', help='Zeichnet den Graphen der Anlage')
    parser.add_argument('-w', '--write', action='store', help='erstellt Ausgabe-file für das Zeichnen von Kabeln in drawdxf')

    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) = 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, "routing.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']))    

    # Ausgabe schreiben
    if args.write:
        basename = os.path.splitext(args.write)[0]
        write_results(to_json(cable_paths), work_dir, f"{basename}.json")