432 lines
18 KiB
Python
432 lines
18 KiB
Python
# -*- 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: <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())
|