sps_skel/lib/model.py

# -*- 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 (<csv>_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 (``<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)
    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