Schoenenberger_Systeme_GmbH/sps_skel
8
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
39b445a7c423de47927f71f022ff292c3596a26e
sps_skel/lib/scl_context.py
T
s.ayadi 39b445a7c4 scl_skeleton: Anlagen-JSON als Eingabe (Stufe 4, Namen) 
--json erkennt <csv>_anlage.json am Schema. scl_context.HwSource setzt
Namen aus bindings (nur "zugeordnet") und conveyors ein: Separator-,
Lagerlinien-, Weichen- und Vario-Antriebsnamen, #fbCarousel<NNNN> am
Kreisel, Foerderernummer im DB_JamArea-Platzhalter (<k> bleibt offen).
Eingesetzte Namen werden gegen <projekt>_hw.json geprueft (--hw,
--strict bricht ab). --eplan bleibt fuer die TRO-JSON; Templates und
Golden-Files unveraendert.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-16 10:53:52 +02:00

810 lines
30 KiB
Python
Raw Blame History

# -*- coding: utf-8 -*-
"""
scl_context.py - Render-Kontext fuer das FB_Main-Geruest.
Schritt 2 aus doc/SCL_Templates.md: alles, was scl_skeleton aus der TRO-JSON
*ableitet*, steht hier - Reihenfolge, Bauteilbloecke, Staubereiche,
Weichenziele, Namen. Die Ausgabe (heute die emit_*-Funktionen, spaeter die
Templates in lib/scl_templates/) liest nur noch diesen Kontext und rechnet
nichts mehr selbst.
Schichten:
model.py TroDocument (die TRO-JSON)
scl_context.py TroDocument -> SclFile <- hier
scl_skeleton.py SclFile -> SCL-Text
Namen (Separator-, Weichen-, MainTro-Nummer, Adressen ...) sind ein
:class:`SclName`: Wert plus Merker, ob er noch aus der E-Planung kommen muss.
Ohne Hardwareplanung sind alle Platzhalter ``todo=True``. Quelle der echten
Namen ist :class:`HwSource`:
- die Anlagen-JSON aus merge_hw_layout (``anlage``, Stufe 4 in
doc/Konzept_Layout_Hardware_SCL.md): Zuordnungen je TRO-ID + Slot + Rolle
(Separator, Lagerlinie, Weiche, Antrieb) und die Foerderernummer je
Layout-Foerderer (Kreisel-Freigabe, Staubereichs-DB);
- oder, auslaufend, die Zuordnung aus lib/e_planning.py (``eplan``):
nur Separator-Sensor und Stopperventil, je Layout-ID.
Ein eingesetzter Name loescht ``todo``; alles andere bleibt offen. Die
Namensregeln aus Schritt 4 (cfg/scl_naming.ini) arbeiten spaeter genauso:
Wert ersetzen, Merker loeschen.
"""
from __future__ import annotations
import re
from collections import deque
from datetime import datetime
from typing import TYPE_CHECKING
from pydantic import BaseModel, ConfigDict, Field
from model import AnlageDocument, ConveyorBinding, JamArea, TroBase, TroDocument
from tro_catalog import TroItem
if TYPE_CHECKING:
from e_planning import EplanDocument
__all__ = [
"TODO",
"SclName",
"SclSeparator",
"SclSwitch",
"SclSwitchExit",
"SclScanner",
"SclJam",
"SclSscc",
"SclWorkStation",
"SclStoreLine",
"SclStore",
"SclTro",
"SclFile",
"HwSource",
"build_context",
"flow_order",
"reachable_via",
]
TODO = "TODO(E-Planung)"
_CONFIG = ConfigDict(extra="forbid")
class SclName(BaseModel):
"""Ein Name/eine Adresse in der SCL und ob er noch offen ist."""
model_config = _CONFIG
value: str
# True: Platzhalter, muss aus der E-Planung kommen -> TODO-Kommentar
todo: bool = True
def _offen(platzhalter: str) -> SclName:
return SclName(value=platzhalter, todo=True)
def _bekannt(wert: str) -> SclName:
return SclName(value=wert, todo=False)
# Rolle in e_planning.ini -> wie der Name in der SCL geschrieben wird
# (Sensor als Konstante der DB_Inputs-Sensorliste, Ventil als Variable).
# Die Anlagen-JSON bringt die Konstante aus dem TIA-Import mit; die Vorlage
# gilt nur, wenn sie fehlt (alte eplan-Zuordnung).
_SENSOR = "cIn{bmk}"
_VENTIL = "{bmk}"
# Antriebs-Slot (model._DRIVE_ROLES) -> Parameter des Vario-Aufrufs
_VARIO_SENSOR_ROLES = {
"stSenCarrInPos": "senCarrInPos",
"stSenFinger": "senFinger",
"stSenLastPos": "senLastPos",
"stSenJam": "senJam",
}
class HwSource:
"""Namen aus der Hardwareplanung - eine Abfrage fuer beide Quellen.
``anlage`` (merge_hw_layout): Schluessel TRO-ID + Slot + Rolle, nur
Zuordnungen mit Status ``zugeordnet`` zaehlen; ``offen``/``widerspruch``
bleiben ``TODO(E-Planung)``. ``eplan`` (e_planning.py, auslaufend):
Schluessel Layout-ID, kennt nur Separator-Sensor und -Stopper. Ohne beide
liefert jede Abfrage None.
"""
def __init__(self, anlage: AnlageDocument | None = None,
eplan: EplanDocument | None = None) -> None:
self.anlage = anlage
self.eplan = eplan
self._kinds = ({p.layout_id: p.kind for p in anlage.layout.parts}
if anlage is not None else {})
def device(self, tro_id: str, slot: str, role: str,
layout: str = "") -> tuple[str, str] | None:
"""(BMK, Konstante) einer Slot-Rolle; Konstante "" wenn unbekannt."""
if self.anlage is not None:
b = self.anlage.binding(tro_id, slot, role)
if b is None or b.status != "zugeordnet" or not b.bmk:
return None
return b.bmk, b.constant
if self.eplan is not None and layout and layout != "?":
a = self.eplan.for_layout(layout).get(role)
return (a.bmk, "") if a is not None else None
return None
def sensor(self, tro_id: str, slot: str, role: str, layout: str = "",
offen: str = "cInBG____") -> SclName:
"""Sensor als Konstante der DB_Inputs-Sensorliste (cInBG0532)."""
geraet = self.device(tro_id, slot, role, layout)
if geraet is None:
return _offen(offen)
bmk, konstante = geraet
return _bekannt(konstante or _SENSOR.format(bmk=bmk))
def output(self, tro_id: str, slot: str, role: str, layout: str = "",
offen: str = "MB____", fmt: str = _VENTIL) -> SclName:
"""Ausgang/Variable ueber den BMK (MB0533, "FC0202" ...)."""
geraet = self.device(tro_id, slot, role, layout)
return _bekannt(fmt.format(bmk=geraet[0])) if geraet else _offen(offen)
def conveyor(self, host: str) -> ConveyorBinding | None:
"""Foerderer der Hardwareplanung (K0102) am Layout-Foerderer ("0010-R")."""
if self.anlage is None or not host:
return None
return self.anlage.conveyor_for(host)
def is_carousel(self, host: str) -> bool:
return self._kinds.get(host.split("-", 1)[0]) == "Kreisel"
@property
def source(self) -> str:
"""Fuer den Dateikopf; "" ohne Hardwareplanung."""
if self.anlage is not None:
return (f"{self.anlage.hw_file or '-'} + {self.anlage.layout_file or '-'} "
f"(Anlagen-JSON vom {self.anlage.created})")
if self.eplan is not None:
return f"{self.eplan.eplan_file} (Zuordnung vom {self.eplan.created})"
return ""
def _foerderer_nr(conveyor: ConveyorBinding) -> str:
"""K0102 / #106 -> "0102" bzw. "0106" (Schreibweise der Referenz-DBs)."""
ziffern = re.sub(r"\D", "", conveyor.eplan_name)
return ziffern.zfill(4) if ziffern else ""
class SclSeparator(BaseModel):
"""Ein stInSeparator<n>-Block."""
model_config = _CONFIG
n: int
# Separatornummer aus dem Layout ("?" wenn der Typ mehr Separatoren hat
# als im Layout gefunden wurden)
layout: str
no: SclName
sensor: SclName
stopper: SclName
# Teilesensor (stSenPart) - nur in Templates, deren Referenz ihn hat (1Sep2Swi)
part_sensor: SclName
class SclSwitchExit(BaseModel):
"""Ein Abgang eines Weichen-TRO und die darueber erreichbaren TROs."""
model_config = _CONFIG
via: str
targets: list[str]
class SclSwitch(BaseModel):
"""Ein stInSwitch<n>-Block plus sein FC_Direction-Aufruf."""
model_config = _CONFIG
n: int
no: SclName
valve: SclName
# None: mehrere Weichen teilen sich die Abgaenge - Aufteilung ist eine
# Planungsentscheidung und wird nicht geraten
to1_destinations: int | None
# Ziel von nOutDirection, z. B. "#TRO07.nInDirection" bzw. "...nInDirection2"
direction_target: str
# Staubereich, dessen erster Traeger in FC_Direction ausgewertet wird
jam_db: SclName
class SclScanner(BaseModel):
"""Ein stInBarcodeReader-Block ("" als Suffix, wenn es nur einen gibt).
Die Cognex-Verdrahtung (nur in den Templates): ``interface`` ist der
Schluessel in #arCognexInterface (z. B. "cScannerBX2100"), ``tag`` der
Praefix der Hardware-Symbole ("bx2100-TriggerReady" ...), ``hw_results``
das Ergebnis-Symbol der Kamera.
"""
model_config = _CONFIG
n: int
suffix: str
interface: SclName
tag: SclName
hw_results: SclName
class SclJam(BaseModel):
"""Ein Staubereich am Ein- oder Ausgang (JamEntr<n> / JamExit<n>).
Ein Eingang ist ein Staubereich des TRO selbst (TroDocument.jamAreas), ein
Ausgang der Eingangs-Staubereich des Nachfolgers.
"""
model_config = _CONFIG
n: int
# Kennung aus der TRO-JSON (J<Separator>); "" am Ausgang zu einer Station
jam_id: str = ""
# Eingang: die Vorgaenger, Ausgang: der Nachfolger - als Text fuer den
# Kommentar, SYSn als Station markiert
neighbour: str
db: SclName
# Stausensor des Ausgangs (stInSenJam<n>); None an Eingaengen
sensor: SclName | None = None
# Traeger-Array und Staudaten des Bereichs - im Normalfall <db>.arCarrier
# bzw. <db>.stData; an Uebergaben auch eigene Variablen (#arCarrierHandover...)
carriers: SclName
data: SclName
@property
def title(self) -> str:
"""Fuer den Kommentar: 'Staubereich J0003' (ohne Kennung nur 'Staubereich')."""
return f"Staubereich {self.jam_id}" if self.jam_id else "Staubereich"
class SclSscc(BaseModel):
"""SSCC-Scanner eines 1Sep_SSCC: zweite Kamera, Endmessung, HMI-Anzeige."""
model_config = _CONFIG
# interface ist hier die Struktur selbst (z. B. "#stCognexIntfTro105")
scanner: SclScanner
end_sensor: SclName
# Schluessel der HMI-Anzeige, z. B. "SSCC1107"
hmi: SclName
class SclWorkStation(BaseModel):
"""Arbeitsplatz eines Vario_workStation."""
model_config = _CONFIG
station_sensor: SclName
station_release: SclName
loaded_error_sensor: SclName
class SclStoreLine(BaseModel):
"""Eine Speicherlinie eines PinStore_Auto (arInSep[n])."""
model_config = _CONFIG
n: int
# Separatornummer der Linie aus dem Layout
layout: str
jam_length: SclName
sensor: SclName
jam_sensor: SclName
stopper: SclName
class SclStore(BaseModel):
"""Alles fuer ein automatisches Pin-Lager (PinStore_Auto)."""
model_config = _CONFIG
encoder: SclName
dest: SclName
conv_stations: SclName
stations_max: SclName
threshold: SclName
reference_sensor: SclName
station: SclName
quit_local: SclName
test_target: SclName
run_in: SclName
run_out: SclName
db: SclName
removal: SclName
conv_db: SclName
# Einschleus-Separator (Pin) - aus dem Layout, wenn tro_flow einen
# steuernden Separator vor den Linien gefunden hat, sonst "?"
pin_layout: str
pin_sensor: SclName
pin_stopper: SclName
# Eine Linie je Speicherbahn im Layout (nicht die 19 aus dem Katalog)
lines: list[SclStoreLine]
@property
def title(self) -> str:
"""Fuer den Kommentar: 'Staubereich J0003' (ohne Kennung nur 'Staubereich')."""
return f"Staubereich {self.jam_id}" if self.jam_id else "Staubereich"
class SclTro(BaseModel):
"""Alles, was eine REGION fuer einen TRO braucht."""
model_config = _CONFIG
id: str
# Nummernteil der ID, wie geschrieben ("TRO01" -> "01"); "" ohne Ziffern
nr: str
type: str
fb: str
region_title: str
layout_separators: list[str]
position: tuple[float, float] | None
predecessors: list[str]
successors: list[str]
confidence: str
separators: list[SclSeparator]
switches: list[SclSwitch]
switch_exits: list[SclSwitchExit]
has_vario: bool
# Vario-Sensoren in Aufrufreihenfolge (Parametername -> Sensor); leer ohne Vario
vario_sensors: dict[str, SclName]
scanners: list[SclScanner]
storage_lines: int
jam_entries: list[SclJam]
jam_exits: list[SclJam]
main_tro: SclName
safety: SclName
release: str
motor_protection: SclName
vario_motor: SclName
machine_state: SclName
# Nur in den Templates (Referenz-Aufrufstellen), nicht im generischen Aufbau:
# UDT des Prioritaetsmanagers ("udt_TRO107") und ob Richtung 2 Vorrang hat
priority_udt: SclName
prio_dir2: SclName
# Kreisel-Freigabe (xInCarouselRun1). Ob ein TRO am Kreisel liegt, steht
# nicht in der TRO-JSON - deshalb immer offen (doc/SCL_Templates.md §6).
carousel_run: SclName
# -- nur fuer die Templates (Referenz-Aufrufstellen, doc/SCL_Templates.md §6)
# Einziger Staubereich am Ein-/Ausgang fuer FBs ohne Nummerierung
# (Vario: JamEntr/JamFinger, PinStore: JamEntr/JamExit): der erste
# Vorgaenger bzw. Nachfolger, sonst ein offener Platzhalter
jam_in: SclJam
jam_out: SclJam
error_quit: SclName
automatic: SclName
# Vario: stInSettings.xIsBorner (Bedeutung mit SPS klaeren)
is_border: SclName
# Station fuer FC_CARR_GLOB_CORRECTION (1Sep_SSCC, PinStore_Auto)
correction_station: SclName
sscc: SclSscc | None = None
work_station: SclWorkStation | None = None
store: SclStore | None = None
@property
def has_scanner(self) -> bool:
return bool(self.scanners)
class SclFile(BaseModel):
"""Kontext der ganzen Datei: Kopf und die TROs in Fliessrichtung."""
model_config = _CONFIG
source: str
created: str
start_reason: str
tros: list[SclTro] = Field(default_factory=list)
# Zuordnung zur E-Planung, aus der Namen eingesetzt wurden ("" = keine)
eplan_source: str = ""
# Anlagen-JSON (merge_hw_layout), aus der Namen eingesetzt wurden ("" = keine)
hw_source: str = ""
# ---------------------------------------------------------------------------
# Graph
# ---------------------------------------------------------------------------
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 _nummer(tro_id: str) -> str:
treffer = re.search(r"(\d+)$", tro_id)
return treffer.group(1) if treffer else ""
# ---------------------------------------------------------------------------
# Kontext bauen
# ---------------------------------------------------------------------------
def _scanner(n: int, suffix: str, interface: str = "cScanner____") -> SclScanner:
return SclScanner(
n=n, suffix=suffix, interface=_offen(interface), tag=_offen("bx____"),
hw_results=_offen("BX____~Ergebnisse_\u2013_64_Bytes_1"),
)
def _jam(n: int, neighbour: str, jam_db: str, sensor: SclName | None = None,
jam_id: str = "") -> SclJam:
return SclJam(
n=n, jam_id=jam_id, neighbour=neighbour, db=_offen(jam_db), sensor=sensor,
carriers=_offen(f"{jam_db}.arCarrier"), data=_offen(f"{jam_db}.stData"),
)
def _store(tro: TroBase, hw: HwSource) -> SclStore:
"""PinStore_Auto: eine Linie je Speicherbahn (Host) des Layouts.
tro_flow legt die Separatoren als [steuernde ..., Linien ...] ab; die
Linien sind die letzten len(hosts). Fehlen Hosts (fremde JSON), zaehlt
jeder Separator als Linie. Sensor und Stopper jeder Linie (Slot
``line<n>``) und des Pin-Separators (Slot ``separator1``, falls tro_flow
einen steuernden Separator gefunden hat) kommen aus der Hardwareplanung
wie bei den Separatorbloecken.
"""
hosts = list(getattr(tro, "hosts", []) or [])
seps = list(tro.separators)
anzahl = len(hosts) if 0 < len(hosts) <= len(seps) else len(seps)
steuer, linien = seps[:len(seps) - anzahl], seps[len(seps) - anzahl:]
if steuer:
pin_sensor, pin_stopper = _sep_namen(tro.tro_id, "separator1", steuer[0], hw)
else:
pin_sensor, pin_stopper = _offen("cInBG____"), _offen("MB____")
zeilen = []
for n, sep in enumerate(linien, start=1):
slot = f"line{n}"
sensor, stopper = _sep_namen(tro.tro_id, slot, sep, hw)
zeilen.append(SclStoreLine(n=n, layout=sep,
jam_length=_offen('"cStorage____ExitJam"'),
sensor=sensor,
jam_sensor=hw.sensor(tro.tro_id, slot, "jamSensor"),
stopper=stopper))
return SclStore(
encoder=_offen("#fbEncoder____"),
dest=_offen("cDest____"),
conv_stations=_offen("cHooksSorterMaxPin____"),
stations_max=_offen("cStationsMax____"),
threshold=_offen("cNoPositionsBlock____"),
reference_sensor=_offen("cInBG____"),
station=_offen("____"),
quit_local=_offen('"SF____"'),
test_target=_offen('"IBN".nTestTarget____'),
run_in=_offen("#fbSortConv____.xOutConveyorFullSpeed"),
run_out=_offen("#fbExitConv____.xOutConveyorFullSpeed"),
db=_offen('"DB_Storage____"'),
removal=_offen('"DB_HandshakeStorage".Str_StorageRemoval.'
'arrudt_StorageRemovalHandshakeTROsLvl1[____]'),
conv_db=_offen('"DB_ConveyorStations".stConveyorStations____'),
pin_layout=", ".join(steuer) or "?",
pin_sensor=pin_sensor,
pin_stopper=pin_stopper,
lines=zeilen,
)
def _sep_namen(tro_id: str, slot: str, layout: str,
hw: HwSource) -> tuple[SclName, SclName]:
"""Sensor und Stopperventil eines Separators aus der Hardwareplanung.
Gilt fuer die Separatorbloecke (stInSeparator<n>, Slot ``separator<n>``)
und genauso fuer die Linien eines PinStore_Auto (Slot ``line<n>``) - beide
sind Separatoren im Layout. Ohne Zuordnung bleiben beide offen.
"""
return (hw.sensor(tro_id, slot, "sensor", layout),
hw.output(tro_id, slot, "stopper", layout))
def _separator(i: int, tro_id: str, layout: str, hw: HwSource) -> SclSeparator:
slot = f"separator{i}"
sensor, stopper = _sep_namen(tro_id, slot, layout, hw)
return SclSeparator(
n=i,
layout=layout,
no=_offen("cSep_____"),
sensor=sensor,
stopper=stopper,
# Teilesensor: nur die Anlagen-JSON kennt ihn (Rolle partSensor)
part_sensor=hw.sensor(tro_id, slot, "partSensor"),
)
_JAM_DB = '"DB_JamArea____".stJam____'
def _jam_db(area: JamArea | None, hw: HwSource) -> str:
"""Platzhalter des Staubereichs-DB, mit Foerderernummer soweit bekannt.
Schema der Referenzen: "DB_JamArea<NNNN>".stJam<NNNN>_<k>. <NNNN> ist der
Foerderer, auf dem der Staubereich liegt (Anlagen-JSON, conveyors[]);
<k> - die laufende Nummer entlang des Flusses auf diesem Foerderer - ist
noch nicht festgelegt (doc/Konzept_Layout_Hardware_SCL.md "Staubereiche").
Der Name bleibt deshalb immer offen.
"""
conveyor = hw.conveyor(area.host) if area is not None else None
nr = _foerderer_nr(conveyor) if conveyor is not None else ""
return f'"DB_JamArea{nr}".stJam{nr}_____' if nr else _JAM_DB
def _entries(tro: TroBase, eigene: list[JamArea],
alle: dict[str, TroBase], hw: HwSource) -> list[SclJam]:
"""Ein JamEntr je eigenem Staubereich - nicht je Vorgaenger: mehrere
Vorgaenger fahren in denselben Staubereich ein (Referenzen UH01-UH05)."""
eingaenge = []
for area in eigene:
von = ", ".join(_nachbar(v, alle) for v in area.predecessors) or "-"
if area.note:
von = f"{von} ({area.note})"
eingaenge.append(_jam(area.entry, von, _jam_db(area, hw), jam_id=area.jam_id))
return eingaenge
def _exit_area(tro_id: str, nach: str,
bereiche: dict[str, list[JamArea]]) -> JamArea | None:
"""Staubereich, in den tro_id beim Nachfolger einfaehrt."""
ziel = bereiche.get(nach, [])
passend = [a for a in ziel if tro_id in a.predecessors] or ziel
return passend[0] if passend else None
def _drive(tro: TroBase, hw: HwSource) -> tuple[SclName, SclName]:
"""Motor und Motorschutz eines Vario-Antriebs.
Erst der Antriebs-Slot (Rollen motor/motorProtection); fehlt er, der
Foerderer der Hardwareplanung am Host des TRO - der Vario-Antrieb ist
dessen Motor (MA/FC-Paar aus hw_import).
"""
motor = hw.output(tro.tro_id, "drive", "motor", offen="MA____")
schutz = hw.output(tro.tro_id, "drive", "motorProtection",
offen='"FC____"', fmt='"{bmk}"')
hosts = list(getattr(tro, "hosts", []) or [])
conveyor = hw.conveyor(hosts[0]) if hosts else None
if conveyor is not None:
if motor.todo and conveyor.motor:
motor = _bekannt(conveyor.motor)
if schutz.todo and conveyor.protection:
schutz = _bekannt(f'"{conveyor.protection}"')
return motor, schutz
def _carousel_run(tro: TroBase, hw: HwSource) -> SclName:
"""xInCarouselRun1: Freigabe des Kreisels, auf dem der TRO sitzt.
Bekannt nur, wenn der Host ein Kreisel ist und die Anlagen-JSON dessen
Foerderernummer kennt (Referenz: #fbCarousel0102.xOutConveyorFullSpeed).
"""
hosts = list(getattr(tro, "hosts", []) or [])
host = hosts[0] if hosts else ""
conveyor = hw.conveyor(host) if hw.is_carousel(host) else None
nr = _foerderer_nr(conveyor) if conveyor is not None else ""
if not nr:
return _offen("#fbCarousel____.xOutConveyorFullSpeed")
return _bekannt(f"#fbCarousel{nr}.xOutConveyorFullSpeed")
def _tro_context(tro: TroBase, alle: dict[str, TroBase],
hw: HwSource | None = None,
bereiche: dict[str, list[JamArea]] | None = None) -> SclTro:
items = tro.items
n_sep = items.get(TroItem.SEPARATOR, 1)
n_swi = items.get(TroItem.SWITCH, 0)
n_scan = items.get(TroItem.SCANNER, 0)
tro_id = tro.tro_id
hw = hw or HwSource()
bereiche = bereiche or {}
separators = [
_separator(i, tro_id,
tro.separators[i - 1] if i <= len(tro.separators) else "?", hw)
for i in range(1, n_sep + 1)
]
switches = [
SclSwitch(
n=j,
no=_offen("cSwi_____"),
valve=hw.output(tro_id, f"switch{j}", "valve"),
# Eine Weiche bedient alle Abgaenge - direkt ableitbar.
to1_destinations=len(tro.successors) if n_swi == 1 else None,
direction_target=(f"#{tro_id}.nInDirection" if n_swi == 1
else f"#{tro_id}.nInDirection{j}"),
jam_db=_offen(_JAM_DB),
)
for j in range(1, n_swi + 1)
]
has_vario = bool(items.get(TroItem.VARIO_DRIVE))
vario_sensors = {
name: hw.sensor(tro_id, "drive", rolle)
for name, rolle in _VARIO_SENSOR_ROLES.items()
} if has_vario else {}
# Nur ein Vario hat einen eigenen Antrieb; sonst bleiben die (ungenutzten)
# Platzhalter offen, statt den Motor des Kreisels zu tragen.
vario_motor, motor_protection = _drive(tro, hw) if has_vario else (
_offen("MA____"), _offen('"FC____"'))
switch_exits = [
SclSwitchExit(via=abgang, targets=reachable_via(tro_id, abgang, alle))
for abgang in tro.successors
] if n_swi else []
# Staubereiche aus der TRO-JSON (jamAreas): Eingaenge sind die eigenen
# (ein JamEntr je eigenem Staubereich, nicht je Vorgaenger), Ausgaenge die
# Eingaenge der Nachfolger. Der DB-Name braucht Foerderernummer (aus der
# Anlagen-JSON) und laufende Nummer (nicht festgelegt) und bleibt offen.
jam_entries = _entries(tro, bereiche.get(tro_id, []), alle, hw)
jam_exits = []
for n, nach in enumerate(tro.successors, start=2):
area = _exit_area(tro_id, nach, bereiche)
jam_exits.append(_jam(n, _nachbar(nach, alle), _jam_db(area, hw),
sensor=_offen("cInBG____"),
jam_id=area.jam_id if area is not None else ""))
return SclTro(
id=tro_id,
nr=_nummer(tro_id),
type=tro.type_name,
fb=tro.fb_block or "FB_UNBEKANNT",
region_title=_region_title(tro),
layout_separators=list(tro.separators),
position=(tro.x, tro.y) if tro.position else None,
predecessors=list(tro.predecessors),
successors=list(tro.successors),
confidence=tro.confidence,
separators=separators,
switches=switches,
switch_exits=switch_exits,
has_vario=has_vario,
vario_sensors=vario_sensors,
scanners=[
_scanner(i, "" if n_scan == 1 else str(i))
for i in range(1, n_scan + 1)
],
storage_lines=items.get(TroItem.STORAGE_LINE, 0),
jam_entries=jam_entries,
jam_exits=jam_exits,
main_tro=_offen("cMainTro____"),
safety=_offen('"DB_InterfaceSafety".stFromSafety.EStopMain.Q'),
release=f'"DB_TRO_TEST".{tro_id}',
motor_protection=motor_protection,
vario_motor=vario_motor,
# Schaltschrank ist anlagenspezifisch ("...OF1" in der Referenz)
machine_state=_offen("#stInOutControlUnitCabinet"),
priority_udt=_offen("udt_____"),
# Vorschlag aus der Referenz TRO107, Prioritaet bleibt Planungsentscheidung
prio_dir2=_offen("TRUE"),
carousel_run=_carousel_run(tro, hw),
# dasselbe Objekt wie der erste Ein-/Ausgang: ein Name, den die
# Namensregeln dort setzen, gilt damit auch hier
jam_in=jam_entries[0] if jam_entries else _jam(1, "-", _JAM_DB),
jam_out=jam_exits[0] if jam_exits else _jam(2, "-", _JAM_DB),
error_quit=_offen('"SF____"'),
automatic=_offen("#t_automatik"),
is_border=_offen("TRUE"),
correction_station=_offen("0"),
sscc=(SclSscc(scanner=_scanner(1, "SSCC", "#stCognexIntf____"),
end_sensor=_offen("cInBG____"), hmi=_offen("SSCC____"))
if items.get(TroItem.SSCC) else None),
work_station=(SclWorkStation(station_sensor=_offen("cInBG____"),
station_release=_offen('#"rTrig_Workstation ____".CLK'),
loaded_error_sensor=_offen("cInBG____"))
if items.get(TroItem.WORK_STATION) else None),
store=_store(tro, hw) if items.get(TroItem.STORAGE_LINE) else None,
)
def build_context(document: TroDocument, source: str, start_id: str = "",
created: str | None = None,
eplan: EplanDocument | None = None,
anlage: AnlageDocument | None = None) -> SclFile:
"""TroDocument -> SclFile (TROs in Fliessrichtung).
anlage: Anlagen-JSON (merge_hw_layout) - ihr Layout ist ``document``,
ihre Zuordnungen liefern die Namen. eplan: alte Zuordnung Layout ->
E-Planung (e_planning.py), nur ohne ``anlage``. Ohne beide bleiben alle
Namen offen und die Ausgabe ist die bisherige.
"""
if anlage is not None and eplan is not None:
raise ValueError("anlage und eplan schliessen sich aus")
hw = HwSource(anlage=anlage, eplan=eplan)
alle = {t.tro_id: t for t in document.tros}
bereiche: dict[str, list[JamArea]] = {}
for area in document.jam_areas_or_derived():
bereiche.setdefault(area.tro, []).append(area)
geordnet, startgrund = flow_order(document.tros, start_id)
return SclFile(
source=source,
created=created or datetime.now().strftime("%Y-%m-%d %H:%M"),
start_reason=startgrund,
tros=[_tro_context(t, alle, hw, bereiche) for t in geordnet],
eplan_source=hw.source if eplan is not None else "",
hw_source=hw.source if anlage is not None else "",
)
Reference in New Issue View Git Blame Copy Permalink