Schoenenberger_Systeme_GmbH/sps_skel
8
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
6c8c235287e82d3da363429e32f5868020423970
sps_skel/lib/model.py
T
s.ayadi 6c8c235287 tro_annotate --read-tros: TRO-Handkorrekturen aus BricsCAD, Staubereiche vor dem TRO 
- Marker tragen KEY (ID der ID-Datei) und SLOTS (Bauteil je Slot), die
  Zeichnung die Bauteiltabelle SPS_SKEL_PARTS (XRECORD) fuer "Im Plan
  waehlen" im Dialog TRO_EDIT (dxfmakros)
- --read-tros schreibt geaenderte, neue und geloeschte Marker als
  Abschnitte edit/edit_add nach cfg/tro_overrides.ini; tro_flow.apply_edits
  wendet nur diese Arten an (Bauteile mit Lage, exportfest)
- --emit-lisp schreibt dazu *TRO-SLOTS* und *TRO-SLOT-KINDS*
- model: Lagerlinien zaehlen je TRO (Katalog nur Vorgabe), Mubea-Speicher
  jetzt 20 Linien (Golden-Datei)
- Staubereich als stehendes graues Rechteck direkt vor seinem TRO-Marker
  (jam_abstand statt jam_versatz); Marker 150 mm, Text 120 mm

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-26 14:35:04 +01:00

768 lines
30 KiB
Python
Raw Blame History

# -*- coding: utf-8 -*-
"""
model.py - Gemeinsames Datenmodell der TRO-Instanzen und der TRO-JSON.
Schichten (wer darf wen importieren):
tro_catalog.py Typ-Ebene: was ein TRO-*Typ* ist (Bauteile, FB, Farbe, Symbol)
model.py Instanz-Ebene: ein konkreter TRO, die ganze TRO-JSON <- hier
importiert nur tro_catalog
tro_flow.py Ableitung: CSV-Graph -> Analysis -> TroDocument
scl_skeleton.py Ausgabe: TroDocument -> FB_Main.scl
tro_annotate.py Ausgabe: Analysis -> annotierte Zeichnung (nur Ansicht)
Klassen:
TroBase was jeder TRO hat, egal woher er stammt: ID, Typ, Separatoren,
Vorgaenger/Nachfolger, Position. FB-Baustein und Bauteile
werden aus dem Katalog gelesen, nie gespeichert.
DerivedTro TroBase + was tro_flow bei der Ableitung zusaetzlich weiss
(Hosts, Scanner, Begruendung, Positionsquelle ...)
Connection eine gerichtete Verbindung zwischen zwei TROs/Stationen
TroCandidate eine im Review gemeldete Stelle, an der ein TRO erwartet
wird, die aber (noch) keiner ist
Slot eine Stelle eines TRO, an die Hardware gehoert (Separator,
Weiche, Scanner, Antrieb) mit den Rollen, die die
Hardwareplanung dort fuellen muss
JamArea ein Staubereich: Eingang genau eines TRO (je Separator
laut Katalog), mit allen Vorgaengern, die dort einfahren
LayoutPart ein Bauteil des Layouts mit Lage und Huellbox - das Ziel
der raeumlichen Zuordnung in merge_hw_layout (die TRO-JSON traegt
es, damit merge_hw_layout das CSV nicht lesen muss)
Analysis Arbeitsergebnis von tro_flow.analyse_tros
TroDocument die TRO-JSON (<csv>_tro.json): von tro_flow --json
geschrieben, von scl_skeleton gelesen und geprueft
Binding eine Zuordnung Hardware -> TRO-Slot-Rolle (merge_hw_layout)
ConveyorBinding ein Foerderer der Hardwareplanung (K0102) an einem
Layout-Foerderer, mit Antrieb, Motorschutz, Geber
AnlageDocument die Anlagen-JSON (<csv>_anlage.json): Layout unveraendert
plus bindings[] - von merge_hw_layout geschrieben
In der JSON stehen die Schluessel wie bisher in der TRO-JSON (`id`, `type`,
`fbType`, `candidates`, ...); im Python-Code heissen die Felder weiter
`tro_id`/`type_name` (Aliase, `populate_by_name`). Gleiche Daten heissen in
allen Klassen gleich (auch in tro_overrides.py): `type_name`, `separators`,
`hosts`, `position`, `reason`, `confidence`.
"""
from __future__ import annotations
from datetime import datetime
from typing import TYPE_CHECKING, Any
from pydantic import (
BaseModel,
ConfigDict,
Field,
computed_field,
field_serializer,
field_validator,
model_validator,
)
from tro_catalog import TroDefinition, TroItem, get_tro
__all__ = [
"SCHEMA",
"TroBase",
"Slot",
"JamArea",
"LayoutPart",
"DerivedTro",
"Connection",
"TroCandidate",
"Analysis",
"TroDocument",
"Registration",
"Binding",
"ConveyorBinding",
"AnlageDocument",
"ANLAGE_SCHEMA",
"BINDING_STATUS",
"derive_slots",
"derive_jam_areas",
"storage_lines",
"SLOT_PART_KINDS",
"slot_kind",
"slot_names",
"has_lines",
]
# Kennung der TRO-JSON. v1 war die Ausgabe von tro_extract (aus der Zeichnung),
# v2 die von tro_flow --json (aus dem CSV, ueber dieses Modell), v3 wie v2 mit
# `candidates` statt `openPoints`, v4 wie v3 mit `slots` je TRO und `jamAreas`.
# Eine v3-Datei liest sich weiter (die neuen Felder fehlen dann und werden beim
# Rendern abgeleitet, siehe TroDocument.jam_areas_or_derived). Seit 2026-10-15
# traegt v4 optional `parts` (Lage der Layout-Bauteile fuer merge_hw_layout); eine
# v4-Datei ohne `parts` liest sich weiter, nur merge_hw_layout braucht sie.
# Seit 2026-10-16 traegt jeder TRO optional `switches` (Weichen-Bauteile, fuellt
# die switch<n>-Slots); fehlt es, bleiben die Slots wie bisher ohne layoutId.
SCHEMA = "sps_skel-tro-v4"
# Kennung der Anlagen-JSON (<csv>_anlage.json, merge_hw_layout)
ANLAGE_SCHEMA = "sps_skel-anlage-v1"
# Zustand einer Zuordnung (doc/Konzept_Layout_Hardware_SCL.md, Umbau 3):
# zugeordnet ein BMK steht fest
# offen Pflichtrolle eines Slots ohne BMK -> TODO(E-Planung)
# widerspruch BMK steht fest, passt aber nicht (fehlt im TIA-Import,
# doppelt vergeben, Handkorrektur auf unbekanntes BMK)
# ohne_slot E-Bauteil einem Layout-Bauteil zugeordnet, das (noch) an
# keinem TRO-Slot haengt (z. B. Ausschleuselemente)
BINDING_STATUS = ("zugeordnet", "offen", "widerspruch", "ohne_slot")
# Gemeinsame Einstellungen: Pruefung bei Erzeugung und Zuweisung, unbekannte
# Schluessel sind ein Fehler (ein umbenannter JSON-Schluessel faellt auf statt
# still auf einen Standardwert zurueckzufallen), Aliase und Feldnamen gelten
# beide.
_CONFIG = ConfigDict(
validate_assignment=True,
extra="forbid",
populate_by_name=True,
)
class _Positioned(BaseModel):
"""
Anlagenkoordinate (mm) als Feld ``position``, in der JSON {"x", "y", "z"}.
Das Feld selbst deklariert jede Unterklasse, damit es dort in der
JSON-Reihenfolge an seiner Stelle steht.
"""
if TYPE_CHECKING:
position: Any
@field_validator("position", mode="before", check_fields=False)
@classmethod
def _position_from_json(cls, value: Any) -> Any:
if isinstance(value, dict):
return (value["x"], value["y"], value.get("z", 0.0))
return value
@field_serializer("position", when_used="json-unless-none", check_fields=False)
def _position_to_json(self, value: tuple[float, float, float]) -> dict[str, float]:
return {"x": round(value[0], 2), "y": round(value[1], 2), "z": round(value[2], 2)}
@property
def x(self) -> float | None:
return self.position[0] if self.position else None
@property
def y(self) -> float | None:
return self.position[1] if self.position else None
@property
def z(self) -> float | None:
return self.position[2] if self.position else None
def describe_position(self) -> str:
"""Koordinate als Text, z. B. '1030, 4700'."""
if not self.position:
return "-"
return f"{self.position[0]:.0f}, {self.position[1]:.0f}"
class TroBase(_Positioned):
"""
Kern eines TRO (Instanz in einer Anlage), unabhaengig von seiner Quelle.
Gegenstueck zu :class:`tro_catalog.TroDefinition`: die Definition beschreibt
den *Typ* (FB-Baustein, Bauteile), ein TRO die konkrete Stelle im Layout.
Verknuepft ueber ``type_name``, siehe :meth:`definition`.
FB-Baustein und Bauteile haengen nur am Typ. Sie werden nicht gespeichert,
sondern bei jedem Zugriff aus dem Katalog gelesen (``fb_block``, ``items``)
und so auch in die JSON geschrieben. Steht beim Einlesen einer JSON ein
Wert drin, der nicht zum Katalog passt, ist das ein Fehler - den Typ
aendern, nicht die Bauteile.
Listenfelder werden beim Pruefen kopiert; ``tro.hosts.append(...)`` aendert
also nur diesen TRO.
"""
model_config = _CONFIG
tro_id: str = Field(alias="id")
type_name: str = Field(alias="type")
confidence: str = ""
separators: list[str] = Field(default_factory=list)
predecessors: list[str] = Field(default_factory=list)
successors: list[str] = Field(default_factory=list)
# Anlagenkoordinate (mm); in der JSON als {"x": .., "y": .., "z": ..}
position: tuple[float, float, float] | None = None
# -- aus dem Katalog -----------------------------------------------------
def definition(self) -> TroDefinition | None:
"""Typdefinition aus tro_catalog (None wenn Typ unbekannt)."""
return get_tro(self.type_name)
@computed_field(alias="fbType") # type: ignore[prop-decorator]
@property
def fb_block(self) -> str:
"""FB-Baustein laut Katalog, z. B. 'FB_ILS_MTRO_1Sep' ("" wenn Typ unbekannt)."""
definition = self.definition()
return definition.get_fb_block() if definition else ""
@computed_field # type: ignore[prop-decorator]
@property
def items(self) -> dict[str, int]:
"""Bauteile laut Katalog, z. B. {'Separator': 1} ({} wenn Typ unbekannt).
Ausnahme Lagerlinien: der Katalog nennt nur die Vorgabe (PinStore_Auto
19); hat der TRO Linien, gilt deren Zahl (Layout oder Handkorrektur im
Dialog TRO_EDIT).
"""
definition = self.definition()
if not definition:
return {}
return _with_line_count(definition.get_items(), self.separators,
getattr(self, "hosts", []))
@model_validator(mode="before")
@classmethod
def _check_catalog_fields(cls, data: Any) -> Any:
"""`items`/`fbType` aus einer JSON gegen den Katalog pruefen und entfernen."""
if not isinstance(data, dict) or not ({"items", "fbType", "fb_block"} & data.keys()):
return data
data = dict(data)
items = data.pop("items", None)
fb_block = data.pop("fbType", data.pop("fb_block", None))
type_name = data.get("type", data.get("type_name"))
definition = get_tro(type_name) if isinstance(type_name, str) else None
expected_items = (_with_line_count(definition.get_items(),
data.get("separators", []),
data.get("hosts", []))
if definition else {})
# Dateien vor 2026-10-26 tragen bei Lagerlinien die Katalogvorgabe
catalog_items = definition.get_items() if definition else {}
expected_fb = definition.get_fb_block() if definition else ""
if items is not None and dict(items) not in (expected_items, catalog_items):
raise ValueError(
f"items {items} passt nicht zum Typ '{type_name}' laut tro_catalog "
f"({expected_items}) - den Typ aendern, nicht die Bauteile"
)
if fb_block is not None and (fb_block or "") != expected_fb:
raise ValueError(
f"fbType '{fb_block}' passt nicht zum Typ '{type_name}' laut "
f"tro_catalog ('{expected_fb}')"
)
return data
class Slot(BaseModel):
"""
Eine Stelle eines TRO, an die Hardware gehoert (Konzept Umbau 1).
``slot`` ist der stabile Schluessel innerhalb des TRO (``separator1``,
``switch1``, ``scanner1``, ``drive``); zusammen mit der TRO-ID adressiert er
die Zuordnung der Hardwareplanung. ``layout_id`` ist die TeileId des
Bauteils im aktuellen Export - nur zur Nachverfolgung, nicht stabil ("",
wenn das Bauteil im Layout nicht bekannt ist). Rollen mit ``?`` sind
optional: fehlt das Geraet, wird der zugehoerige xPlugged*-Schalter FALSE.
"""
model_config = _CONFIG
slot: str
layout_id: str = Field(default="", alias="layoutId")
roles: list[str] = Field(default_factory=list)
class JamArea(_Positioned):
"""
Ein Staubereich: die Warteschlange vor einem Separator, Eingang genau eines
TRO (``tro``, Eingang Nummer ``entry``).
Kennung ``J<Separator>`` (Mubea: Sep1024 -> J1024). Mehrere Vorgaenger
koennen in denselben Staubereich einfahren (Zusammenfuehrung); deren
Ausgang zeigt dann auf diesen Staubereich. Der SCL-Name
``"DB_JamArea<NNNN>".stJam<NNNN>_<k>`` (Standard UH01-UH05) braucht die
Foerderernummer aus der Hardwareplanung und steht deshalb nicht hier.
"""
model_config = _CONFIG
jam_id: str = Field(alias="id")
tro: str
entry: int = 1
separator: str = ""
# tragender Foerderer (erster Host des TRO im Layout)
host: str = ""
predecessors: list[str] = Field(default_factory=list)
note: str = ""
position: tuple[float, float, float] | None = None
class LayoutPart(_Positioned):
"""
Ein Bauteil des Layouts (eine CSV-Zeile mit Lage), z. B. Separator,
Ausschleuselement, Kreisel.
Ziel der raeumlichen Zuordnung in merge_hw_layout: die Slots nennen nur die
``layout_id``, die Lage steht hier. ``bbox`` ist die achsparallele
Ausdehnung (mm) um ``position`` aus der CSV-Spalte "Boundingbox" -
gebraucht, um Geraete einem langen Foerderer (Kreisel) zuzuordnen.
"""
model_config = _CONFIG
layout_id: str = Field(alias="layoutId")
kind: str
name: str = ""
position: tuple[float, float, float] | None = None
bbox: tuple[float, float, float] | None = None
def distance_to_box(self, x: float, y: float) -> float | None:
"""Abstand eines Punkts zur Huellbox (0 innerhalb); None ohne Lage/Box."""
if not self.position or not self.bbox:
return None
dx = max(abs(x - self.position[0]) - self.bbox[0] / 2, 0.0)
dy = max(abs(y - self.position[1]) - self.bbox[1] / 2, 0.0)
return (dx * dx + dy * dy) ** 0.5
class DerivedTro(TroBase):
"""
Ein von tro_flow aus dem CSV abgeleiteter TRO (oder eine Systemgrenze).
Zusaetzlich zum Kern: aus welchen Bauteilen er gebildet wurde, warum er
diesen Typ hat, und woher seine Position stammt.
"""
label: str
hosts: list[str]
scanners: list[str]
# Weichen-Bauteile (Ausschleuselement oder ILS Weiche), Reihenfolge wie die
# Katalog-Slots switch1, switch2 ...; switch<i> gehoert bei 2Sep-Typen zu
# separator<i>. "" = Weiche ohne Layout-Bauteil (nicht exportierter
# Beruehrpunkt, Aufwertung nach Abgaengen). Fehlt in Dateien vor 2026-10-16.
switches: list[str] = Field(default_factory=list)
reason: str
colocated: list[str] = Field(default_factory=list)
# Mittelpunkt der Bauteile, aus denen dieser TRO gebildet wurde
position_from: str = Field(default="", alias="positionFrom")
position_points: int = Field(default=0, alias="positionPoints")
# Stellen fuer die Hardwareplanung (derive_slots); leer bei Systemgrenzen
slots: list[Slot] = Field(default_factory=list)
@property
def connected(self) -> bool:
return bool(self.successors or self.predecessors or self.colocated)
class Connection(BaseModel):
"""Gerichtete Verbindung zwischen zwei TROs/Stationen (JSON: from/to/kind)."""
model_config = _CONFIG
src: str = Field(alias="from")
dst: str = Field(alias="to")
# transfer / umlauf / weiche / unbestimmt / normal (siehe tro_flow.link_tros)
kind: str = ""
class TroCandidate(_Positioned):
"""
Eine im Review gemeldete Stelle, an der ein TRO erwartet wird.
Kein TRO: keine Separatoren, keine Verbindungen, keine SCL. Das Bauteil
steckt heute in einem anderen TRO (``belongs_to``). ``type_name`` ist der
hier erwartete Katalogtyp.
"""
model_config = _CONFIG
name: str
type_name: str = Field(default="", alias="type")
position: tuple[float, float, float]
belongs_to: str = Field(default="", alias="belongsTo")
note: str = ""
class Analysis(BaseModel):
"""Ergebnis von tro_flow.analyse_tros: TROs, Kanten, Befunde."""
model_config = _CONFIG
tros: list[DerivedTro] = Field(default_factory=list)
edges: list[tuple[str, str]] = Field(default_factory=list)
# (von, nach) -> Kantenart (transfer / umlauf / weiche / unbestimmt / normal)
edge_kinds: dict[tuple[str, str], str] = Field(default_factory=dict)
isolated: list[DerivedTro] = Field(default_factory=list)
unknown_types: list[str] = Field(default_factory=list)
findings: list[str] = Field(default_factory=list)
# Wie Host/Fanout ermittelt wurden (Doku/Bericht)
method: str = "Materialfluss-Graph (Zuordnung/Nachbarn)"
# Systemgrenzen (BTMT Be-/Entladestationen) - siehe derive_boundaries; kein
# TRO, nur damit der tatsaechliche Materialein-/ausgang sichtbar wird.
boundaries: list[DerivedTro] = Field(default_factory=list)
# Handkorrekturen aus BricsCAD (tro_flow.apply_edits): umbenannte TROs
# (neue ID -> ID aus <csv>_tro_ids.ini) und von Hand angelegte TROs - beide
# gehoeren nicht unter ihrer neuen Kennung in die ID-Datei.
renamed: dict[str, str] = Field(default_factory=dict)
added: list[str] = Field(default_factory=list)
def by_id(self, tro_id: str) -> DerivedTro | None:
for tro in self.tros + self.boundaries:
if tro.tro_id == tro_id:
return tro
return None
def edit_key(self, tro_id: str) -> str:
"""Schluessel des TRO in tro_overrides.ini (ID der ID-Datei bzw. Kennung
des von Hand angelegten TRO)."""
return self.renamed.get(tro_id, tro_id)
class TroDocument(BaseModel):
"""
Die TRO-JSON einer Anlage (``<csv>_tro.json``).
Geschrieben von ``tro_flow --json`` (:meth:`from_analysis`, :meth:`to_json`),
gelesen von scl_skeleton (:meth:`from_json`). Weil das Modell beide Seiten
definiert, faellt ein umbenannter oder falscher Schluessel beim Einlesen
als Fehler auf.
"""
model_config = _CONFIG
schema_id: str = Field(default=SCHEMA, alias="schema")
source: str
created: str = Field(default_factory=lambda: datetime.now().strftime("%Y-%m-%d %H:%M"))
note: str = ""
tros: list[DerivedTro] = Field(default_factory=list)
# Systemgrenzen (SYSn): Materialein-/ausgang, kein TRO - keine SCL
boundaries: list[DerivedTro] = Field(default_factory=list)
connections: list[Connection] = Field(default_factory=list)
candidates: list[TroCandidate] = Field(default_factory=list)
jam_areas: list[JamArea] = Field(default_factory=list, alias="jamAreas")
# Layout-Bauteile mit Lage (fuer merge_hw_layout); leer in aelteren Dateien
parts: list[LayoutPart] = Field(default_factory=list)
findings: list[str] = Field(default_factory=list)
@classmethod
def from_analysis(cls, analysis: Analysis, source: str, note: str = "",
parts: list[LayoutPart] | None = None) -> TroDocument:
"""Dokument aus dem Ergebnis von tro_flow.analyse_tros bilden.
Slots und Staubereiche werden hier abgeleitet, damit die JSON sie
traegt und nicht erst das Rendern sie kennt. ``parts`` (Lage der
Layout-Bauteile) gibt tro_flow aus dem CSV mit.
"""
tros = [t.model_copy(update={"slots": derive_slots(t)}) for t in analysis.tros]
return cls(
source=source,
note=note,
tros=tros,
boundaries=analysis.boundaries,
jam_areas=derive_jam_areas(tros),
parts=list(parts or []),
connections=[
Connection(src=src, dst=dst, kind=analysis.edge_kinds.get((src, dst), ""))
for src, dst in analysis.edges
],
findings=analysis.findings,
)
def part(self, layout_id: str) -> LayoutPart | None:
"""Layout-Bauteil zur TeileId; ein Bahn-Suffix ("0010-R") zaehlt nicht."""
base = layout_id.split("-", 1)[0]
return next((p for p in self.parts if p.layout_id == base), None)
@classmethod
def from_json(cls, text: str) -> TroDocument:
"""JSON-Text einlesen und pruefen (pydantic.ValidationError bei Fehlern)."""
return cls.model_validate_json(text)
def jam_areas_or_derived(self) -> list[JamArea]:
"""jamAreas aus der Datei; bei einer v3-Datei (ohne jamAreas) abgeleitet."""
return self.jam_areas or derive_jam_areas(self.tros)
def to_json(self) -> str:
"""Als JSON-Text mit den Schluesseln der TRO-JSON (id, type, fbType ...)."""
return self.model_dump_json(by_alias=True, indent=2) + "\n"
def by_id(self, tro_id: str) -> DerivedTro | None:
for tro in self.tros + self.boundaries:
if tro.tro_id == tro_id:
return tro
return None
# ---------------------------------------------------------------------------
# Anlagen-JSON: Layout + Zuordnungen der Hardwareplanung (merge_hw_layout)
# ---------------------------------------------------------------------------
class Registration(BaseModel):
"""Lagebezug E-Zeichnung -> Layout, mit dem die Zuordnung lief (Nachweis).
Kopie der Kennwerte aus dxf_registration.Transform - model.py importiert
dxf_registration bewusst nicht (Schichten, siehe Kopf).
"""
model_config = _CONFIG
dx: float
dy: float
rotation: float = 0.0
source: str = ""
matched: int = 0
candidates: int = 0
residual_max: float = Field(default=0.0, alias="residualMax")
description: str = ""
class Binding(BaseModel):
"""
Eine Zuordnung: Rolle ``role`` im Slot ``slot`` des TRO ``tro`` <- BMK.
Stabiler Schluessel ist TRO-ID + Slot + Rolle (TRO-IDs aus der ID-Datei);
``layout_id`` dient nur zur Nachverfolgung im aktuellen Layoutstand. Bei
``status == "ohne_slot"`` sind ``tro``/``slot`` leer und ``role`` ist die
Regel (``Ausschleus.stau``). ``constant``/``address`` kommen aus dem
TIA-Import und sind das, was im SCL steht (``cInBG0532``, ``%E053.2``).
"""
model_config = _CONFIG
tro: str = ""
slot: str = ""
role: str
layout_id: str = Field(default="", alias="layoutId")
bmk: str = ""
constant: str = ""
address: str = ""
# positionsdraw | hw_overrides.ini
source: str = ""
# raeumlich | nummer | hand
rule: str = ""
distance: float | None = None
runner_up: float | None = Field(default=None, alias="runnerUp")
status: str = "zugeordnet"
note: str = ""
@field_validator("status")
@classmethod
def _known_status(cls, value: str) -> str:
if value not in BINDING_STATUS:
raise ValueError(f"status '{value}' unbekannt, erlaubt: {', '.join(BINDING_STATUS)}")
return value
class ConveyorBinding(BaseModel):
"""
Ein Foerderer der Hardwareplanung (``K0102``, ``K0106``) an einem
Layout-Foerderer (``layout_id`` "0010" = Kreisel3).
Die Namenswelten passen nicht zusammen (Layout: Kreisel1..3, E-Planung:
K0102/K0104/K0105), deshalb raeumlich: alle Feldgeraete, deren
Bezeichnung den Foerderer nennt, stimmen fuer den naechsten
Layout-Foerderer ab (``votes`` von ``total``).
"""
model_config = _CONFIG
eplan_name: str = Field(alias="eplanName")
layout_id: str = Field(default="", alias="layoutId")
layout_name: str = Field(default="", alias="layoutName")
votes: int = 0
total: int = 0
motor: str = ""
protection: str = ""
encoder: str = ""
# mehrheit | hand
rule: str = ""
status: str = "zugeordnet"
note: str = ""
class AnlageDocument(BaseModel):
"""
Die Anlagen-JSON ``<csv>_anlage.json``: das Layout (TRO-JSON) unveraendert
unter ``layout``, daneben die Zuordnungen der Hardwareplanung.
Geschrieben von merge_hw_layout, kuenftig die eine Eingabe von scl_skeleton
(doc/Konzept_Layout_Hardware_SCL.md, Umbau 4).
"""
model_config = _CONFIG
schema_id: str = Field(default=ANLAGE_SCHEMA, alias="schema")
created: str = Field(default_factory=lambda: datetime.now().strftime("%Y-%m-%d %H:%M"))
layout_file: str = Field(default="", alias="layoutFile")
hw_file: str = Field(default="", alias="hwFile")
project: str = ""
registration: Registration | None = None
layout: TroDocument
bindings: list[Binding] = Field(default_factory=list)
conveyors: list[ConveyorBinding] = Field(default_factory=list)
# Feldgeraete mit Lage, die keine Zuordnung bekommen haben (BMK)
unassigned: list[str] = Field(default_factory=list)
warnings: list[str] = Field(default_factory=list)
def binding(self, tro_id: str, slot: str, role: str) -> Binding | None:
"""Zuordnung einer Slot-Rolle (Rolle ohne "?"), None wenn keine."""
role = role.rstrip("?")
return next((b for b in self.bindings
if b.tro == tro_id and b.slot == slot and b.role == role), None)
def conveyor_for(self, layout_id: str) -> ConveyorBinding | None:
"""Foerderer der Hardwareplanung an einem Layout-Foerderer ("0010-R" geht)."""
base = layout_id.split("-", 1)[0]
return next((c for c in self.conveyors
if c.layout_id == base and c.status == "zugeordnet"), None)
def to_json(self) -> str:
return self.model_dump_json(by_alias=True, indent=2) + "\n"
@classmethod
def from_json(cls, text: str) -> AnlageDocument:
return cls.model_validate_json(text)
# ---------------------------------------------------------------------------
# Ableitungen aus Katalogtyp und Topologie
# ---------------------------------------------------------------------------
# Rollen je Slot-Art (doc/Konzept_Layout_Hardware_SCL.md, Umbau 1); "?" = optional
_SEPARATOR_ROLES = ("sensor", "stopper", "jamSensor?", "partSensor?")
_SWITCH_ROLES = ("valve", "exitSensor?")
_SCANNER_ROLES = ("reader",)
_DRIVE_ROLES = ("motor", "motorProtection", "senCarrInPos", "senFinger",
"senLastPos", "senJam")
_STORAGE_LINE_ROLES = ("sensor", "stopper", "jamSensor?")
def _at(values: list[str], index: int) -> str:
return values[index] if index < len(values) else ""
def _lines_of(separators: list[str], hosts: list[str]) -> list[str]:
seps, hosts = list(separators or []), list(hosts or [])
count = len(hosts) if 0 < len(hosts) <= len(seps) else len(seps)
return seps[len(seps) - count:]
def storage_lines(tro: TroBase) -> list[str]:
"""Separatoren der Lagerlinien eines Speicher-TRO.
tro_flow legt sie als [steuernde ..., Linien ...] ab; die Linien sind die
letzten len(hosts) (eine je Speicherbahn). Fehlen Hosts, zaehlt jeder
Separator als Linie - dieselbe Regel wie scl_context._store.
"""
return _lines_of(tro.separators, getattr(tro, "hosts", []))
def _with_line_count(items: dict[str, int], separators: list[str],
hosts: list[str]) -> dict[str, int]:
"""Katalog-Bauteile, Lagerlinien auf die Zahl der Linien des TRO gesetzt."""
if items.get(TroItem.STORAGE_LINE) and (lines := _lines_of(separators, hosts)):
items = {**items, TroItem.STORAGE_LINE: len(lines)}
return items
# Slot-Art (Name ohne Nummer) -> Bauteilarten (Element.kind im CSV), die in
# diesen Slot passen. Gebraucht vom Dialog TRO_EDIT (Pruefung beim Waehlen im
# Plan, tro_annotate --emit-lisp) und beim Anwenden der Handkorrekturen
# (tro_flow.apply_edits). "drive" ist der tragende Foerderer.
SLOT_PART_KINDS: dict[str, tuple[str, ...]] = {
"separator": ("Separator",),
"line": ("Separator",),
"switch": ("Ausschleus", "Weiche"),
"scanner": ("Scanner",),
"drive": ("Strecke", "Gefaellestrecke", "Kreisel"),
}
def slot_kind(slot: str) -> str:
"""'separator2' -> 'separator', 'line17' -> 'line', 'drive' -> 'drive'."""
return slot.rstrip("0123456789")
def slot_names(type_name: str) -> list[str]:
"""Slots eines Typs in der Reihenfolge von derive_slots, ohne Lagerlinien.
Die Lagerlinien (line1..lineN) haengen am TRO, nicht am Typ - ob ein Typ
sie hat, sagt :func:`has_lines`.
"""
definition = get_tro(type_name)
items = definition.get_items() if definition else {}
names = [f"separator{i}" for i in range(1, items.get(TroItem.SEPARATOR, 0) + 1)]
names += [f"switch{i}" for i in range(1, items.get(TroItem.SWITCH, 0) + 1)]
names += [f"scanner{i}" for i in range(1, items.get(TroItem.SCANNER, 0) + 1)]
if items.get(TroItem.VARIO_DRIVE):
names.append("drive")
return names
def has_lines(type_name: str) -> int:
"""Vorgabe der Lagerlinien laut Katalog (0 = Typ hat keine)."""
definition = get_tro(type_name)
return definition.get_items().get(TroItem.STORAGE_LINE, 0) if definition else 0
def derive_slots(tro: DerivedTro) -> list[Slot]:
"""Slots eines TRO: Anzahl aus dem Katalog, Layout-ID aus dem TRO.
Weichen-Slots tragen das Weichen- bzw. Ausschleuselement aus
``tro.switches`` (leer, wenn tro_flow keins kennt). Ein Speicher bekommt je
Lagerlinie einen Slot ``line<n>``; der erste Separator-Slot und
``line1`` koennen dasselbe Bauteil sein (wie im PinStore-Template).
"""
items = tro.items
slots: list[Slot] = []
for i in range(items.get(TroItem.SEPARATOR, 0)):
slots.append(Slot(slot=f"separator{i + 1}", layout_id=_at(tro.separators, i),
roles=list(_SEPARATOR_ROLES)))
if items.get(TroItem.STORAGE_LINE):
for n, sep in enumerate(storage_lines(tro), start=1):
slots.append(Slot(slot=f"line{n}", layout_id=sep,
roles=list(_STORAGE_LINE_ROLES)))
for i in range(items.get(TroItem.SWITCH, 0)):
slots.append(Slot(slot=f"switch{i + 1}", layout_id=_at(tro.switches, i),
roles=list(_SWITCH_ROLES)))
for i in range(items.get(TroItem.SCANNER, 0)):
slots.append(Slot(slot=f"scanner{i + 1}", layout_id=_at(tro.scanners, i),
roles=list(_SCANNER_ROLES)))
if items.get(TroItem.VARIO_DRIVE):
slots.append(Slot(slot="drive", layout_id=_at(tro.hosts, 0),
roles=list(_DRIVE_ROLES)))
return slots
def derive_jam_areas(tros: list[DerivedTro]) -> list[JamArea]:
"""Staubereiche: je Separator laut Katalog ein Eingang des TRO.
Kennung J<Separator>; fehlt der Separator im Layout, J<TRO-ID>_<n>. Alle
Vorgaenger fahren in den ersten Eingang ein - bei zwei Eingaengen
(2Sep-Typen) ist die Aufteilung aus dem Layout nicht ableitbar und als
Hinweis vermerkt.
"""
areas: list[JamArea] = []
for tro in tros:
n_entries = max(1, tro.items.get(TroItem.SEPARATOR, 1))
if n_entries > 1 and tro.predecessors:
note = "Aufteilung der Vorgaenger auf die Eingaenge offen"
elif tro.items.get(TroItem.STORAGE_LINE):
# Die Separatoren eines Speichers sitzen auf den Lagerlinien; der
# Eingang liegt davor und hat im Layout keinen eigenen Separator.
note = "Eingang des Speichers - Kennung vorlaeufig nach erstem Linien-Separator"
else:
note = ""
for n in range(1, n_entries + 1):
sep = _at(tro.separators, n - 1)
areas.append(JamArea(
jam_id=f"J{sep}" if sep else f"J{tro.tro_id}_{n}",
tro=tro.tro_id,
entry=n,
separator=sep,
host=_at(tro.hosts, 0),
predecessors=list(tro.predecessors) if n == 1 else [],
note=note,
position=tro.position if n == 1 else None,
))
return areas
Reference in New Issue View Git Blame Copy Permalink