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
    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
    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`, `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",
    "DerivedTro",
    "Connection",
    "TroCandidate",
    "Analysis",
    "TroDocument",
    "derive_slots",
    "derive_jam_areas",
]

# 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).
SCHEMA = "sps_skel-tro-v4"

# 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)."""
        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


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 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")
    # 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)

    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)
    candidates: list[TroCandidate] = Field(default_factory=list)
    jam_areas: list[JamArea] = Field(default_factory=list, alias="jamAreas")
    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.

        Slots und Staubereiche werden hier abgeleitet, damit die JSON sie
        traegt und nicht erst das Rendern sie kennt.
        """
        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),
            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 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


# ---------------------------------------------------------------------------
# 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")


def _at(values: list[str], index: int) -> str:
    return values[index] if index < len(values) else ""


def derive_slots(tro: DerivedTro) -> list[Slot]:
    """Slots eines TRO: Anzahl aus dem Katalog, Layout-ID aus dem TRO.

    Weichen tragen noch keine Layout-ID - tro_flow speichert das Weichen-
    bzw. Ausschleuselement heute nicht am TRO.
    """
    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)))
    for i in range(items.get(TroItem.SWITCH, 0)):
        slots.append(Slot(slot=f"switch{i + 1}", 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