mistangl/sps_skel
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

model: TroCandidate statt OpenPoint, tro_overrides nutzt model-Klassen (Schema v3)

Browse Source 
- OpenPoint -> model.TroCandidate (type, position, belongsTo, note);
  TroDocument.candidates ersetzt openPoints, Schema sps_skel-tro-v3
- Positionsbehandlung von TroBase und TroCandidate im Mixin _Positioned
- tro_overrides.py: eigene ConnectionSpec/OpenPoint entfallen, stattdessen
  model.Connection / model.TroCandidate; einheitliche Feldnamen
  (tro_id/separator, hosts als Liste)
- Fixtures, Beispiel-JSON und Doku auf v3 nachgezogen

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Michael Stangl
2026-10-10 11:31:05 +02:00
co-authored by Claude Opus 5.5
parent de8ecb9460
commit 29077caa81
10 changed files with 153 additions and 134 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/material_flow.py
+5 -5
View File
@@ -2349,7 +2349,7 @@ def render_connect_ini(graph: Graph, section: str = "Anlage", overrides=None) ->
add("")
if overrides is not None and (overrides.tros or overrides.splits
or overrides.open_points):
or overrides.candidates):
add(f"[{section}.manual]")
add("; Vor Ort bestaetigt, im CSV-Export nicht enthalten.")
add(f"; Quelle: {overrides.source}")
@@ -2358,16 +2358,16 @@ def render_connect_ini(graph: Graph, section: str = "Anlage", overrides=None) ->
for section_key, override in sorted(overrides.tros.items()):
if not override.type_name:
continue
wie = f"Separator {override.key}" if override.by_separator else override.key
wie = f"Separator {override.separator}" if override.separator else override.tro_id
add(f"{section_key} = {override.type_name} | {wie} | {override.reason}")
for spec in overrides.splits:
add(f"{spec.name} = NEU {spec.type_name} "
f"| Separator {', '.join(spec.separators)}"
+ (f" + Scanner {', '.join(spec.scanners)}" if spec.scanners else "")
+ (f" | Bahn {spec.host}" if spec.host else "")
+ (f" | Bahn {', '.join(spec.hosts)}" if spec.hosts else "")
+ f" | {spec.reason}")
for point in overrides.open_points:
add(f"{point.name} = OFFEN {point.expect or '?'} "
for point in overrides.candidates:
add(f"{point.name} = OFFEN {point.type_name or '?'} "
f"| x={point.x:.2f} y={point.y:.2f} "
f"| heute Teil von {point.belongs_to or '-'} "
f"| {point.note}")
lib/model.py
+64 -49
View File
@@ -19,20 +19,23 @@ Klassen:
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
TroCandidate eine im Review gemeldete Stelle, an der ein TRO erwartet
wird, die aber (noch) keiner 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`).
`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 Any
from typing import TYPE_CHECKING, Any
from pydantic import (
BaseModel,
@@ -51,14 +54,15 @@ __all__ = [
"TroBase",
"DerivedTro",
"Connection",
"OpenPoint",
"TroCandidate",
"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"
# v2 die von tro_flow --json (aus dem CSV, ueber dieses Modell), v3 wie v2 mit
# `candidates` statt `openPoints`.
SCHEMA = "sps_skel-tro-v3"
# Gemeinsame Einstellungen: Pruefung bei Erzeugung und Zuweisung, unbekannte
# Schluessel sind ein Fehler (ein umbenannter JSON-Schluessel faellt auf statt
@@ -71,7 +75,48 @@ _CONFIG = ConfigDict(
)
class TroBase(BaseModel):
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.
@@ -145,37 +190,6 @@ class TroBase(BaseModel):
)
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):
"""
@@ -210,22 +224,23 @@ class Connection(BaseModel):
kind: str = ""
class OpenPoint(BaseModel):
"""Eine im Review gemeldete Stelle, die noch kein eigener TRO ist."""
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
x: float
y: float
z: float | None = None
expect: str = ""
type_name: str = Field(default="", alias="type")
position: tuple[float, float, float]
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."""
@@ -272,7 +287,7 @@ class TroDocument(BaseModel):
# 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")
candidates: list[TroCandidate] = Field(default_factory=list)
findings: list[str] = Field(default_factory=list)
@classmethod
lib/tro_overrides.py
+41 -60
View File
@@ -26,15 +26,19 @@ mehrere Anlagen in einer Datei stehen koennen):
reason = ... Begruendung (kommt in die Doku)
confidence = bestaetigt Vertrauensangabe (Default: bestaetigt)
[<csv>.openpoint.<Name>] offener Punkt - noch KEIN TRO
[<csv>.openpoint.<Name>] offener Punkt - noch KEIN TRO (model.TroCandidate)
x = 529.95 Anlagenkoordinate in mm
y = -3895.27
z = 1468
expect = 1Sep1Swi hier erwarteter TRO-Typ
expect = 1Sep1Swi hier erwarteter TRO-Typ (-> type_name)
belongs_to = TRO10 TRO, der das Bauteil heute enthaelt
note = ... Erlaeuterung
Offene Punkte aendern die TRO-Struktur ausdruecklich **nicht**.
Die Felder heissen wie in lib/model.py (`type_name`, `separators`, `hosts`,
`reason`, `confidence`); Verbindungen und offene Punkte sind direkt
`model.Connection` bzw. `model.TroCandidate`.
"""
from __future__ import annotations
@@ -44,6 +48,7 @@ import os
from dataclasses import dataclass, field
from pathlib import Path
from model import Connection, TroCandidate
from tro_catalog import get_tro
OVERRIDE_FILE = "tro_overrides.ini"
@@ -63,16 +68,17 @@ class TroOverride:
wegfallen; die Separator-Nummer aus dem CSV-Export tut das nicht.
"""
key: str
by_separator: bool = False
# genau eins von beiden ist gesetzt
tro_id: str = ""
separator: str = ""
type_name: str = ""
reason: str = ""
confidence: str = ""
def matches(self, tro) -> bool:
if self.by_separator:
return self.key in tro.separators
return self.key == tro.tro_id
if self.separator:
return self.separator in tro.separators
return self.tro_id == tro.tro_id
@dataclass
@@ -90,21 +96,12 @@ class SplitSpec:
type_name: str
separators: list[str] = field(default_factory=list)
scanners: list[str] = field(default_factory=list)
host: str = ""
hosts: list[str] = field(default_factory=list)
label: str = ""
reason: str = ""
confidence: str = DEFAULT_CONFIDENCE
@dataclass
class ConnectionSpec:
"""Eine vor Ort bestaetigte Verbindung zwischen zwei TROs."""
src: str
dst: str
kind: str = ""
@dataclass
class MergeSpec:
"""Mehrere abgeleitete TROs zu einem Steuerobjekt zusammenfassen.
@@ -119,28 +116,12 @@ class MergeSpec:
name: str
type_name: str
separators: list[str] = field(default_factory=list)
host: str = ""
hosts: list[str] = field(default_factory=list)
label: str = ""
reason: str = ""
confidence: str = DEFAULT_CONFIDENCE
@dataclass
class OpenPoint:
"""Eine im Review gemeldete Stelle, die noch kein eigener TRO ist."""
name: str
x: float
y: float
z: float | None = None
expect: str = ""
belongs_to: str = ""
note: str = ""
def describe_position(self) -> str:
return f"{self.x:.0f}, {self.y:.0f}"
@dataclass
class Overrides:
"""Alle Korrekturen zu einer CSV-Datei."""
@@ -150,15 +131,15 @@ class Overrides:
splits: list[SplitSpec] = field(default_factory=list)
merges: list[MergeSpec] = field(default_factory=list)
# Gesetzte Topologie: ersetzt die abgeleiteten Kanten vollstaendig
connections: list[ConnectionSpec] = field(default_factory=list)
open_points: list[OpenPoint] = field(default_factory=list)
connections: list[Connection] = field(default_factory=list)
candidates: list[TroCandidate] = field(default_factory=list)
# Protokoll der tatsaechlich angewandten Korrekturen (fuer die Doku)
applied: list[str] = field(default_factory=list)
warnings: list[str] = field(default_factory=list)
def __bool__(self) -> bool:
return bool(self.tros or self.splits or self.merges or self.connections
or self.open_points)
or self.candidates)
def override_path(cfg_dir: Path | None = None) -> Path:
@@ -179,6 +160,10 @@ def _as_float(raw: str, where: str, warnings: list[str]) -> float | None:
return None
def _as_list(raw: str) -> list[str]:
return [s.strip() for s in raw.split(",") if s.strip()]
def load_overrides(csv_name: str, cfg_dir: Path | None = None) -> Overrides:
"""Korrekturen zu `csv_name` laden (leeres Ergebnis, wenn es keine gibt)."""
path = override_path(cfg_dir)
@@ -205,7 +190,7 @@ def load_overrides(csv_name: str, cfg_dir: Path | None = None) -> Overrides:
)
continue
result.connections.append(
ConnectionSpec(src=src, dst=dst, kind=rest.strip())
Connection(src=src, dst=dst, kind=rest.strip())
)
continue
@@ -230,8 +215,8 @@ def load_overrides(csv_name: str, cfg_dir: Path | None = None) -> Overrides:
key = name.upper()
by_sep = key.startswith("SEP")
result.tros[key] = TroOverride(
key=key[3:] if by_sep else key,
by_separator=by_sep,
tro_id="" if by_sep else key,
separator=key[3:] if by_sep else "",
type_name=type_name,
reason=values.get("reason", "").strip(),
confidence=values.get("confidence", "").strip() or DEFAULT_CONFIDENCE,
@@ -244,7 +229,7 @@ def load_overrides(csv_name: str, cfg_dir: Path | None = None) -> Overrides:
f"- neuer TRO nicht angelegt"
)
continue
seps = [s.strip() for s in values.get("separators", "").split(",") if s.strip()]
seps = _as_list(values.get("separators", ""))
if not seps:
result.warnings.append(
f"[{section}]: 'separators' fehlt - ohne Separator kein TRO"
@@ -255,9 +240,8 @@ def load_overrides(csv_name: str, cfg_dir: Path | None = None) -> Overrides:
name=name,
type_name=type_name,
separators=seps,
scanners=[s.strip() for s in values.get("scanners", "").split(",")
if s.strip()],
host=values.get("host", "").strip(),
scanners=_as_list(values.get("scanners", "")),
hosts=_as_list(values.get("hosts", values.get("host", ""))),
label=values.get("label", "").strip(),
reason=values.get("reason", "").strip(),
confidence=values.get("confidence", "").strip() or DEFAULT_CONFIDENCE,
@@ -271,7 +255,7 @@ def load_overrides(csv_name: str, cfg_dir: Path | None = None) -> Overrides:
f"- Zusammenfassung nicht durchgefuehrt"
)
continue
seps = [s.strip() for s in values.get("separators", "").split(",") if s.strip()]
seps = _as_list(values.get("separators", ""))
if len(seps) < 2:
result.warnings.append(
f"[{section}]: 'separators' braucht mindestens zwei Nummern"
@@ -282,7 +266,7 @@ def load_overrides(csv_name: str, cfg_dir: Path | None = None) -> Overrides:
name=name,
type_name=type_name,
separators=seps,
host=values.get("host", "").strip(),
hosts=_as_list(values.get("hosts", values.get("host", ""))),
label=values.get("label", "").strip(),
reason=values.get("reason", "").strip(),
confidence=values.get("confidence", "").strip() or DEFAULT_CONFIDENCE,
@@ -294,30 +278,27 @@ def load_overrides(csv_name: str, cfg_dir: Path | None = None) -> Overrides:
if x is None or y is None:
continue
z_raw = values.get("z", "").strip()
expect = values.get("expect", "").strip()
if expect and get_tro(expect) is None:
z = _as_float(z_raw, f"[{section}] z", result.warnings) if z_raw else None
type_name = values.get("expect", "").strip()
if type_name and get_tro(type_name) is None:
result.warnings.append(
f"[{section}]: erwarteter Typ '{expect}' steht nicht im "
f"[{section}]: erwarteter Typ '{type_name}' steht nicht im "
f"TRO-Katalog"
)
result.open_points.append(
OpenPoint(
result.candidates.append(
TroCandidate(
name=name,
x=x,
y=y,
z=_as_float(z_raw, f"[{section}] z", result.warnings)
if z_raw
else None,
expect=expect,
type_name=type_name,
position=(x, y, z if z is not None else 0.0),
belongs_to=values.get("belongs_to", "").strip(),
note=values.get("note", "").strip(),
)
)
else:
result.warnings.append(
f"[{section}]: unbekannte Art '{kind}' - erwartet 'tro' oder "
f"'openpoint'"
f"[{section}]: unbekannte Art '{kind}' - erwartet 'tro', 'split', "
f"'merge' oder 'openpoint'"
)
result.open_points.sort(key=lambda p: p.name)
result.candidates.sort(key=lambda p: p.name)
return result