From 9818ca0c321496f717ad8aacd439187978b235ad Mon Sep 17 00:00:00 2001 From: Michael Stangl Date: Tue, 7 Jul 2026 09:55:10 +0200 Subject: [PATCH] scl Skeletcode aus eigener config lesen --- lib/create_skel.py | 194 ++++++++++++++++++++++++++++++++++++++++++--- 1 file changed, 184 insertions(+), 10 deletions(-) diff --git a/lib/create_skel.py b/lib/create_skel.py index 6798f60..9ee3972 100644 --- a/lib/create_skel.py +++ b/lib/create_skel.py @@ -7,11 +7,13 @@ Datenquellen (Pfade ueber Umgebungsvariablen aus bin\\setenv.bat): ST500592_10_5-14_ILS-_TIA.xlsx (PLC Tags + Constants) ST500592_10_5-9_ILS_positions.json (Positionen, VERW, mappings) PV_RESULTS Zielordner fuer die erzeugten Dateien - PV_CFG Ordner mit skel.cfg (JSON) — enthaelt die globalen - Definitionen (Dateinamen, Vokabular, FB-Typ-Regeln, - SCL-Templates). Wird beim Programmstart geladen und - ersetzt die eingebauten Defaults vollstaendig, wenn - vorhanden. Ohne skel.cfg laufen die Defaults unten weiter. + PV_CFG Ordner mit den Konfigurationsdateien, geladen beim Start: + skel.cfg (JSON) Dateinamen, Vokabular, FB-Typ-Regeln + skel_scl.cfg (INI-Style) SCL-Templates, zeilenweise + lesbar; jede [sektion] = ein Template, + Inhalt verbatim (Einrueckung/Leerzeilen + gehoeren zum Template) + Fehlende Dateien fallen auf die eingebauten Defaults zurueck. Erzeugt je Steuerung UH01..UH05: skeleton_UH0x.json JSON-Zwischendatei-Skeleton (Kap. 14 im @@ -19,10 +21,17 @@ Erzeugt je Steuerung UH01..UH05: FB_Main_skel_UH0x.scl SCL-Grundgeruest (VAR-Deklaration + Aufrufe) FB_CallSensors_skel_UH0x.scl Sensor-Aufrufe -Aufruf: bin\\create_skel.bat [--uh 1..5] [--no-scl] +Mit --con-ini bzw. --con-json wird zusaetzlich die TRO-Topologie +(Routing / Zusammenhalt, Format hmf-connect-v1) aus data\\connect.ini +bzw. data\\connect.json eingelesen und als "topology"-Abschnitt in die +Skeleton-JSONs uebernommen (Verbindungen, offene Enden und je TRO die +aufgeloeste JamArea-Verdrahtung). + +Aufruf: bin\\create_skel.bat [--uh 1..5] [--no-scl] [--con-ini | --con-json] """ import argparse +import configparser import json import os import re @@ -116,6 +125,8 @@ AREA_RES = [ # ---------------------------------------------------------------------------- # SCL-Templates: die "ueblichen Codebloecke" fuer die Aufrufe in FB_Main / # FB_CallSensors. Platzhalter im str.format()-Stil. +# Eingebaute Defaults — die gepflegte, zeilenweise lesbare Fassung liegt in +# cfg\skel_scl.cfg (INI-Style) und ueberschreibt diese beim Programmstart. # ---------------------------------------------------------------------------- SCL_TEMPLATES = { # VAR-Deklaration einer Instanz @@ -319,6 +330,47 @@ def load_cfg(): globals_[const_name] = value +def load_scl_cfg(): + """Laedt die SCL-Templates aus PV_CFG\\skel_scl.cfg (INI-Style). + + Format: jede [sektion] ist ein Template-Schluessel (var_decl, call_1sep, + ...); der Sektionsinhalt wird VERBATIM uebernommen — Einrueckung und + Leerzeilen (auch am Blockende) gehoeren zum Template. Zeilen, die mit + ';' beginnen, sind Kommentare. SCL selbst beginnt nie mit ';', daher + ist das kollisionsfrei. + + Fehlt die Datei, bleiben die Templates aus skel.cfg bzw. den eingebauten + Defaults aktiv. + """ + cfg_dir = os.environ.get("PV_CFG", Path(__file__).resolve().parent.parent / "cfg") + cfg_file = Path(cfg_dir) / "skel_scl.cfg" + if not cfg_file.exists(): + print(f"Hinweis: {cfg_file} nicht gefunden — " + f"verwende SCL-Templates aus skel.cfg/Defaults.") + return + + print(f"Lade SCL-Templates: {cfg_file}") + templates = {} + name, buf = None, [] + + def flush(): + if name is not None: + templates[name] = "\n".join(buf) + "\n" + + for line in cfg_file.read_text(encoding="utf-8").splitlines(): + if line.startswith(";"): + continue + m = re.fullmatch(r"\[(\w+)\]\s*", line) + if m: + flush() + name, buf = m.group(1), [] + elif name is not None: + buf.append(line) + flush() + SCL_TEMPLATES.update(templates) + print(f"SCL-Templates geladen: {', '.join(templates)}") + + def project_paths(): """Pfade aus den setenv.bat-Umgebungsvariablen (mit Fallbacks).""" root = Path(os.environ.get("PROJECT", Path(__file__).resolve().parent.parent)) @@ -385,6 +437,92 @@ def read_tia(data_dir, n): return tags, constants +# ---------------------------------------------------------------------------- +# Connect-Daten (Topologie, Format hmf-connect-v1) +# ---------------------------------------------------------------------------- + +def expand_jam(short): + """JA0101_1 -> "DB_JamArea0101".stJam0101_1 ; sonst unveraendert.""" + m = re.fullmatch(r"JA(\w+)_(\w+)", short) + if m: + return f'"DB_JamArea{m.group(1)}".stJam{m.group(1)}_{m.group(2)}' + return short + + +def read_connect_json(data_dir): + """connect.json -> {UHxx: {nodes, connections, externals}}.""" + path = data_dir / FILES["connect_json"] + data = json.loads(path.read_text(encoding="utf-8")) + return data.get("plcs", {}) + + +def read_connect_ini(data_dir): + """connect.ini -> gleiche Struktur wie read_connect_json().""" + path = data_dir / FILES["connect_ini"] + cp = configparser.ConfigParser(interpolation=None) + cp.optionxform = str # Gross-/Kleinschreibung der TRO-Namen erhalten + cp.read(path, encoding="utf-8") + plcs = {} + for section in cp.sections(): + uh, _, part = section.partition(".") + plc = plcs.setdefault(uh, {"nodes": {}, "connections": [], + "externals": []}) + if part == "nodes": + for name, val in cp[section].items(): + fbtype, _, comment = (x.strip() for x in val.partition("|")) + plc["nodes"][name] = {"fbType": fbtype, "comment": comment} + elif part == "connections": + for _, val in cp[section].items(): + fields = [x.strip() for x in val.split("|")] + src, _, dst = (x.strip() for x in fields[0].partition("->")) + plc["connections"].append({ + "from": src, "to": dst, "jamArea": fields[1], + "kind": fields[2] if len(fields) > 2 else "normal", + "dbRef": expand_jam(fields[1])}) + elif part == "externals": + for _, val in cp[section].items(): + route, _, jam = (x.strip() for x in val.partition("|")) + src, _, dst = (x.strip() for x in route.partition("->")) + node, direction = (dst, "in") if src == "EXTERN" else (src, "out") + plc["externals"].append({ + "node": node, "direction": direction, "jamArea": jam, + "dbRef": expand_jam(jam)}) + return plcs + + +def derive_topology(topo): + """Bereitet die Connect-Daten einer Steuerung fuer das Skeleton auf: + je TRO die aufgeloeste JamArea-Verdrahtung (entries/exits).""" + tros = {name: {"id": name, "fbType": nd["fbType"], "comment": nd["comment"], + "entries": [], "exits": []} + for name, nd in topo["nodes"].items()} + for c in topo["connections"]: + db = c.get("dbRef") or expand_jam(c["jamArea"]) + if c["from"] in tros: + tros[c["from"]]["exits"].append( + {"jamArea": c["jamArea"], "dbRef": db, "to": c["to"], + "kind": c["kind"]}) + if c["to"] in tros: + tros[c["to"]]["entries"].append( + {"jamArea": c["jamArea"], "dbRef": db, "from": c["from"], + "kind": c["kind"]}) + for e in topo["externals"]: + if e["node"] not in tros: + continue + db = e.get("dbRef") or expand_jam(e["jamArea"]) + side = "entries" if e["direction"] == "in" else "exits" + peer = "from" if e["direction"] == "in" else "to" + tros[e["node"]][side].append( + {"jamArea": e["jamArea"], "dbRef": db, peer: "EXTERN", + "kind": "extern"}) + return { + "format": "hmf-connect-v1", + "tros": [tros[k] for k in sorted(tros)], + "connections": topo["connections"], + "externals": topo["externals"], + } + + # ---------------------------------------------------------------------------- # Klassifikation # ---------------------------------------------------------------------------- @@ -429,7 +567,7 @@ def guess_fbtype(roles): # Skeleton-Aufbau # ---------------------------------------------------------------------------- -def build_skeleton(n, tags, constants, positions): +def build_skeleton(n, tags, constants, positions, connect=None): """Baut das JSON-Skeleton fuer eine Steuerung (ILS-n = UH0n).""" uh = f"UH0{n}" sps = str(n) @@ -511,7 +649,7 @@ def build_skeleton(n, tags, constants, positions): "cabinets": sorted(a["cabinets"]), } for aid, a in sorted(areas.items())] - return { + skel = { "$schema": "hmf-layout-v1.schema.json", "generatedBy": "create_skel.py", "plc": {"id": uh, "configName": f"=A01+{uh}-KF00"}, @@ -534,6 +672,23 @@ def build_skeleton(n, tags, constants, positions): ], } + if connect is not None: + topo = connect.get(uh) + if topo: + skel["topology"] = derive_topology(topo) + skel["statistics"]["troNodes"] = len(topo["nodes"]) + skel["statistics"]["connections"] = len(topo["connections"]) + skel["statistics"]["externals"] = len(topo["externals"]) + skel["todo"] = [ + "topology[] aus connect-Daten uebernommen — TRO-IDs den " + "Einheiten (units[]) noch zuordnen", + "Timings, Priority, Release, customCode manuell pflegen", + ] + else: + print(f"WARNUNG: keine Connect-Daten fuer {uh} gefunden.") + + return skel + # ---------------------------------------------------------------------------- # SCL-Rendering @@ -599,9 +754,15 @@ def main(): help="nur diese Steuerung (1..5), sonst alle") ap.add_argument("--no-scl", action="store_true", help="nur JSON-Skeleton, kein SCL-Grundgeruest") + con = ap.add_mutually_exclusive_group() + con.add_argument("--con-ini", action="store_true", + help="Topologie aus data\\connect.ini uebernehmen") + con.add_argument("--con-json", action="store_true", + help="Topologie aus data\\connect.json uebernehmen") args = ap.parse_args() load_cfg() + load_scl_cfg() root, data_dir, results = project_paths() print(f"Daten: {data_dir}") print(f"Results: {results}") @@ -609,17 +770,30 @@ def main(): positions = read_positions(data_dir) print(f"Positions-JSON: {len(positions)} Signale indiziert") + connect = None + if args.con_ini or args.con_json: + src = FILES["connect_ini"] if args.con_ini else FILES["connect_json"] + if not (data_dir / src).exists(): + sys.exit(f"FEHLER: {data_dir / src} nicht gefunden.") + connect = (read_connect_ini if args.con_ini + else read_connect_json)(data_dir) + total = sum(len(p["connections"]) for p in connect.values()) + print(f"Connect-Daten: {src} — {len(connect)} PLCs, " + f"{total} Verbindungen") + for n in ([args.uh] if args.uh else range(1, 6)): uh = f"UH0{n}" tags, constants = read_tia(data_dir, n) - skel = build_skeleton(n, tags, constants, positions) + skel = build_skeleton(n, tags, constants, positions, connect) out_json = results / f"skeleton_{uh}.json" out_json.write_text(json.dumps(skel, indent=2, ensure_ascii=False), encoding="utf-8", newline="\n") s = skel["statistics"] + topo_info = (f", {s['connections']} Verbindungen" + if "connections" in s else "") print(f"{uh}: {s['tags']} Tags, {s['units']} Einheiten, " - f"{s['areas']} Bereiche -> {out_json.name}") + f"{s['areas']} Bereiche{topo_info} -> {out_json.name}") if not args.no_scl: fb_main, fb_sensors = render_scl(skel, uh)