# -*- coding: utf-8 -*- # Absichtlich ohne Shebang: der Windows-py-Launcher wuerde daraus "python3" # ableiten und scheitert, wenn keine PythonCore-Installation registriert ist. """ scl_skeleton.py - FB_Main-Geruest als SCL aus der TRO-Liste erzeugen. Letzter Schritt der Kette: material_flow.py Objekte -> Materialfluss tro_flow.py Fluss -> TROs tro_annotate.py TROs -> Marker in der Zeichnung tro_extract.py Zeichnung -> JSON scl_skeleton.py JSON -> FB_Main.scl <- hier Erzeugt wird ein **Geruest**, kein fertiges Programm: je TRO eine REGION mit Instanzaufruf und allen Parameterzeilen in der richtigen Reihenfolge, so wie es die realen Aufrufstellen in doc/TRO_Katalog/scl_templates/*.scl zeigen. Was aus dem Layout kommt, steht drin. Was aus der E-Planung kommen muss, steht als Platzhalter mit `TODO(E-Planung)` da - bewusst als offensichtliche Luecke und nicht als erfundener Wert, denn eine plausibel aussehende falsche Sensoradresse ist schlimmer als eine leere. Alle Platzhalter sind ueber genau diesen Marker auffindbar. Ableitbar und daher gefuellt: - Reihenfolge der REGIONs (Materialfluss), Instanzliste, FB-Typ je TRO - Anzahl der Separator-/Weichen-/Scanner-Parameterbloecke je TRO - nTo1Destinations einer Weiche = Anzahl ihrer Abgaenge - die Zielliste je Weichenabgang (Erreichbarkeit im TRO-Graphen) - Querverweis auf die Bauteile im Layout (Separatornummern, Koordinate) Aufruf ueber bin/scl_skeleton.bat bzw. bin/scl_skeleton.sh. """ from __future__ import annotations import argparse import json import sys from collections import deque from datetime import datetime from pathlib import Path from material_flow import env_dir TODO = "TODO(E-Planung)" # Zeitwerte aus den Referenz-Aufrufstellen. Sie sind plausible Startwerte, KEINE # fuer diese Anlage ermittelten - deshalb ueberall mit Pruefhinweis. DEFAULT_SEP_TIMES = ( ("tDelayToNextItem", '"TimeForSepWaiting_Short"', None), ("tTrailingTime", "t#600ms", "pruefen"), ("tHandlingTime", "t#200ms", "pruefen"), ("tJam", '"TimeForSepJam"', None), ) def load_tros(path: Path) -> dict: return json.loads(path.read_text(encoding="utf-8")) def flow_order(tros: list[dict], start_id: str = "") -> tuple[list[dict], str]: """TROs in Fliessrichtung sortieren. Startpunkt ist ein TRO ohne Vorgaenger - also die Stelle, an der Material in die Anlage kommt. Solange Be-/Entladestation fehlen, ist der Fluss ein geschlossener Kreis und es gibt keinen solchen TRO; dann muss der Kreis irgendwo aufgetrennt werden. Das ist eine willkuerliche Wahl, deshalb wird sie zurueckgegeben und im Kopf der SCL vermerkt (und laesst sich mit --start ueberschreiben). Nicht erreichte TROs haengt die Funktion hinten an, damit nie einer verlorengeht. """ nach_id = {t["id"]: t for t in tros} ohne_vorgaenger = sorted(t["id"] for t in tros if not t.get("predecessors")) if start_id and start_id in nach_id: start, grund = [start_id], "per --start gesetzt" elif ohne_vorgaenger: start, grund = ohne_vorgaenger, "TRO ohne Vorgaenger (Materialeingang)" else: start = [min(nach_id)] grund = ("Kreislauf ohne Eingang - willkuerlich aufgetrennt; sobald " "Be-/Entladestation im Export stehen, ergibt sich der Anfang") gesehen: list[str] = [] queue = deque(start) while queue: tro_id = queue.popleft() if tro_id in gesehen: continue gesehen.append(tro_id) for nxt in nach_id.get(tro_id, {}).get("successors", []): if nxt not in gesehen: queue.append(nxt) for tro_id in sorted(nach_id): if tro_id not in gesehen: gesehen.append(tro_id) return [nach_id[i] for i in gesehen], grund def reachable_via(start: str, first: str, tros: dict[str, dict]) -> list[str]: """Welche TROs sind ueber diesen Abgang erreichbar? Das ist die Zielliste eines Weichenabgangs: von `first` aus weiterlaufen, ohne ueber die Weiche selbst zurueckzugehen. Genau die Information, die FC_Direction braucht, um zu entscheiden - und sie steht im Graphen. """ gesehen: set[str] = set() queue = deque([first]) while queue: node = queue.popleft() if node in gesehen or node == start: continue gesehen.add(node) queue.extend(tros.get(node, {}).get("successors", [])) return sorted(gesehen) def region_title(tro: dict) -> str: seps = ", ".join(tro.get("separators", [])) teile = f" Sep {seps}" if seps else "" return f"{tro['id']} ({tro['type']}){teile}" def emit_separators(add, tro: dict, anzahl: int, hat_scanner: bool) -> None: seps = tro.get("separators", []) for i in range(1, anzahl + 1): pfad = f"#{tro['id']}.stInSeparator{i}" layout = seps[i - 1] if i <= len(seps) else "?" add(f" // Separator {i} = Bauteil {layout} aus dem Layout") add(f' {pfad}.Settings.nSeparatorNo := "cSep_____"; // {TODO}') for name, wert, hinweis in DEFAULT_SEP_TIMES: kommentar = f" // {TODO}: {hinweis}" if hinweis else "" add(f" {pfad}.Settings.{name} := {wert};{kommentar}") add(f" {pfad}.Settings.xPluggedJam := TRUE;") add(f" {pfad}.Settings.xPluggedPart := FALSE;") add(f" {pfad}.Settings.xPluggedScanner := " f"{'TRUE' if hat_scanner else 'FALSE'};") add(f' {pfad}.stSenInSep := "DB_Inputs".Sensors["cInBG____"]; // {TODO}') add(f" {pfad}.Settings.tSenFree := T#1500ms; // {TODO}: pruefen") add(f" {pfad}.Settings.tSenWait := T#4000ms; // {TODO}: pruefen") add("") def emit_switches(add, tro: dict, anzahl: int, alle: dict[str, dict]) -> None: abgaenge = tro.get("successors", []) for j in range(1, anzahl + 1): pfad = f"#{tro['id']}.stInSwitch{j}" add(f' {pfad}.Settings.nSwitchNo := "cSwi_____"; // {TODO}') add(f" {pfad}.Settings.xPluggedExSen3 := FALSE;") add(f" {pfad}.Settings.xPluggedExSen4 := FALSE;") if anzahl == 1: # Eine Weiche bedient alle Abgaenge - direkt ableitbar. add(f" {pfad}.Settings.nTo1Destinations := {len(abgaenge)};" f" // Abgaenge laut Topologie") else: # Bei mehreren Weichen ist es eine Planungsentscheidung, welche # Weiche welche Abgaenge bedient - die Summe ist bekannt, die # Aufteilung nicht. Deshalb hier kein geratener Wert. add(f" {pfad}.Settings.nTo1Destinations := _;" f" // {TODO}: dieser TRO hat {len(abgaenge)} " f"Abgaenge auf {anzahl} Weichen - Aufteilung festlegen") add("") if abgaenge: add(" // Zielliste je Abgang (aus dem TRO-Graphen berechnet) -") add(" // Grundlage fuer die Zuordnung Ziel -> Richtung in FC_Direction:") for abgang in abgaenge: ziele = reachable_via(tro["id"], abgang, alle) add(f" // ueber {abgang}: {', '.join(ziele) if ziele else '-'}") add("") def emit_vario(add, tro: dict) -> None: pfad = f"#{tro['id']}.stInVario" add(f" {pfad}.Settings.xContinousEmptying := FALSE;") add(f" {pfad}.Settings.tWaitForJog := T#300ms; // {TODO}: pruefen") add(f" {pfad}.Settings.tConvEmpty := T#45000ms; // {TODO}: pruefen") add(f" {pfad}.Settings.nCountCarrier := 1; // {TODO}: pruefen") add(f" {pfad}.Settings.tWaitForStop := t#3s;") add(f" {pfad}.Settings.tCarrierInPosDelay := t#1200ms;") add(f" {pfad}.Settings.tCarrierInPosEdgeDelay := t#100ms;") add(f" {pfad}.Settings.xJamSenDelayActive := TRUE;") add(f" {pfad}.Settings.tJamSenDelay := T#2s;") add(f' {pfad}.stSenCarrInPos := "DB_Inputs".Sensors["cInBG____"]; // {TODO}') add(f' {pfad}.stSenFinger := "DB_Inputs".Sensors["cInBG____"]; // {TODO}') add(f' {pfad}.stSenLastPos := "DB_Inputs".Sensors["cInBG____"]; // {TODO}') add(f' {pfad}.stSenJam := "DB_Inputs".Sensors["cInBG____"]; // {TODO}') add("") def emit_barcode(add, tro: dict, anzahl: int) -> None: for i in range(1, anzahl + 1): suffix = "" if anzahl == 1 else str(i) pfad = f"#{tro['id']}.stInBarcodeReader{suffix}" add(f" {pfad}.Settings.xActivateCheckASCI := TRUE;") add(f" {pfad}.Settings.tTimeout := T#3000ms;") add(f' // {TODO}: Cognex-Interface verdrahten ' f'(#arCognexInterface["cScanner____"])') add("") def emit_call(add, tro: dict, items: dict[str, int]) -> None: """Aufrufstelle schreiben. Die Zeilen werden als (Code, Kommentar) gefuehrt und das Komma erst beim Zusammensetzen gesetzt - sonst landet es hinter dem "//" und ist Teil des Kommentars, womit die Parameterliste nicht mehr uebersetzt. """ tro_id = tro["id"] zeilen: list[tuple[str, str]] = [ ('nInMainTroNo := "cMainTro____"', TODO), ('xInSftyOk := "DB_InterfaceSafety".stFromSafety.EStopMain.Q', TODO), ("xInAllRdyToStart := TRUE", ""), (f'xInRelease := "DB_TRO_TEST".{tro_id}', ""), ] if items.get("VarioDrive"): zeilen.append(('xInMotorProtection := "FC____"', TODO)) zeilen.append(("xInAutomatic := #t_automatik", "")) zeilen.append(("stInOutMachineState := #stInOutControlUnitCabinet", "")) zeilen.append(("stInOutCognexInterface := #stCognexInterfaceDummy", "")) # Staubereiche: je ankommender/abgehender Verbindung einer. Welche # DB-Struktur dahintersteht, ist eine Planungsentscheidung - hier steht # nur, WIE VIELE es sein muessen und zu welchem Nachbarn sie gehoeren. for n, vor in enumerate(tro.get("predecessors", []), start=1): zeilen.append((f'arInOutJamEntr{n} := "DB_JamArea____".stJam____.arCarrier', f"{TODO} - Staubereich von {vor}")) zeilen.append((f'stInOutJamEntr{n} := "DB_JamArea____".stJam____.stData', "")) for n, nach in enumerate(tro.get("successors", []), start=2): zeilen.append((f'arInOutJamExit{n} := "DB_JamArea____".stJam____.arCarrier', f"{TODO} - Staubereich nach {nach}")) zeilen.append((f'stInOutJamExit{n} := "DB_JamArea____".stJam____.stData', "")) for i in range(1, items.get("Separator", 1) + 1): zeilen.append((f'xOutStopper{i} => "MB____"', TODO)) if items.get("VarioDrive"): zeilen.append(('xOutVarioMotor => "MA____"', TODO)) for j in range(1, items.get("Switch", 0) + 1): zeilen.append((f'xOutSw{j}ExTo3 => "MB____"', TODO)) zeilen.append((f'stInOutHMI := "DB_Interface_HMI".stTRO.{tro_id}', "")) zeilen.append(('nOutStateLast => "DB_GLOB_TroState".arTroState["cMainTro____"]', TODO)) kopf = f" #{tro_id}(" einzug = " " * len(kopf) for index, (code, kommentar) in enumerate(zeilen): trenner = ");" if index == len(zeilen) - 1 else "," prefix = kopf if index == 0 else einzug text = f"{prefix}{code}{trenner}" add(f"{text} // {kommentar}" if kommentar else text) def emit_direction(add, tro: dict, anzahl: int) -> None: """Weichenlogik - je Weiche ein FC_Direction-Aufruf.""" add("") add(" REGION Direction") for j in range(1, anzahl + 1): ziel = (f"#{tro['id']}.nInDirection" if anzahl == 1 else f"#{tro['id']}.nInDirection{j}") add(f' "FC_Direction"(nInSwitchNo := "cSwi_____", // {TODO}') add(f" nOutDirection => {ziel},") add(f' InOutCarrier := "DB_JamArea____".stJam____' f".arCarrier[1]);") if j < anzahl: add("") add(" END_REGION") def render(daten: dict, quelle: str, start_id: str = "") -> str: tros = daten["tros"] alle = {t["id"]: t for t in tros} geordnet, startgrund = flow_order(tros, start_id) out: list[str] = [] add = out.append add("// ============================================================") add("// FB_Main - GERUEST, erzeugt von lib/scl_skeleton.py") add(f"// Quelle : {quelle}") add(f"// Erzeugt: {datetime.now().strftime('%Y-%m-%d %H:%M')}") add("//") add("// Das ist KEIN lauffaehiges Programm, sondern ein Geruest zum") add("// Ausfuellen. Struktur, Reihenfolge und Bauteilzahl stammen aus dem") add("// Anlagenlayout und der vor Ort aufgenommenen Topologie.") add("//") add(f"// Jede Stelle, die aus der E-Planung kommen muss, ist mit") add(f"// '{TODO}' markiert - danach suchen, dann ist nichts vergessen.") add("// Offen sind durchgaengig: Sensor-/Aktoradressen (cInBG/MB/MA),") add("// Separator- und Weichennummern, MainTro-Nummern, die Staubereiche") add("// (DB_JamArea...) und die Zeiten, soweit als 'pruefen' vermerkt.") add("//") add("// Reihenfolge der REGIONs = Materialfluss.") add(f"// Anfang bei {geordnet[0]['id']}: {startgrund}.") add("// ============================================================") add("") add('FUNCTION_BLOCK "FB_Main"') add("{ S7_Optimized_Access := 'TRUE' }") add("VERSION : 0.1") add(" VAR ") for tro in geordnet: fb = tro.get("fbType") or "FB_UNBEKANNT" kommentar = f" // {tro['type']}" if tro.get("separators"): kommentar += f", Sep {', '.join(tro['separators'])}" add(f' {tro["id"]} : "{fb}";{kommentar}') add(" END_VAR") add("") add("BEGIN") for tro in geordnet: items = tro.get("items", {}) n_sep = items.get("Separator", 1) n_swi = items.get("Switch", 0) n_scan = items.get("Scanner", 0) add("") add(f"REGION {region_title(tro)}") pos = tro.get("position") if pos: add(f" // Anlagenkoordinate x/y: {pos['x']:.0f} / {pos['y']:.0f} mm") add(f" // Vorgaenger: {', '.join(tro.get('predecessors', [])) or '-'}") add(f" // Nachfolger: {', '.join(tro.get('successors', [])) or '-'}") if tro.get("confidence"): add(f" // Einstufung: {tro['confidence']}") add("") emit_separators(add, tro, n_sep, hat_scanner=bool(n_scan)) if n_swi: emit_switches(add, tro, n_swi, alle) if items.get("VarioDrive"): emit_vario(add, tro) if n_scan: emit_barcode(add, tro, n_scan) if items.get("StorageLine"): add(f" // {TODO}: Linienspeicher mit {items['StorageLine']} Linien - " f"Parametrierung je Linie") add("") add(" #%s.stInPriorityManager.xReleaseOk := TRUE;" % tro["id"]) add("") emit_call(add, tro, items) if n_swi: emit_direction(add, tro, n_swi) add("END_REGION") add("") add("END_FUNCTION_BLOCK") add("") return "\n".join(out) def main(argv: list[str] | None = None) -> int: args = parse_args(argv) results = env_dir("SKEL_RESULTS", "results") json_file = Path(args.json) if not json_file.is_absolute() and json_file.parent == Path("."): json_file = results / args.json if not json_file.is_file(): print(f"FEHLER: JSON nicht gefunden: {json_file}", file=sys.stderr) print(" Zuerst bin/tro_extract.bat laufen lassen.", file=sys.stderr) return 1 try: daten = load_tros(json_file) except (OSError, ValueError) as exc: print(f"FEHLER: {json_file} nicht lesbar: {exc}", file=sys.stderr) return 1 if not daten.get("tros"): print(f"FEHLER: {json_file} enthaelt keine TROs.", file=sys.stderr) return 1 text = render(daten, json_file.name, args.start) out_file = results / (args.out or f"{json_file.stem.replace('_tro', '')}_FB_Main.scl") out_file.write_text(text, encoding="utf-8") offen = text.count(TODO) geordnet, startgrund = flow_order(daten["tros"], args.start) print("") print("================================================================") print("FB_MAIN-GERUEST") print("================================================================") print(f"Quelle = {json_file}") print(f"TROs = {len(daten['tros'])}") print(f"Reihenfolge = {' -> '.join(t['id'] for t in geordnet)}") print(f"Anfang = {geordnet[0]['id']} ({startgrund})") print(f"Offene Stellen = {offen} (Marker '{TODO}')") print(f"SCL = {out_file}") print("================================================================") print("") return 0 def parse_args(argv: list[str] | None = None) -> argparse.Namespace: parser = argparse.ArgumentParser( prog="scl_skeleton", description="Erzeugt aus der extrahierten TRO-Liste ein FB_Main-Geruest " "als SCL zum Import in TIA Portal.", formatter_class=argparse.RawDescriptionHelpFormatter, epilog="Beispiele:\n" " scl_skeleton.bat\n" " scl_skeleton.bat --json export_tro.json --out FB_Main.scl\n" "\n" "Exit-Codes:\n" " 0 ok\n" " 1 Eingabe- oder Aufruffehler\n", ) parser.add_argument( "--json", default="export_tro.json", metavar="NAME", help="TRO-JSON aus tro_extract.py in %%SKEL_RESULTS%% " "(oder ein vollstaendiger Pfad). Standard: %(default)s", ) parser.add_argument( "--out", metavar="NAME", help="Name der SCL-Datei in %%SKEL_RESULTS%%. Standard: _FB_Main.scl", ) parser.add_argument( "--start", default="", metavar="TRO", help="TRO, bei dem die Reihenfolge beginnt. Nur noetig, solange der " "Fluss ein geschlossener Kreis ohne Be-/Entladestation ist.", ) return parser.parse_args(argv) if __name__ == "__main__": sys.exit(main())