326 lines
13 KiB
Python
326 lines
13 KiB
Python
# -*- 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()
|