# -*- coding: utf-8 -*- """Extrahiert den TRO-Topologie-Graphen aller 5 UH-Steuerungen aus FB_Main.scl und erzeugt eine Graphviz-DOT-Datei. Prinzip: Jeder TRO-Baustein bekommt JamArea-Slots als Parameter (arInOutJamEntr*, arInOutJamExit*, arInOutJamBypass, arInOutJamFinger). Zwei TROs sind verbunden, wenn der Exit-Slot des einen dieselbe JamArea referenziert wie der Entry-Slot des anderen. Aufruf: python bin/generate_tro_graph.py [ausgabe.dot] Ohne Argument wird nach doc/TRO_Graph_UH01-UH05.dot geschrieben. Zum Rendern als SVG anschliessend: dot -Tsvg doc/TRO_Graph_UH01-UH05.dot -o doc/TRO_Graph_UH01-UH05.svg """ import re import sys from collections import defaultdict from pathlib import Path REPO = Path(__file__).resolve().parent.parent UHS = ["UH01", "UH02", "UH03", "UH04", "UH05"] decl_re = re.compile(r'^\s*(\w+)\s*(?:\{[^}]*\})?\s*:\s*"([\w.]+)"\s*;') jamparam_re = re.compile( r'arInOut(Jam\w+|StorageLines|CarrierStopper|CarrierExit)' r'\s*:=\s*"([^"]+)"\.(\w+)') call_start_re = re.compile(r'#(\w+)\s*\(') region_re = re.compile(r'^\s*REGION\b\s*(.*)$') endregion_re = re.compile(r'^\s*END_REGION') def parse_uh(uh): path = REPO / f"=A01+{uh}-KF00" / "Programmbausteine" / "100_Main" / "FB_Main.scl" text = path.read_text(encoding="utf-8-sig") text = re.sub(r'\(\*.*?\*\)', '', text, flags=re.S) # Blockkommentare lines = [re.sub(r'//.*$', '', ln) for ln in text.splitlines()] # 1) Instanz-Deklarationen -> FB-Typ types = {} for ln in lines: m = decl_re.match(ln) if m: types[m.group(1)] = m.group(2) # 2) Aufrufe finden inkl. REGION-Kontext nodes = {} # inst -> dict(type, region, jams=[(role, jamref)]) region_stack = [] i = 0 while i < len(lines): ln = lines[i] if endregion_re.match(ln): if region_stack: region_stack.pop() i += 1 continue m = region_re.match(ln) if m and "END_REGION" not in ln.split("REGION")[0]: region_stack.append(m.group(1).strip()) i += 1 continue m = call_start_re.search(ln) if m and types.get(m.group(1)): inst = m.group(1) # Parameterblock bis zur schliessenden Klammer einsammeln buf = ln[m.end() - 1:] depth = 0 j = i done = False while j < len(lines) and not done: seg = lines[j] if j > i else buf for ch in seg: if ch == "(": depth += 1 elif ch == ")": depth -= 1 if depth == 0: done = True break j += 1 block = "\n".join([buf] + lines[i + 1:j]) jams = [] for jm in jamparam_re.finditer(block): role, db, member = jm.group(1), jm.group(2), jm.group(3) jams.append((role, f"{db}.{member}")) if jams: # aussagekraeftigste REGION waehlen (generische wie "CALL", # "TRO 101 CALL", "Enable und RequestRun" ueberspringen) region = "" for r in reversed(region_stack): if not re.match(r'^(.*\bCALL\b.*|Enable und RequestRun)$', r, re.I): region = r break node = nodes.setdefault(inst, { "type": types[inst], "region": region, "jams": []}) for j2 in jams: if j2 not in node["jams"]: node["jams"].append(j2) i = j continue i += 1 return nodes def role_dir(role): if role.startswith("JamEntr") or role in ("StorageLines", "CarrierStopper"): return "in" if (role.startswith(("JamExit", "JamBypass", "JamFinger")) or role == "CarrierExit"): return "out" return "?" def short_jam(jamref): # DB_JamArea0101.stJam0101_1 -> JA0101_1 m = re.match(r"DB_JamArea(\w+)\.stJam\w+_(\w+)", jamref) if m: return f"JA{m.group(1)}_{m.group(2)}" return jamref.replace("DB_", "").replace(".arLine", "") def main(): all_nodes = {} report = [] for uh in UHS: nodes = parse_uh(uh) all_nodes[uh] = nodes report.append(f"{uh}: {len(nodes)} TRO-Knoten") # Kanten je UH ueber gemeinsame JamAreas edges = defaultdict(list) # uh -> [(from, to, jam, tag)] externals = defaultdict(list) # uh -> [(node, dir, jam, role)] for uh, nodes in all_nodes.items(): jam_users = defaultdict(lambda: {"in": [], "out": []}) for inst, nd in nodes.items(): for role, jam in nd["jams"]: d = role_dir(role) if d in ("in", "out"): jam_users[jam][d].append((inst, role)) for jam, users in sorted(jam_users.items()): for (src, srole) in users["out"]: tag = {"JamExit3": "Dir2", "JamExit4": "Dir3", "JamBypass": "Bypass", "JamFinger": "Finger" }.get(srole, "") if users["in"]: for (dst, drole) in users["in"]: if dst != src: edges[uh].append((src, dst, jam, tag)) else: externals[uh].append((src, "out", jam, srole)) for (dst, drole) in users["in"]: if not users["out"]: externals[uh].append((dst, "in", jam, drole)) for uh in UHS: report.append(f"{uh}: {len(edges[uh])} Kanten, " f"{len(externals[uh])} offene Enden") print("\n".join(report)) # Knoten ohne jede Kante? for uh, nodes in all_nodes.items(): connected = set() for (s, d, _, _) in edges[uh]: connected.add(s) connected.add(d) for (n, _, _, _) in externals[uh]: connected.add(n) iso = set(nodes) - connected if iso: print(f"{uh} isoliert: {sorted(iso)}") write_dot(all_nodes, edges, externals) COLORS = { "FB_ILS_MTRO_1Sep": ("#d4e8ff", "box"), "FB_ILS_MTRO_1Sep_SSCC": ("#ffd4d4", "box"), "FB_ILS_MTRO_1Sep1Swi": ("#ffe4a0", "diamond"), "FB_ILS_MTRO_1Sep2Swi": ("#ffc060", "diamond"), "FB_ILS_MTRO_2Sep1Swi": ("#ffe4a0", "diamond"), "FB_ILS_MTRO_Vario": ("#e8d4f0", "parallelogram"), "FB_ILS_MTRO_Vario_workStation": ("#c0f0f0", "parallelogram"), "FB_EmptyCarrBuffer": ("#d4f0d4", "box3d"), } def node_style(fbtype): for k in sorted(COLORS, key=len, reverse=True): if fbtype.startswith(k): return COLORS[k] return ("#f0f0f0", "box") def esc(s): return s.replace("\\", "\\\\").replace('"', '\\"') HEADER = '''\ // ============================================================================ // TRO-Topologie-Graph der Gesamtanlage HundM_Fortna (UH01 - UH05) // ============================================================================ // // Format: Graphviz DOT (https://graphviz.org) // Rendern z.B. mit: dot -Tsvg TRO_Graph_UH01-UH05.dot -o graph.svg // oder online: https://dreampuf.github.io/GraphvizOnline // // Quelle: Automatisch extrahiert aus den FB_Main.scl der 5 Steuerungen // (=A01+UH01-KF00 ... =A01+UH05-KF00, Ordner 100_Main). // Erzeugt von bin/generate_tro_graph.py. // // Methode: Jeder TRO-Baustein (FB_ILS_MTRO_*, FB_EmptyCarrBuffer, // FB_LoadingBoom_INBOUND) referenziert JamAreas als Parameter: // arInOutJamEntr* / arInOutStorageLines / arInOutCarrierStopper = Eingang // arInOutJamExit* / arInOutJamBypass / arInOutJamFinger / // arInOutCarrierExit = Ausgang // Zwei TROs sind verbunden, wenn der Ausgangs-Slot des einen // dieselbe JamArea referenziert wie der Eingangs-Slot des anderen. // Auskommentierter Code wurde ignoriert. // // Knoten: Ein Knoten je TRO-Instanz, Label: Name, [FB-Typ], REGION-Kommentar // aus FB_Main. Farbe/Form nach FB-Typ: // hellblau/Box = 1Sep (Standard-Separator) // rot/Box = 1Sep_SSCC (SSCC-Scanner + WCS-Telegramm) // gelb/Raute = 1Sep1Swi, 1Sep2Swi, 2Sep1Swi (Weichen) // violett/Parallelogr.= Vario (Kettenfoerderer) // tuerkis/Parallelogr.= Vario_workStation (manueller Arbeitsplatz) // gruen/3D-Box = EmptyCarrBuffer (Leertablar-Puffer) // grau = Sonstige (z.B. LoadingBoom, PinStore) // // Kanten: Label = JamArea-Slot (JAxxxx_y = "DB_JamAreaxxxx".stJamxxxx_y). // Zusatz "Dir2"/"Dir3" = Weichen-Ausgang (arInOutJamExit3/4), // Richtungsentscheidung kommt aus FC_Direction. // Gestrichelt "Bypass" = Bypass-Spur (PinStore), // gestrichelt "Finger" = Finger-Uebergabe eines Vario-Foerderers. // // EXTERN: Graue Kanten zu EXTERN-Knoten = JamAreas, die nur auf einer // Seite belegt sind (Anlagengrenze, Uebergabe an WES/WCS, AMR // oder andere SPS). CPU-CPU-Uebergaben zwischen den Steuerungen // sind an den Labels "IO_Device_UHxx_TO_UHyy..." erkennbar // (PROFINET-Direktkopplung, kein gemeinsamer JamArea-DB). // // Hinweis: Kanten existieren nur innerhalb einer Steuerung, da JamArea-DBs // nicht SPS-uebergreifend geteilt werden. Der physische Fluss // zwischen den UHs laeuft ueber die EXTERN-Uebergaben. // ''' def write_dot(all_nodes, edges, externals): body = [] B = body.append B(HEADER.rstrip()) total_n = sum(len(v) for v in all_nodes.values()) total_e = sum(len(set(v)) for v in edges.values()) B(f"// Umfang: {total_n} TRO-Knoten, {total_e} Verbindungen " f"(Stand der Extraktion siehe Git-Historie)") for uh in UHS: B(f"// {uh}: {len(all_nodes[uh])} TROs, " f"{len(set(edges[uh]))} Verbindungen") B("") B("digraph TRO_Topologie {") B(' rankdir = "LR";') B(' fontname = "Arial";') B(' node [fontname="Arial", fontsize=10];') B(' edge [fontname="Arial", fontsize=8];') B(' label = "TRO-Topologie HundM_Fortna (UH01-UH05)";') B(' labelloc = "t";') B("") for uh in UHS: nodes = all_nodes[uh] B(f' subgraph "cluster_{uh}" {{') B(f' label = "{uh} (=A01+{uh}-KF00)";') B(' style = "rounded";') B(' color = "#888888";') for inst in sorted(nodes): nd = nodes[inst] color, shape = node_style(nd["type"]) t = nd["type"].replace("FB_ILS_MTRO_", "").replace("FB_", "") reg = esc(nd["region"]) B(f' "{uh}.{inst}" [label="{inst}\\n[{t}]\\n{reg}", ' f'shape={shape}, style=filled, fillcolor="{color}"];') # Externe Enden als ein Sammel-Knoten je Richtung ext = externals[uh] if any(d == "in" for (_, d, _, _) in ext): B(f' "{uh}.EXTERN_IN" [label="EXTERN\\n(Zulauf)", ' f'shape=ellipse, style="filled,dashed", fillcolor="#f8f8f8"];') if any(d == "out" for (_, d, _, _) in ext): B(f' "{uh}.EXTERN_OUT" [label="EXTERN\\n(Ablauf)", ' f'shape=ellipse, style="filled,dashed", fillcolor="#f8f8f8"];') seen = set() for (s, d, jam, tag) in edges[uh]: key = (s, d, jam, tag) if key in seen: continue seen.add(key) lbl = short_jam(jam) + (f" {tag}" if tag else "") style = ', style=dashed' if tag in ("Bypass", "Finger") else "" B(f' "{uh}.{s}" -> "{uh}.{d}" [label="{lbl}"{style}];') seen = set() for (n, d, jam, role) in ext: key = (n, d, jam) if key in seen: continue seen.add(key) if d == "in": B(f' "{uh}.EXTERN_IN" -> "{uh}.{n}" ' f'[label="{short_jam(jam)}", color="#999999"];') else: B(f' "{uh}.{n}" -> "{uh}.EXTERN_OUT" ' f'[label="{short_jam(jam)}", color="#999999"];') B(" }") B("") B("}") dest = Path(sys.argv[1]) if len(sys.argv) > 1 else REPO / "doc" / "TRO_Graph_UH01-UH05.dot" dest.write_text("\n".join(body) + "\n", encoding="utf-8", newline="\n") print("DOT geschrieben:", dest) if __name__ == "__main__": main()