# -*- coding: utf-8 -*- """create_skel.py — Skeleton-Generator fuer die JSON-Zwischendatei und ein SCL-Grundgeruest, abgeleitet aus den Eingabelisten in data/. Datenquellen (Pfade ueber Umgebungsvariablen aus bin\\setenv.bat): PV_DATA Ordner mit den Eingabelisten 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 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 Json_Layout-Konzept.md) FB_Main_skel_UH0x.scl SCL-Grundgeruest (VAR-Deklaration + Aufrufe) FB_CallSensors_skel_UH0x.scl Sensor-Aufrufe 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 import sys from collections import defaultdict from pathlib import Path try: import openpyxl except ImportError: sys.exit("openpyxl fehlt — bitte bin\\install_py.bat ausfuehren.") # ============================================================================ # GLOBALE DEFINITIONEN (eingebaute Defaults) # Alle folgenden Bloecke sind bewusst als Modul-Konstanten definiert. Sie # sind zugleich der Inhalt von cfg\\skel.cfg (JSON) — load_cfg() liest diese # Datei aus PV_CFG beim Programmstart und ersetzt die Defaults hier 1:1 # durch ihren Inhalt. Ohne skel.cfg (z.B. Erststart, Datei geloescht) bleiben # die Defaults unten aktiv, das Skript funktioniert also auch dann. # ============================================================================ #: Dateinamen der Eingabelisten ({n} = ILS-Nummer 1..5) FILES = { "tia_xlsx": "ST500592_10_5-14_ILS-{n}_TIA.xlsx", "positions_json": "ST500592_10_5-9_ILS_positions.json", # Routing / Zusammenhalt der TROs (hmf-connect-v1, extrahiert aus # doc/TRO_Graph_UH01-UH05.dot) — fuer die kuenftige Topologie-Auswertung "connect_json": "connect.json", "connect_ini": "connect.ini", } #: VERW-Vokabular (Positions-JSON) -> Signalrolle #: (siehe doc/EA-Listen-Analyse.md, Abschnitt "Verwendung von Positionen") VERW_ROLES = { "In sep": "sensorInSep", "Sep": "stopperOutput", "Jam detector": "sensorJam", "Jam detector (LP)": "sensorJam", "jam detector (LP)": "sensorJam", "ES branch": "switchOutput", "ES brunch": "switchOutput", # Schreibvariante in den Daten "conveyor full": "sensorConvFull", "MT in Position": "sensorCarrInPos", "MT in position": "sensorCarrInPos", "Finger in position": "sensorFinger", "Finger in Position": "sensorFinger", "PIN query": "sensorPinQuery", "Rad dreht ": "sensorWheel", "Rad dreht": "sensorWheel", } #: Fallback: Regex auf den Kommentartext -> Signalrolle #: (falls kein VERW vorliegt, z.B. Signal nicht im Positions-JSON) COMMENT_ROLES = [ (r"^in Separator", "sensorInSep"), (r"^Separator", "stopperOutput"), (r"Stausensor|Staumelder", "sensorJam"), (r"AE Abzweig", "switchOutput"), (r"F.?rderer voll", "sensorConvFull"), (r"MT in Position", "sensorCarrInPos"), (r"Mitnehmerfinger|Finger in", "sensorFinger"), (r"Pin Abfrage|Pr.?fung MT Pos", "sensorPinQuery"), (r"Rad dreht", "sensorWheel"), (r"^Motor ", "motorOutput"), (r"MSS ausgel", "motorProtection"), ] #: Regeln zur FB-Typ-Bestimmung je Einheit. #: Reihenfolge = Prioritaet; "roles" muessen alle vorhanden sein. FBTYPE_RULES = [ {"roles": ["sensorPinQuery"], "fbType": "FB_ILS_MTRO_PinStore_Auto"}, {"roles": ["sensorFinger"], "fbType": "FB_ILS_MTRO_Vario"}, {"roles": ["sensorCarrInPos"], "fbType": "FB_ILS_MTRO_Vario"}, {"roles": ["switchOutput", "sensorInSep"], "fbType": "FB_ILS_MTRO_1Sep1Swi"}, {"roles": ["sensorInSep"], "fbType": "FB_ILS_MTRO_1Sep"}, {"roles": ["motorOutput"], "fbType": "FB_Conveyor"}, {"roles": ["motorProtection"], "fbType": "FB_Conveyor"}, ] #: Erkennung der Foerdertechnik-Einheit (M-Nummer) im Kommentartext UNIT_RE = r"\b(KR|CV)[-\s]{0,2}(P?[MK]\d{4}(?:_[A-Z])?)\b" #: Erkennung der Bereichsgruppen im Kommentartext AREA_RES = [ (r"Block ?(\d\.\d(?:\.\d)?)", "pinStoreBlock", "Block {m}"), (r"\bLS ?(\d)\b", "loadingStation", "LS{m}"), (r"\bWS ?(\d\.\d)\b", "workStation", "WS{m}"), (r"\bUZ(\d{4})\b", "safetyZone", "UZ{m}"), ] # ---------------------------------------------------------------------------- # 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 "var_decl": ' {inst} : "{fbtype}"; // {comment}\n', # Aufruf FB_ILS_MTRO_1Sep (Standard-Separator) "call_1sep": """\ REGION {inst} ({shorttype}) — {comment} #{inst}.stInSeparator1.Settings.nSeparatorNo := "cSep{tro_no}.1"; #{inst}.stInSeparator1.Settings.tDelayToNextItem := "TimeForSepWaiting_Short"; #{inst}.stInSeparator1.Settings.tTrailingTime := T#500ms; // TODO Timing #{inst}.stInSeparator1.Settings.tHandlingTime := T#200ms; // TODO Timing #{inst}.stInSeparator1.Settings.tJam := "TimeForSepJam"; #{inst}.stInSeparator1.stSenInSep := "DB_Inputs".Sensors["cIn{sen_in_sep}"]; #{inst}.stInSenJam2 := "DB_Inputs".Sensors["cIn{sen_jam}"]; #{inst}.stInPriorityManager.xReleaseOk := TRUE; // TODO Priority #{inst}(nInMainTroNo := "cMainTro{tro_no}", xInSftyOk := TRUE, // TODO Safety xInAllRdyToStart := TRUE, xInCarouselRun := TRUE, // TODO Conveyor xInRelease := TRUE, // TODO Release stInOutMachineState := #stInOutControlUnit, // TODO ControlUnit arInOutJamEntr1 := "DB_JamArea{tro_no}".stJam{tro_no}_1.arCarrier, // TODO Topologie stInOutJamEntr1 := "DB_JamArea{tro_no}".stJam{tro_no}_1.stData, arInOutJamExit2 := "DB_JamArea{tro_no}".stJam{tro_no}_2.arCarrier, // TODO Topologie stInOutJamExit2 := "DB_JamArea{tro_no}".stJam{tro_no}_2.stData, xOutStopper1 => "{stopper}", nOutStateLast => "DB_GLOB_TroState".arTroState["cMainTro{tro_no}"], stInOutHMI := "DB_Interface_HMI".stTRO.{inst}); END_REGION """, # Aufruf FB_ILS_MTRO_1Sep1Swi (Separator + Weiche) "call_1sep1swi": """\ REGION {inst} ({shorttype}) — {comment} #{inst}.stInSeparator1.Settings.nSeparatorNo := "cSep{tro_no}.1"; #{inst}.stInSeparator1.Settings.tDelayToNextItem := "TimeForSepWaiting_Short"; #{inst}.stInSeparator1.Settings.tTrailingTime := T#100ms; // TODO Timing #{inst}.stInSeparator1.Settings.tHandlingTime := T#50ms; // TODO Timing #{inst}.stInSeparator1.Settings.tJam := "TimeForSepJam"; #{inst}.stInSeparator1.stSenInSep := "DB_Inputs".Sensors["cIn{sen_in_sep}"]; #{inst}.stInSenJam2 := "DB_Inputs".Sensors["cIn{sen_jam}"]; #{inst}.stInSwitch1.Settings.nSwitchNo := "cSwi{tro_no}.1"; #{inst}(nInMainTroNo := "cMainTro{tro_no}", xInSftyOk := TRUE, // TODO Safety xInAllRdyToStart := TRUE, xInCarouselRun1 := TRUE, // TODO Conveyor xInRelease := TRUE, // TODO Release stInOutMachineState := #stInOutControlUnit, // TODO ControlUnit arInOutJamEntr1 := "DB_JamArea{tro_no}".stJam{tro_no}_1.arCarrier, // TODO Topologie stInOutJamEntr1 := "DB_JamArea{tro_no}".stJam{tro_no}_1.stData, arInOutJamExit2 := "DB_JamArea{tro_no}".stJam{tro_no}_2.arCarrier, // TODO Topologie (Dir1) stInOutJamExit2 := "DB_JamArea{tro_no}".stJam{tro_no}_2.stData, arInOutJamExit3 := "DB_JamArea{tro_no}".stJam{tro_no}_3.arCarrier, // TODO Topologie (Dir2) stInOutJamExit3 := "DB_JamArea{tro_no}".stJam{tro_no}_3.stData, xOutStopper1 => "{stopper}", xOutSw1ExTo3 => "{switch_out}", nOutStateLast => "DB_GLOB_TroState".arTroState["cMainTro{tro_no}"], stInOutHMI := "DB_Interface_HMI".stTRO.{inst}); END_REGION """, # Aufruf FB_ILS_MTRO_Vario (Kettenfoerderer) "call_vario": """\ REGION {inst} ({shorttype}) — {comment} #{inst}.stInSeparator1.Settings.nSeparatorNo := "cSep{tro_no}.1"; #{inst}.stInSeparator1.Settings.tDelayToNextItem := "TimeForSepWaiting_Short"; #{inst}.stInSeparator1.stSenInSep := "DB_Inputs".Sensors["cIn{sen_in_sep}"]; #{inst}.stInVario.stSenCarrInPos := "DB_Inputs".Sensors["cIn{sen_carr_in_pos}"]; #{inst}.stInVario.stSenFinger := "DB_Inputs".Sensors["cIn{sen_finger}"]; #{inst}.stInVario.stSenJam := "DB_Inputs".Sensors["cIn{sen_jam}"]; #{inst}(nInMainTroNo := "cMainTro{tro_no}", xInSftyOk := TRUE, // TODO Safety xInAllRdyToStart := TRUE, xInRelease := TRUE, // TODO Release stInOutMachineState := #stInOutControlUnit, // TODO ControlUnit arInOutJamEntr1 := "DB_JamArea{tro_no}".stJam{tro_no}_1.arCarrier, // TODO Topologie stInOutJamEntr1 := "DB_JamArea{tro_no}".stJam{tro_no}_1.stData, arInOutJamFinger := "DB_JamArea{tro_no}".stJam{tro_no}_2.arCarrier, // TODO Topologie xOutStopper => "{stopper}", xOutVarioMotor => "{motor}", nOutStateLast => "DB_GLOB_TroState".arTroState["cMainTro{tro_no}"], stInOutHMI := "DB_Interface_HMI".stTRO.{inst}); END_REGION """, # Aufruf FB_Conveyor (Kreisel / Foerderer ohne TRO-Logik) "call_conveyor": """\ REGION {inst} — {comment} #{inst}.Settings.tEnergySafeTime := "cTimeEnergySafe"; #{inst}.Settings.xGridActive := FALSE; // TODO Grid #{inst}.Settings.sMotorProtectName := '{motor_protection}'; #{inst}(xInEnabled := TRUE, // TODO Enable xInMotorProtection := "{motor_protection}", xInPowerContactor := "{motor}", stInOutHMI := "DB_Interface_HMI".stConveyors.{inst}); END_REGION """, # Eintrag in FB_CallSensors je Sensor "call_sensor": """\ // {comment} #fbSensorInput(stInSensorHW := "{bmk}", nInSensorNo := "cIn{bmk}", stInOutSensor := "DB_Inputs".Sensors["cIn{bmk}"]); """, # Rahmen FB_Main "fb_main_frame": """\ FUNCTION_BLOCK "FB_Main_skel_{uh}" {{ S7_Optimized_Access := 'TRUE' }} VERSION : 0.1 // ============================================================ // AUTOMATISCH GENERIERTES GRUNDGERUEST — create_skel.py // Quelle: E/A-Listen {uh} (data/) // TODO-Marken kennzeichnen manuell zu ergaenzende Stellen // (Timings, Topologie/JamAreas, Safety, Release, Priority). // ============================================================ VAR {var_block} END_VAR BEGIN {call_block}END_FUNCTION_BLOCK """, # Rahmen FB_CallSensors "fb_sensors_frame": """\ FUNCTION_BLOCK "FB_CallSensors_skel_{uh}" {{ S7_Optimized_Access := 'TRUE' }} VERSION : 0.1 // AUTOMATISCH GENERIERT — create_skel.py ({count} Sensoren) VAR fbSensorInput : "FB_SensorInput"; END_VAR BEGIN {call_block}END_FUNCTION_BLOCK """, } # Zuordnung FB-Typ -> Template-Schluessel FBTYPE_TEMPLATE = { "FB_ILS_MTRO_1Sep": "call_1sep", "FB_ILS_MTRO_1Sep1Swi": "call_1sep1swi", "FB_ILS_MTRO_Vario": "call_vario", "FB_ILS_MTRO_PinStore_Auto": "call_1sep", # eigenes Template folgt spaeter "FB_Conveyor": "call_conveyor", } # ============================================================================ # Ende der globalen Definitionen # ============================================================================ #: Schluessel in skel.cfg -> Modul-Konstante, die beim Laden ersetzt wird _CFG_KEYS = { "files": "FILES", "verw_roles": "VERW_ROLES", "comment_roles": "COMMENT_ROLES", "fbtype_rules": "FBTYPE_RULES", "unit_re": "UNIT_RE", "area_res": "AREA_RES", "scl_templates": "SCL_TEMPLATES", "fbtype_template": "FBTYPE_TEMPLATE", } def load_cfg(): """Laedt cfg\\skel.cfg (JSON) aus PV_CFG und ersetzt die globalen Definitionen (FILES, VERW_ROLES, ..., SCL_TEMPLATES) durch ihren Inhalt. Fehlt PV_CFG oder die Datei, bleiben die eingebauten Defaults oben aktiv (Cold-Start-Faehigkeit ohne Konfigurationsdatei). """ cfg_dir = os.environ.get("PV_CFG", Path(__file__).resolve().parent.parent / "cfg") cfg_file = Path(cfg_dir) / "skel.cfg" if not cfg_file.exists(): print(f"Hinweis: {cfg_file} nicht gefunden — verwende eingebaute Defaults.") return print(f"Lade Konfiguration: {cfg_file}") data = json.loads(cfg_file.read_text(encoding="utf-8")) globals_ = globals() for cfg_key, const_name in _CFG_KEYS.items(): if cfg_key not in data: continue value = data[cfg_key] # Listen von Listen (comment_roles, fbtype_rules, area_res) als # Tupel normalisieren, damit das Unpacking im Code unveraendert bleibt if const_name in ("COMMENT_ROLES", "AREA_RES"): value = [tuple(item) for item in value] 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)) data_dir = Path(os.environ.get("PV_DATA", root / "data")) results = Path(os.environ.get("PV_RESULTS", root / "results")) results.mkdir(parents=True, exist_ok=True) return root, data_dir, results # ---------------------------------------------------------------------------- # Einlesen # ---------------------------------------------------------------------------- def read_positions(data_dir): """Positions-JSON: (BMK, SPS) -> {verw, pos, cabinet, bezeichnung}.""" path = data_dir / FILES["positions_json"] if not path.exists(): print(f"WARNUNG: {path} nicht gefunden — Positionen/VERW entfallen.") return {} data = json.loads(path.read_text(encoding="utf-8")) index = {} for section in ("sensors", "schaltschrank_elemente"): for key, val in data.get(section, {}).items(): if "@" not in key: continue bmk, sps = key.rsplit("@", 1) index[(bmk, sps)] = { "verw": (val.get("VERW") or "").strip(), "pos": val.get("pos"), "bezeichnung": val.get("BEZEICHNUNG") or "", "kennzeichnung": val.get("KENNZEICHNUNG") or "", } # mappings: Verteiler -> [BMK@SPS] umdrehen zu (BMK, SPS) -> Verteiler for dist, members in data.get("mappings", {}).items(): for key in members: if "@" not in key: continue bmk, sps = key.rsplit("@", 1) index.setdefault((bmk, sps), {})["cabinet"] = dist return index def read_tia(data_dir, n): """TIA-Liste ILS-n: Liste von Tag-Dicts + Konstanten-Map.""" path = data_dir / FILES["tia_xlsx"].format(n=n) wb = openpyxl.load_workbook(path, read_only=True, data_only=True) tags = [] for row in wb["PLC Tags"].iter_rows(min_row=2, values_only=True): name, _path, dtype, addr, comment = (tuple(row) + (None,) * 5)[:5] if not name: continue tags.append({ "name": str(name), "dataType": str(dtype) if dtype else "Bool", "address": str(addr) if addr else "", "comment": str(comment) if comment else "", }) constants = {} if "Constants" in wb.sheetnames: for row in wb["Constants"].iter_rows(min_row=2, values_only=True): cname, _path, _dt, value = (tuple(row) + (None,) * 4)[:4] if cname: constants[str(cname)] = value 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 # ---------------------------------------------------------------------------- def classify_role(tag, posinfo): """Signalrolle: primaer VERW-Vokabular, sonst Kommentar-Heuristik.""" verw = (posinfo or {}).get("verw", "") if verw in VERW_ROLES: return VERW_ROLES[verw] for pattern, role in COMMENT_ROLES: if re.search(pattern, tag["comment"], re.I): return role return "" def extract_unit(comment): """M-Nummer der Foerdertechnik-Einheit aus dem Kommentar.""" m = re.search(UNIT_RE, comment) if m: return m.group(2), m.group(1) # ("M0101", "KR") return None, None def extract_areas(comment): """Bereichsgruppen (Block/LS/WS/UZ) aus dem Kommentar.""" found = [] for pattern, kind, fmt in AREA_RES: for m in re.finditer(pattern, comment, re.I): found.append({"id": fmt.format(m=m.group(1)), "kind": kind}) return found def guess_fbtype(roles): """FB-Typ einer Einheit aus der Menge ihrer Signalrollen.""" for rule in FBTYPE_RULES: if all(r in roles for r in rule["roles"]): return rule["fbType"] return "UNBEKANNT" # ---------------------------------------------------------------------------- # Skeleton-Aufbau # ---------------------------------------------------------------------------- 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) sensors = [] units = defaultdict(lambda: {"signals": [], "roles": set(), "areas": [], "kind": None, "pos": None, "cabinets": set()}) areas = defaultdict(lambda: {"kind": None, "members": set(), "cabinets": set()}) unassigned = [] for tag in tags: posinfo = positions.get((tag["name"], sps)) role = classify_role(tag, posinfo) unit_id, unit_kind = extract_unit(tag["comment"]) area_hits = extract_areas(tag["comment"]) cabinet = (posinfo or {}).get("cabinet") entry = { "id": tag["name"], "constant": f"cIn{tag['name']}" if tag["address"].startswith("%E") else None, "hwAddress": tag["address"], "comment": tag["comment"], "role": role or None, "unit": unit_id, "cabinet": cabinet, "pos": (posinfo or {}).get("pos"), } if tag["address"].startswith("%E"): sensors.append(entry) if unit_id: u = units[unit_id] u["kind"] = unit_kind u["signals"].append(entry) if role: u["roles"].add(role) if u["pos"] is None and entry["pos"]: u["pos"] = entry["pos"] if cabinet: u["cabinets"].add(cabinet) for a in area_hits: if a not in u["areas"]: u["areas"].append(a) elif not area_hits and not role: unassigned.append(tag["name"]) for a in area_hits: ar = areas[a["id"]] ar["kind"] = a["kind"] if unit_id: ar["members"].add(unit_id) if cabinet: ar["cabinets"].add(cabinet) unit_list = [] for unit_id in sorted(units): u = units[unit_id] fbtype = guess_fbtype(u["roles"]) io = {} for sig in u["signals"]: if sig["role"]: io.setdefault(sig["role"], []).append(sig["id"]) unit_list.append({ "unit": unit_id, "unitKind": u["kind"], "fbType": fbtype, "areas": u["areas"], "cabinet": sorted(u["cabinets"])[0] if u["cabinets"] else None, "pos": u["pos"], "io": io, "signalCount": len(u["signals"]), }) area_list = [{ "id": aid, "kind": a["kind"], "members": sorted(a["members"]), "cabinets": sorted(a["cabinets"]), } for aid, a in sorted(areas.items())] skel = { "$schema": "hmf-layout-v1.schema.json", "generatedBy": "create_skel.py", "plc": {"id": uh, "configName": f"=A01+{uh}-KF00"}, "statistics": { "tags": len(tags), "sensors": len(sensors), "constants": len(constants), "units": len(unit_list), "areas": len(area_list), "unassignedSignals": len(unassigned), }, "sensors": sensors, "units": unit_list, "areas": area_list, "unassignedSignals": sorted(unassigned), "todo": [ "connections[] (Topologie) manuell ergaenzen oder aus SCL extrahieren", "Timings, Priority, Release, customCode manuell pflegen", "TRO-IDs den Einheiten zuweisen (Schema TRO)", ], } 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 # ---------------------------------------------------------------------------- def first(io, role, default="BGxxxx"): vals = io.get(role) return vals[0] if vals else default def render_scl(skeleton, uh): """FB_Main- und FB_CallSensors-Grundgeruest aus den Templates.""" var_block = [] call_block = [] tro_seq = 0 for u in skeleton["units"]: fbtype = u["fbType"] if fbtype == "UNBEKANNT": continue tro_seq += 1 inst = f"TRO{uh[-1]}{tro_seq:02d}" if fbtype != "FB_Conveyor" \ else f"fbConveyor{u['unit'][1:]}" comment = f"{u['unitKind']}-{u['unit']}" + \ (f" / {u['areas'][0]['id']}" if u["areas"] else "") var_block.append(SCL_TEMPLATES["var_decl"].format( inst=inst, fbtype=fbtype, comment=comment)) tpl = SCL_TEMPLATES[FBTYPE_TEMPLATE[fbtype]] io = u["io"] call_block.append(tpl.format( inst=inst, shorttype=fbtype.replace("FB_ILS_MTRO_", "").replace("FB_", ""), comment=comment, tro_no=f"{uh[-1]}{tro_seq:02d}", sen_in_sep=first(io, "sensorInSep"), sen_jam=first(io, "sensorJam"), sen_finger=first(io, "sensorFinger"), sen_carr_in_pos=first(io, "sensorCarrInPos"), stopper=first(io, "stopperOutput", "MBxxxx"), switch_out=first(io, "switchOutput", "MBxxxx"), motor=first(io, "motorOutput", "MAxxxx"), motor_protection=first(io, "motorProtection", "FCxxxx"), )) fb_main = SCL_TEMPLATES["fb_main_frame"].format( uh=uh, var_block="".join(var_block), call_block="".join(call_block)) sensor_calls = [SCL_TEMPLATES["call_sensor"].format( bmk=s["id"], comment=s["comment"]) for s in skeleton["sensors"]] fb_sensors = SCL_TEMPLATES["fb_sensors_frame"].format( uh=uh, count=len(sensor_calls), call_block="".join(sensor_calls)) return fb_main, fb_sensors # ---------------------------------------------------------------------------- # main # ---------------------------------------------------------------------------- def main(): ap = argparse.ArgumentParser(description=__doc__) ap.add_argument("--uh", type=int, choices=range(1, 6), 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}") 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, 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{topo_info} -> {out_json.name}") if not args.no_scl: fb_main, fb_sensors = render_scl(skel, uh) (results / f"FB_Main_skel_{uh}.scl").write_text( fb_main, encoding="utf-8", newline="\n") (results / f"FB_CallSensors_skel_{uh}.scl").write_text( fb_sensors, encoding="utf-8", newline="\n") print(f"{uh}: FB_Main_skel_{uh}.scl, FB_CallSensors_skel_{uh}.scl") if __name__ == "__main__": main()