# -*- 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 OpenPoint eine gemeldete Stelle, die noch kein TRO ist Analysis Arbeitsergebnis von tro_flow.analyse_tros TroDocument die TRO-JSON (_tro.json): von tro_flow --json geschrieben, von scl_skeleton gelesen und geprueft In der JSON stehen die Schluessel wie bisher in der TRO-JSON (`id`, `type`, `fbType`, `openPoints`, ...); im Python-Code heissen die Felder weiter `tro_id`/`type_name` (Aliase, `populate_by_name`). """ from __future__ import annotations from datetime import datetime from typing import Any from pydantic import ( BaseModel, ConfigDict, Field, computed_field, field_serializer, field_validator, model_validator, ) from tro_catalog import TroDefinition, get_tro __all__ = [ "SCHEMA", "TroBase", "DerivedTro", "Connection", "OpenPoint", "Analysis", "TroDocument", ] # 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). SCHEMA = "sps_skel-tro-v2" # 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 TroBase(BaseModel): """ 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).""" definition = self.definition() return definition.get_items() if definition else {} @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 = definition.get_items() if definition else {} expected_fb = definition.get_fb_block() if definition else "" if items is not None and dict(items) != expected_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 # -- Position ------------------------------------------------------------ @field_validator("position", mode="before") @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") 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 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] 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") @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 OpenPoint(BaseModel): """Eine im Review gemeldete Stelle, die noch kein eigener TRO ist.""" model_config = _CONFIG name: str x: float y: float z: float | None = None expect: str = "" belongs_to: str = Field(default="", alias="belongsTo") note: str = "" def describe_position(self) -> str: return f"{self.x:.0f}, {self.y:.0f}" 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) 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 class TroDocument(BaseModel): """ Die TRO-JSON einer Anlage (``_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) open_points: list[OpenPoint] = Field(default_factory=list, alias="openPoints") findings: list[str] = Field(default_factory=list) @classmethod def from_analysis(cls, analysis: Analysis, source: str, note: str = "") -> TroDocument: """Dokument aus dem Ergebnis von tro_flow.analyse_tros bilden.""" return cls( source=source, note=note, tros=analysis.tros, boundaries=analysis.boundaries, connections=[ Connection(src=src, dst=dst, kind=analysis.edge_kinds.get((src, dst), "")) for src, dst in analysis.edges ], findings=analysis.findings, ) @classmethod def from_json(cls, text: str) -> TroDocument: """JSON-Text einlesen und pruefen (pydantic.ValidationError bei Fehlern).""" return cls.model_validate_json(text) 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