222 lines
7.9 KiB
Python
222 lines
7.9 KiB
Python
import os
|
|
import sys
|
|
import argparse
|
|
from ioconverter import ExcelConverter
|
|
import json
|
|
from pathlib import Path
|
|
|
|
from utils import check_file_in_work, load_json
|
|
|
|
def process_item(sname, sdata):
|
|
data = {
|
|
"id": sdata.get("IO", ""),
|
|
"bezeichnung": sdata.get("BEZEICHNUNG", ""),
|
|
"verwendung": sdata.get("VERW", ""),
|
|
"kennzeichnung": sdata.get("KENNZEICHNUNG", ""),
|
|
"text_d": sdata.get("TEXT-D", ""),
|
|
"text_e": sdata.get("TEXT-E", ""),
|
|
"text_es": sdata.get("TEXT-ES", ""),
|
|
"text_f": sdata.get("TEXT-F", ""),
|
|
"sps": sdata.get("SPS", None),
|
|
}
|
|
reihenfolge = ["id", "bezeichnung", "verwendung", "kennzeichnung", "text_d", "text_e", "text_es", "text_f"]
|
|
# CSV-String erzeugen
|
|
csvstr = "','".join(str(data[key]) for key in reihenfolge)
|
|
csvstr = "'"+csvstr+"'"
|
|
|
|
sps_nr = sdata.get("SPS", None)
|
|
return csvstr, sps_nr
|
|
|
|
def prepare_data(rawdata:dict):
|
|
sensors = rawdata["sensors"]
|
|
schaltschrank_elemente = rawdata["schaltschrank_elemente"]
|
|
|
|
dsensors = dict()
|
|
lsensors = list()
|
|
|
|
for sname, sdata in sensors.items():
|
|
if "IO" not in sdata:
|
|
continue
|
|
csvstr, sps_nr = process_item(sname, sdata)
|
|
if sps_nr:
|
|
if sps_nr not in dsensors:
|
|
dsensors[sps_nr] = list()
|
|
dsensors[sps_nr].append(csvstr)
|
|
else:
|
|
lsensors.append(csvstr)
|
|
|
|
for schaltele, sdata in schaltschrank_elemente.items():
|
|
if "IO" not in sdata:
|
|
continue
|
|
csvstr, sps_nr = process_item(schaltele, sdata)
|
|
if sps_nr:
|
|
if sps_nr not in dsensors:
|
|
dsensors[sps_nr] = list()
|
|
dsensors[sps_nr].append(csvstr)
|
|
else:
|
|
lsensors.append(csvstr)
|
|
|
|
return (dsensors, lsensors)
|
|
|
|
|
|
# Helper zum Laden der Error-JSON
|
|
def load_error_json(json_path: Path, work_dir: Path):
|
|
"""
|
|
Lädt die Fehler-JSON-Datei, falls vorhanden.
|
|
|
|
Args:
|
|
json_path (Path): Pfad zur Haupt-JSON-Datei.
|
|
work_dir (Path): Arbeitsverzeichnis.
|
|
|
|
Returns:
|
|
tuple: (errors_dict, warnings_dict) oder (None, None) wenn keine Fehler vorhanden.
|
|
"""
|
|
# Konstruiere Error-JSON-Dateinamen basierend auf der Positions-JSON
|
|
basename = json_path.stem
|
|
if basename.endswith("_positionsconv"):
|
|
basename = basename.replace("_positionsconv", "")
|
|
elif basename.endswith("_positionsdraw"):
|
|
basename = basename.replace("_positionsdraw", "")
|
|
|
|
error_json_name = f"{basename}_errors.json"
|
|
error_json_path = work_dir / error_json_name
|
|
|
|
# Prüfen ob Error-JSON existiert
|
|
if not error_json_path.exists():
|
|
# Auch im gleichen Verzeichnis wie die Positions-JSON prüfen
|
|
error_json_path = json_path.parent / error_json_name
|
|
if not error_json_path.exists():
|
|
return None, None
|
|
|
|
# Error-JSON laden
|
|
try:
|
|
error_data = load_json(error_json_path)
|
|
errors_dict = error_data.get("errors", None)
|
|
warnings_dict = error_data.get("warnings", None)
|
|
return errors_dict, warnings_dict
|
|
except Exception as e:
|
|
print(f"Warnung: Fehler beim Laden der Error-JSON: {e}")
|
|
return None, None
|
|
|
|
# Helper für Export
|
|
def do_export(conv_obj:ExcelConverter, mode_str, base_input, output_dir, lines, getpos_errors=None, getpos_warnings=None):
|
|
# Für WSCAD Dateinamen anpassen (Export-Typ bleibt "WSCAD" oder "WSCAD mit Bezug")
|
|
if mode_str == "WSCAD":
|
|
if args.no_bezug:
|
|
export_type = "WSCAD"
|
|
filename = f"{base_input}_WSCAD_no_Bezug.xlsx"
|
|
else:
|
|
export_type = "WSCAD mit Bezug"
|
|
filename = f"{base_input}_WSCAD.xlsx"
|
|
else:
|
|
export_type = mode_str
|
|
filename = f"{base_input}_{mode_str}.xlsx"
|
|
|
|
export_path = os.path.join(output_dir, filename)
|
|
|
|
try:
|
|
conv_obj.export_excel(lines, export_type, export_path, getpos_errors, getpos_warnings)
|
|
print(f"Export '{export_type}' erfolgreich: {export_path}")
|
|
except Exception as e:
|
|
print(f"Fehler bei Export '{export_type}': {e}")
|
|
|
|
def do_exports(converter, export_mode, output_dir, base_input, lines, getpos_errors=None, getpos_warnings=None):
|
|
if export_mode:
|
|
do_export(converter, export_mode, base_input, output_dir, lines, getpos_errors, getpos_warnings)
|
|
else:
|
|
# Kein Modus → alle drei exportieren
|
|
do_export(converter, "EA", base_input, output_dir, lines, getpos_errors, getpos_warnings)
|
|
do_export(converter, "TIA", base_input, output_dir, lines, getpos_errors, getpos_warnings)
|
|
do_export(converter, "WSCAD", base_input, output_dir, lines, getpos_errors, getpos_warnings)
|
|
|
|
if __name__ == "__main__":
|
|
parser = argparse.ArgumentParser(description="Excel-Konverter für BMK-Daten")
|
|
parser.add_argument("--mode", type=str, choices=["EA", "TIA", "WSCAD"],
|
|
help="Art des Exports (EA, TIA, WSCAD). Wenn nicht gesetzt, werden die ersten drei erzeugt.")
|
|
parser.add_argument("--no_bezug", action="store_true", help="(Nur bei WSCAD) Export ohne Bezugsinformationen")
|
|
|
|
parser.add_argument("--input", type=str, help="Pfad zur Eingabedatei in Form einer Kommaseparierten .csv Datei (.txt)")
|
|
parser.add_argument("--filename", type=str, help="Json Datei von getpositions.py als Eingabe (.json)")
|
|
parser.add_argument("--export_dir", type=str, help="Pfad zum Ordner für die Ausgabe. Default ist PROJECT_WORK")
|
|
parser.add_argument("--outname", type=str, help="Basisname der Zieldateien wie <name>_TIA.xlsx)")
|
|
|
|
args = parser.parse_args()
|
|
|
|
# Umgebungsvariablen
|
|
work_dir = Path(os.environ.get("PROJECT_WORK"))
|
|
config_dir = Path(os.environ.get("PROJECT_CFG"))
|
|
|
|
|
|
# Converter initialisieren
|
|
converter = ExcelConverter()
|
|
output_dir = work_dir
|
|
if args.export_dir:
|
|
output_dir = args.export_dir
|
|
|
|
# Konfigurationsdatei
|
|
# config = configparser.ConfigParser(allow_no_value=True, delimiters=("="))
|
|
# config.optionxform = lambda option: option # preserve case for letters
|
|
# config.read(os.path.join(config_dir, "xxxx.cfg"))
|
|
export_mode = None
|
|
if args.mode:
|
|
export_mode = args.mode
|
|
|
|
|
|
if args.outname:
|
|
basename_out = args.outname
|
|
elif args.filename:
|
|
basename_out = args.filename.split('_')[0]
|
|
elif args.input:
|
|
basename_out = os.path.splitext(os.path.basename(args.input))[0]
|
|
else:
|
|
raise Exception()
|
|
|
|
# Datei einlesen
|
|
if args.input:
|
|
# Wenn Input Pfad kein absoluter Pfad, dann in work nach Dateinamen suchen
|
|
if not os.path.isabs(args.input):
|
|
if work_dir:
|
|
args.input = os.path.join(work_dir, args.input)
|
|
|
|
try:
|
|
with open(args.input, "r", encoding="utf-8") as f:
|
|
lines = f.readlines()
|
|
except Exception as e:
|
|
print(f"Fehler beim Lesen der Datei: {e}")
|
|
sys.exit(1)
|
|
do_exports(converter, export_mode, output_dir, basename_out, lines)
|
|
|
|
elif args.filename:
|
|
filename = Path(args.filename)
|
|
if not filename.suffix == ".json":
|
|
print("only available for .json files")
|
|
exit()
|
|
|
|
json_file = Path(args.filename)
|
|
(json_path, dexists) = check_file_in_work(work_dir, json_file)
|
|
if not dexists:
|
|
print(f"file {json_file} does not exist")
|
|
parser.print_help()
|
|
exit()
|
|
|
|
rawdata = load_json(json_path)
|
|
(dsensors, lsensors) = prepare_data(rawdata)
|
|
|
|
# Fehler-JSON laden (falls vorhanden)
|
|
getpos_errors, getpos_warnings = load_error_json(json_path, work_dir)
|
|
|
|
if len(dsensors) == 0:
|
|
do_exports(converter, export_mode, output_dir, basename_out, lsensors, getpos_errors, getpos_warnings)
|
|
else:
|
|
for sps_nr, lines in dsensors.items():
|
|
name = f"{basename_out}-{sps_nr}"
|
|
do_exports(converter, export_mode, output_dir, name, lines, getpos_errors, getpos_warnings)
|
|
|
|
else:
|
|
parser.print_help()
|
|
exit()
|
|
|
|
|
|
|
|
|