mistangl/sps_skel
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
ba58c988519bdb1922a713a4ae66f40d595240d7
sps_skel/lib/scl_skeleton.py
T
Samer AyadiandClaude Opus 5.5 97a0db6707 model.py: gemeinsames TRO-Modell; Hauptweg CSV -> tro_flow --json -> scl_skeleton 
- lib/model.py (neu, pydantic): TroBase -> DerivedTro, Connection, OpenPoint,
  Analysis (aus tro_flow verschoben), TroDocument = TRO-JSON (Schema
  sps_skel-tro-v2). items/fbType werden immer ueber den Typ aus tro_catalog
  gelesen, nie gespeichert; abweichende Werte in einer JSON sind ein Fehler.
  JSON-Schluessel wie bisher (id, type, fbType ...), extra="forbid".
- tro_flow: Tro/Analysis aus model.py; neuer Schalter --json schreibt
  <name>_tro.json (TroDocument)
- scl_skeleton: liest die JSON ueber TroDocument statt per Dict-Zugriff,
  Bauteile ueber TroItem-Konstanten; ungueltige JSON -> Exit 1 ohne SCL,
  alte tro_extract-JSON (v1) mit Hinweis auf tro_flow --json
- tro_annotate: nur noch Ansicht; derselbe analyse_tros-Aufruf wie tro_flow
  (inkl. sepliste-Gegenprobe), damit Zeichnung und JSON uebereinstimmen
- tro_extract: geparkt (unveraendert)
- tests/test_model.py + conftest.py, pytest in requirements.txt aktiviert
- Doku: Python_Scripts.md (§6b model.py), Commandlines.md, CLAUDE.md,
  Aufrufkommentare in bin/ (scl_skeleton.bat zeigte tro_flow-Aufrufe)

Geprueft: alle tro_flow/tro_annotate-Ausgaben (8 CSVs) bis auf Zeitstempel
identisch zum vorherigen Stand; scl_skeleton alt/neu auf derselben JSON
byte-identisch (export, mubea, HundM05); 10 Tests gruen.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-09 12:42:46 +02:00

483 lines
20 KiB
Python
Raw Blame History

# -*- 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 -> <name>_tro.json (--json)
scl_skeleton.py JSON -> FB_Main.scl <- hier
Die TRO-JSON wird ueber das gemeinsame Modell lib/model.py (TroDocument)
eingelesen und dabei geprueft: ein falscher oder umbenannter Schluessel bricht
mit einer Fehlermeldung ab, statt still einen Standardwert zu nehmen. Bauteile
(items) und FB-Typ kommen aus tro_catalog, ueber den TRO-Typ.
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 pydantic import ValidationError
from material_flow import env_dir
from model import SCHEMA, TroBase, TroDocument
from tro_catalog import TroItem
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_document(path: Path) -> TroDocument:
"""TRO-JSON einlesen und gegen das Modell pruefen."""
return TroDocument.from_json(path.read_text(encoding="utf-8"))
def flow_order(tros: list[TroBase], start_id: str = "") -> tuple[list[TroBase], 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.tro_id: t for t in tros}
ohne_vorgaenger = sorted(t.tro_id for t in tros if not t.predecessors)
# Materialeingang ueber eine Station (SYSn von tro_flow.py): der TRO, den
# sie speist, ist der natuerliche Anfang - auch wenn er zusaetzlich einen
# TRO-Vorgaenger hat (Kreisel-Umlauf).
eingaenge = sorted({
t.tro_id for t in tros
for vor in t.predecessors if vor not in nach_id
})
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)"
elif eingaenge:
stationen = sorted({
vor for i in eingaenge for vor in nach_id[i].predecessors
if vor not in nach_id
})
start = eingaenge
grund = f"Materialeingang ueber {', '.join(stationen)}"
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[tro_id].successors if tro_id in nach_id else []:
# SYSn (Materialausgang) ist kein TRO - keine REGION, keine Instanz.
if nxt in nach_id and 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, TroBase]) -> 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)
if node in tros:
queue.extend(tros[node].successors)
return sorted(gesehen)
def nachbar(node: str, alle: dict[str, TroBase]) -> str:
"""Nachbar fuer Kommentare; SYSn als Station kenntlich machen."""
return node if node in alle else f"{node} (Station, kein TRO)"
def region_title(tro: TroBase) -> str:
seps = ", ".join(tro.separators)
teile = f" Sep {seps}" if seps else ""
return f"{tro.tro_id} ({tro.type_name}){teile}"
def emit_separators(add, tro: TroBase, anzahl: int, hat_scanner: bool) -> None:
seps = tro.separators
for i in range(1, anzahl + 1):
pfad = f"#{tro.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: TroBase, anzahl: int, alle: dict[str, TroBase]) -> None:
abgaenge = tro.successors
for j in range(1, anzahl + 1):
pfad = f"#{tro.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.tro_id, abgang, alle)
add(f" // ueber {abgang}: {', '.join(ziele) if ziele else '-'}")
add("")
def emit_vario(add, tro: TroBase) -> None:
pfad = f"#{tro.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: TroBase, anzahl: int) -> None:
for i in range(1, anzahl + 1):
suffix = "" if anzahl == 1 else str(i)
pfad = f"#{tro.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: TroBase, items: dict[str, int], alle: dict[str, TroBase]) -> 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.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(TroItem.VARIO_DRIVE):
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.predecessors, start=1):
zeilen.append((f'arInOutJamEntr{n} := "DB_JamArea____".stJam____.arCarrier',
f"{TODO} - Staubereich von {nachbar(vor, alle)}"))
zeilen.append((f'stInOutJamEntr{n} := "DB_JamArea____".stJam____.stData', ""))
for n, nach in enumerate(tro.successors, start=2):
zeilen.append((f'arInOutJamExit{n} := "DB_JamArea____".stJam____.arCarrier',
f"{TODO} - Staubereich nach {nachbar(nach, alle)}"))
zeilen.append((f'stInOutJamExit{n} := "DB_JamArea____".stJam____.stData', ""))
for i in range(1, items.get(TroItem.SEPARATOR, 1) + 1):
zeilen.append((f'xOutStopper{i} => "MB____"', TODO))
if items.get(TroItem.VARIO_DRIVE):
zeilen.append(('xOutVarioMotor => "MA____"', TODO))
for j in range(1, items.get(TroItem.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: TroBase, anzahl: int) -> None:
"""Weichenlogik - je Weiche ein FC_Direction-Aufruf."""
add("")
add(" REGION Direction")
for j in range(1, anzahl + 1):
ziel = (f"#{tro.tro_id}.nInDirection" if anzahl == 1
else f"#{tro.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(document: TroDocument, quelle: str, start_id: str = "") -> str:
tros = document.tros
alle = {t.tro_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].tro_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.fb_block or "FB_UNBEKANNT"
kommentar = f" // {tro.type_name}"
if tro.separators:
kommentar += f", Sep {', '.join(tro.separators)}"
add(f' {tro.tro_id} : "{fb}";{kommentar}')
add(" END_VAR")
add("")
add("BEGIN")
for tro in geordnet:
items = tro.items
n_sep = items.get(TroItem.SEPARATOR, 1)
n_swi = items.get(TroItem.SWITCH, 0)
n_scan = items.get(TroItem.SCANNER, 0)
add("")
add(f"REGION {region_title(tro)}")
if tro.position:
add(f" // Anlagenkoordinate x/y: {tro.x:.0f} / {tro.y:.0f} mm")
add(f" // Vorgaenger: {', '.join(tro.predecessors) or '-'}")
add(f" // Nachfolger: {', '.join(tro.successors) or '-'}")
if tro.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(TroItem.VARIO_DRIVE):
emit_vario(add, tro)
if n_scan:
emit_barcode(add, tro, n_scan)
if items.get(TroItem.STORAGE_LINE):
add(f" // {TODO}: Linienspeicher mit {items[TroItem.STORAGE_LINE]} Linien - "
f"Parametrierung je Linie")
add("")
add(" #%s.stInPriorityManager.xReleaseOk := TRUE;" % tro.tro_id)
add("")
emit_call(add, tro, items, alle)
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_flow.bat --file <name>.csv --json laufen lassen.",
file=sys.stderr)
return 1
try:
document = load_document(json_file)
except OSError as exc:
print(f"FEHLER: {json_file} nicht lesbar: {exc}", file=sys.stderr)
return 1
except ValidationError as exc:
schema = _schema_of(json_file)
if schema and schema != SCHEMA:
# Anderes Format (z. B. die alte tro_extract-JSON) - die einzelnen
# Feldfehler helfen dann nicht weiter.
print(f"FEHLER: {json_file} hat das Schema '{schema}', erwartet wird "
f"'{SCHEMA}' ({exc.error_count()} Abweichungen).", file=sys.stderr)
print(" Mit bin/tro_flow.bat --file <name>.csv --json neu erzeugen.",
file=sys.stderr)
return 1
print(f"FEHLER: {json_file} passt nicht zum TRO-Modell (lib/model.py):",
file=sys.stderr)
print(exc, file=sys.stderr)
return 1
if not document.tros:
print(f"FEHLER: {json_file} enthaelt keine TROs.", file=sys.stderr)
return 1
text = render(document, 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(document.tros, args.start)
print("")
print("================================================================")
print("FB_MAIN-GERUEST")
print("================================================================")
print(f"Quelle = {json_file}")
print(f"TROs = {len(document.tros)}")
print(f"Reihenfolge = {' -> '.join(t.tro_id for t in geordnet)}")
print(f"Anfang = {geordnet[0].tro_id} ({startgrund})")
print(f"Offene Stellen = {offen} (Marker '{TODO}')")
print(f"SCL = {out_file}")
print("================================================================")
print("")
return 0
def _schema_of(path: Path) -> str:
"""Schema-Kennung einer JSON-Datei, nur fuer die Fehlermeldung ("" wenn unlesbar)."""
try:
return str(json.loads(path.read_text(encoding="utf-8")).get("schema", ""))
except (OSError, ValueError, AttributeError):
return ""
def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
parser = argparse.ArgumentParser(
prog="scl_skeleton",
description="Erzeugt aus der TRO-JSON (tro_flow --json) 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_flow.py --json 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: <Quelle>_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())
Reference in New Issue View Git Blame Copy Permalink