Einlesen von positions.json und verarbeitung von kabel.cfg sodass kabel nach länge gefiltert in excel ausgabe erscheinen
This commit is contained in:
+137
-15
@@ -48,6 +48,11 @@ class Polylines:
|
||||
errors_existing_dists: List[str]
|
||||
errors_existing_sensors: List[str]
|
||||
|
||||
@dataclass
|
||||
class Sensors:
|
||||
name: str
|
||||
artinr: str
|
||||
pos: List[float]
|
||||
|
||||
def add_polyline(msp, points:Polyline, dxf_attribs):
|
||||
pts = points.to_tuple()
|
||||
@@ -238,7 +243,6 @@ def draw_subdists(plines, doc):
|
||||
text.dxf.valign = valign
|
||||
text.set_placement((pt1.x + offsetx, pt1.y + offsety))
|
||||
|
||||
|
||||
def model_from_json(json_file):
|
||||
with open(json_file, encoding='utf-8') as fh:
|
||||
data = json.load(fh)
|
||||
@@ -249,36 +253,72 @@ def model_from_json(json_file):
|
||||
)
|
||||
return plines
|
||||
|
||||
def parse_sensors_from_json(positions_json):
|
||||
with open(positions_json, encoding='utf-8') as fh:
|
||||
data = json.load(fh)
|
||||
|
||||
sensors = {}
|
||||
for name, data in data.get("sensors", {}).items():
|
||||
sensor = Sensors(
|
||||
name=name,
|
||||
artinr=data.get("ARTINR", ""),
|
||||
pos=data.get("pos", [0.0, 0.0]),
|
||||
)
|
||||
sensors[name] = sensor
|
||||
return sensors
|
||||
|
||||
def export_excel(json_file, out_path):
|
||||
# Hier für Excel Export
|
||||
print("creating excel file with cable information ..")
|
||||
plines = model_from_json(json_file)
|
||||
write_excel_from_json(plines, out_path)
|
||||
sens2cable = map_sensor_to_cable(sensors, plines)
|
||||
write_excel_from_json(plines, sens2cable, out_path)
|
||||
print("done")
|
||||
|
||||
def write_excel_from_json(plines:Polylines, outpath:str):
|
||||
def write_excel_from_json(plines:Polylines, sens2cable: dict, outpath:str):
|
||||
wb = Workbook()
|
||||
length_summary = defaultdict(int)
|
||||
|
||||
#Worksheet 1 - Kabellängen
|
||||
# Dicts für Anzahl bzw kummulierte Länge
|
||||
count_summary = defaultdict(int)
|
||||
length_summary = defaultdict(float)
|
||||
|
||||
#Worksheet 1 - Kabellängen nach Kabel-ID
|
||||
ws1 = wb.active
|
||||
ws1.title = "Length by ID"
|
||||
ws1.append(["Cable-ID", "True Length (m)", "Rounded Length (m)"])
|
||||
ws1.append(["Cable-ID", "True Length (m)", "Cable-ArtNr"])
|
||||
|
||||
for pl in plines.kabel:
|
||||
length = pl.length /1000 # Umrechnung von mm in m
|
||||
rounded_len = math.ceil(length/5)*5
|
||||
length_summary[rounded_len] +=1
|
||||
ws1.append([pl.id, length, int(rounded_len)])
|
||||
artnr = sens2cable[pl.id]
|
||||
ws1.append([pl.id, length, artnr])
|
||||
|
||||
if "MA" in pl.id:
|
||||
length_summary[artnr] += math.ceil(length) # Aufrunden von z.b. 10,3 auf 11 m (volle Meter!)
|
||||
|
||||
|
||||
# MB als Motor gesetzt! Beachte auch Zuordnung in get_cable_artnr!!
|
||||
|
||||
elif "MB" in pl.id: # MB statt MA zur Motor-Kennzeichnung
|
||||
length_summary[artnr] += math.ceil(length) # Aufrunden von z.b. 10,3 auf 11 m (volle Meter!)
|
||||
|
||||
else:
|
||||
count_summary[artnr] += 1
|
||||
|
||||
# Worksheet 2 - Zusammengefasste Längen
|
||||
ws2 = wb.create_sheet("Length Summary")
|
||||
ws2.append(["Rounded Length (m)", "Number of Cables"])
|
||||
# Worksheet 2 - Kabelnummern und Stückzahlen
|
||||
ws2 = wb.create_sheet("Cables SIVAS")
|
||||
ws2.append(["Cable-ArtNr", "Ammount / total length"])
|
||||
|
||||
for rlength in sorted(length_summary):
|
||||
ws2.append([int(rlength), length_summary[rlength]])
|
||||
all_artnrs = set(count_summary.keys()) | set(length_summary.keys())
|
||||
|
||||
for artnr in sorted(all_artnrs):
|
||||
count = count_summary.get(artnr, "")
|
||||
if count == "":
|
||||
count = length_summary.get(artnr, "")
|
||||
ws2.append([artnr, count])
|
||||
|
||||
# Abfage ob Fehler Worsheets ausgegeben werden
|
||||
if len(plines.errors_existing_sensors) > 0 or len(plines.errors_dists) > 0:
|
||||
|
||||
# Worksheet 3 - Nicht an Racks gekoppeltes Equipment
|
||||
ws3 = wb.create_sheet("Not connected Equipment")
|
||||
ws3.append(["Type", "ID", "x", "y"])
|
||||
@@ -388,10 +428,78 @@ def copy_layers_into_dxf_by_filter(dxf_source: ezdxf.document.Drawing, dxf_targe
|
||||
for entity in entities:
|
||||
msp_target.add_entity(entity.copy())
|
||||
|
||||
def get_cable_artnr(section, length):
|
||||
"""
|
||||
Sucht in der angegebenen Config-Section die passende Kabel-Artikelnr. für die gegebene Länge.
|
||||
"""
|
||||
if section not in cable_cfg:
|
||||
return None
|
||||
|
||||
entries = cable_cfg[section]
|
||||
|
||||
# Alle Einträge der Form: "5" = "123456789"
|
||||
try:
|
||||
length_keys = sorted(
|
||||
[float(k) for k in entries.keys() if k.replace('.', '', 1).isdigit()]
|
||||
)
|
||||
except ValueError:
|
||||
return None
|
||||
|
||||
for l in length_keys:
|
||||
if length <= l:
|
||||
return entries[str(int(l))] # Annahme: Keys sind ganze Zahlen in Metern
|
||||
|
||||
return None
|
||||
|
||||
|
||||
def map_sensor_to_cable(sensors, plines):
|
||||
sens2cable = defaultdict()
|
||||
for pl in plines.kabel:
|
||||
# Sensorname von Kabel-ID extraieren
|
||||
sensor_name = pl.id.split('-')[-1]
|
||||
# Kabellänge aus gleichem Eintrag auslesen + in m umrechnen
|
||||
cable_length = pl.length/1000
|
||||
|
||||
# Sensor Artikelnummer aus positions.json holen
|
||||
sensor_artinr = sensors[sensor_name].artinr
|
||||
|
||||
# Verzweigung, um welchen Typ es sich handelt (Special Keys)
|
||||
name_prefix = sensor_name[0:2]
|
||||
|
||||
section = None #Default
|
||||
|
||||
if name_prefix == "MA" or name_prefix == "MB":
|
||||
section = "MA"
|
||||
|
||||
elif name_prefix == "QM":
|
||||
section = "WD_Q"
|
||||
|
||||
elif sensor_name.startswith("B"):
|
||||
if sensor_artinr == 829422026:
|
||||
section = "WD_I-829422026"
|
||||
elif sensor_artinr == 720002003:
|
||||
section = "WD_I-720002003"
|
||||
else:
|
||||
section = "WD_I"
|
||||
|
||||
if section is None:
|
||||
cable_artnr = "KEIN KABEL"
|
||||
else:
|
||||
cable_artnr = get_cable_artnr(section, cable_length)
|
||||
if cable_artnr is None:
|
||||
cable_artnr = "KEIN KABEL"
|
||||
|
||||
sens2cable[pl.id] = cable_artnr
|
||||
|
||||
return sens2cable
|
||||
|
||||
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
parser = argparse.ArgumentParser(description='draws a dxf file with the given cable coordinates', prog='drawdxf')
|
||||
parser.add_argument('-f', '--filename', action='store', required=True, help='this json file contains all cables and its coordinates which should be drawn. Saved with an unique timestamp', metavar='myfile.json')
|
||||
parser.add_argument('-p','--positions', action='store', required=True, help='this json file contains positional information written by getpositions', metavar='myfile.json')
|
||||
parser.add_argument('-d', '--dxf', action='store', help='this dxf drawing will be copied and the new layer with the cables will be added. Original file must be added with --origin', metavar='myfile.dxf')
|
||||
parser.add_argument('-c', '--copy_layer', action='store', help='copy layers of racks, sensors, distributors into a new .dxf-file. File also contains cable paths. Original file must be added with --origin', metavar='original.dxf')
|
||||
parser.add_argument('-n', '--new', action='store', help='create a new dxf file only with cables in it. Name is basename and a timestamp')
|
||||
@@ -403,21 +511,35 @@ if __name__ == '__main__':
|
||||
|
||||
config_dir = os.environ.get("PROJECT_CFG")
|
||||
work_dir = os.fspath(os.environ.get('PROJECT_WORK'))
|
||||
|
||||
json_file = args.filename
|
||||
(json_path, jexists) = check_file_in_work(work_dir, json_file)
|
||||
if not jexists:
|
||||
print(f"file {json_file} does not exist")
|
||||
parser.print_help()
|
||||
exit()
|
||||
|
||||
positions_json = args.positions
|
||||
(positions_path, jexists) = check_file_in_work(work_dir, positions_json)
|
||||
if not jexists:
|
||||
print(f"file {positions_json} does not exist")
|
||||
parser.print_help()
|
||||
exit()
|
||||
|
||||
plines = model_from_json(json_path)
|
||||
sensors = parse_sensors_from_json(positions_path)
|
||||
|
||||
|
||||
# Allgemeine Config laden
|
||||
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"))
|
||||
|
||||
# Config für Kabel-Atikelnummern laden
|
||||
cable_cfg = configparser.ConfigParser()
|
||||
cable_cfg.optionxform = str #Keys case-sensitive
|
||||
cable_cfg.read(os.path.join(config_dir, "kabel.cfg"))
|
||||
|
||||
dxf_file = args.dxf
|
||||
|
||||
if args.dxf or args.copy_layer:
|
||||
|
||||
Reference in New Issue
Block a user