sps_skel/lib/tro_ids.py

# -*- coding: utf-8 -*-
"""
tro_ids.py - stabile TRO-IDs ueber eine ID-Datei je Anlage.

Die TRO-IDs vergeben die SPS-Programmierer; eine ID muss nur eindeutig sein.
Sobald SCL-Code steht, darf sich eine ID nicht mehr aendern, nur weil das
Layout neu exportiert wurde. Die Layout-TeileId taugt dafuer nicht (sie
wechselt zwischen zwei Exporten derselben Anlage), stabil ist nur die Lage.
Siehe doc/Konzept_Layout_Hardware_SCL.md, "Festlegungen" -> "TRO-IDs".

Die ID-Datei (`%SKEL_CFG%/<csv>_tro_ids.ini`, versioniert, von Hand
umbenennbar) haelt je TRO die ID und die Lage seines Anker-Separators:

    [tro]
    TRO01 = 1234.5, -567.8    ; 1Sep1Swi | Separator 0003 auf Kreisel2 Bahn L
    K0102_1 = 2210.0, 410.0   ; von Hand umbenannt

Ablauf (assign_ids):

- ohne ID-Datei bzw. mit renumber: alle TROs in der uebergebenen Reihenfolge
  (entlang des Materialflusses) neu durchnummerieren, TRO01, TRO02, ...
- mit ID-Datei: jeder Eintrag findet den TRO, dessen Separator seiner Lage am
  naechsten liegt (innerhalb der Toleranz, je TRO hoechstens ein Eintrag).
  Nicht gefundene TROs bekommen die naechste freie Nummer, nicht gefundene
  Eintraege bleiben in der Datei stehen und ihre ID wird nicht wiederverwendet.

Das Modul liest und schreibt nur die Datei; Lage und Reihenfolge der TROs
liefert tro_flow.py.
"""

from __future__ import annotations

import configparser
import math
import re
from dataclasses import dataclass, field
from datetime import datetime
from pathlib import Path

from parameter import get_float

__all__ = [
    "SECTION",
    "DEFAULT_TOLERANCE",
    "IdEntry",
    "IdAssignment",
    "TroIds",
    "ids_path",
    "load_ids",
    "assign_ids",
    "render_ids",
    "render_new_entries",
    "insert_entries",
    "other_sections",
]

SECTION = "tro"
ID_FILE_SUFFIX = "_tro_ids.ini"
PREFIX = "TRO"

# Abstand (mm) zwischen Eintrag und Separator, bis zu dem ein TRO als
# wiedergefunden gilt - aus cfg/parameter.ini [tro_ids] (lib/parameter.py).
DEFAULT_TOLERANCE = get_float("tro_ids", "default_tolerance")

# Eine ID wird zum SCL-Bezeichner (Instanz #<ID>, "DB_TRO_TEST".<ID>).
_ID_PATTERN = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$")
# SYSn sind die Systemgrenzen (tro_flow.derive_boundaries), keine TROs.
_RESERVED = re.compile(r"^SYS\d+$")
_NUMBERED = re.compile(rf"^{PREFIX}(\d+)$")


@dataclass
class IdEntry:
    """Ein Eintrag der ID-Datei: ID und Lage des Anker-Separators (mm)."""

    tro_id: str
    x: float
    y: float


@dataclass
class IdAssignment:
    """Ergebnis von assign_ids, fuer den Bericht."""

    renumbered: bool = False
    # (ID, Abstand in mm) der wiedergefundenen TROs
    kept: list[tuple[str, float]] = field(default_factory=list)
    # IDs, die in diesem Lauf neu vergeben wurden (ohne renumbered)
    new: list[str] = field(default_factory=list)
    # Eintraege der Datei ohne passenden TRO
    missing: list[IdEntry] = field(default_factory=list)
    # weggefallene ID -> TRO, dessen Separator dort liegt, der aber schon eine
    # naehere ID traegt (z. B. zwei TROs zu einem 2Sep2Swi zusammengelegt)
    merged_into: dict[str, str] = field(default_factory=dict)


@dataclass
class TroIds:
    """ID-Datei einer Anlage plus Einstellungen fuer einen Lauf.

    entries ist leer, wenn die Datei (noch) nicht existiert. result fuellt
    tro_flow.derive_tros ueber assign_ids.
    """

    path: Path
    entries: list[IdEntry] = field(default_factory=list)
    exists: bool = False
    tolerance: float = DEFAULT_TOLERANCE
    renumber: bool = False
    result: IdAssignment | None = None

    @classmethod
    def load(cls, path: Path, *, tolerance: float = DEFAULT_TOLERANCE,
             renumber: bool = False) -> "TroIds":
        exists = path.is_file()
        entries = load_ids(path) if exists else []
        return cls(path=path, entries=entries, exists=exists,
                   tolerance=tolerance, renumber=renumber)


def ids_path(csv_stem: str, cfg_dir: Path) -> Path:
    """Standardpfad der ID-Datei zu einer CSV: <cfg>/<stem>_tro_ids.ini."""
    return cfg_dir / f"{csv_stem}{ID_FILE_SUFFIX}"


def _parser() -> configparser.ConfigParser:
    parser = configparser.ConfigParser(
        inline_comment_prefixes=(";", "#"),
        interpolation=None,
        strict=True,
    )
    parser.optionxform = str  # IDs nicht in Kleinbuchstaben wandeln
    return parser


def load_ids(path: Path) -> list[IdEntry]:
    """ID-Datei lesen. ValueError bei doppelter ID, falscher Lage oder ID."""
    parser = _parser()
    try:
        parser.read(path, encoding="utf-8")
    except configparser.Error as exc:
        raise ValueError(f"{path}: {exc}") from exc
    if not parser.has_section(SECTION):
        raise ValueError(f"{path}: Abschnitt [{SECTION}] fehlt")

    entries: list[IdEntry] = []
    for tro_id, value in parser.items(SECTION):
        if not _ID_PATTERN.match(tro_id):
            raise ValueError(f"{path}: '{tro_id}' ist kein gueltiger SCL-Bezeichner")
        if _RESERVED.match(tro_id):
            raise ValueError(f"{path}: '{tro_id}' ist fuer Systemgrenzen reserviert")
        parts = [p.strip() for p in value.split(",")]
        try:
            x, y = (float(p) for p in parts)
        except ValueError:
            raise ValueError(f"{path}: {tro_id} = '{value}' - erwartet 'x, y' in mm") \
                from None
        entries.append(IdEntry(tro_id, x, y))
    return entries


def _distance(tro, entry: IdEntry, anchors: dict[str, tuple[float, float]]) -> float:
    """Kleinster Abstand eines Separators des TRO zum Eintrag (inf ohne Lage)."""
    best = math.inf
    for sep in tro.separators:
        pos = anchors.get(sep)
        if pos is not None:
            best = min(best, math.hypot(pos[0] - entry.x, pos[1] - entry.y))
    return best


def _width(count: int) -> int:
    return max(2, len(str(count)))


def assign_ids(
    tros: list,
    anchors: dict[str, tuple[float, float]],
    entries: list[IdEntry],
    *,
    tolerance: float = DEFAULT_TOLERANCE,
    renumber: bool = False,
) -> IdAssignment:
    """tro.tro_id fuer alle TROs setzen (Reihenfolge = Reihenfolge der Liste).

    anchors: Separator-TeileId -> (x, y) in mm.
    """
    result = IdAssignment(renumbered=renumber or not entries)
    if result.renumbered:
        width = _width(len(tros))
        for index, tro in enumerate(tros, start=1):
            tro.tro_id = f"{PREFIX}{index:0{width}d}"
        return result

    # Paare nach Abstand; jeder TRO und jeder Eintrag hoechstens einmal
    pairs = sorted(
        (d, i, j)
        for i, tro in enumerate(tros)
        for j, entry in enumerate(entries)
        if (d := _distance(tro, entry, anchors)) <= tolerance
    )
    tro_taken: dict[int, tuple[int, float]] = {}
    entry_taken: set[int] = set()
    for d, i, j in pairs:
        if i in tro_taken or j in entry_taken:
            continue
        tro_taken[i] = (j, d)
        entry_taken.add(j)

    used = {e.tro_id for e in entries}
    numbers = [int(m.group(1)) for e in entries if (m := _NUMBERED.match(e.tro_id))]
    widths = [len(m.group(1)) for e in entries if (m := _NUMBERED.match(e.tro_id))]
    width = max(widths + [_width(len(tros))])
    next_number = max(numbers, default=0) + 1

    for i, tro in enumerate(tros):
        if i in tro_taken:
            j, d = tro_taken[i]
            tro.tro_id = entries[j].tro_id
            result.kept.append((tro.tro_id, d))
            continue
        while f"{PREFIX}{next_number:0{width}d}" in used:
            next_number += 1
        tro.tro_id = f"{PREFIX}{next_number:0{width}d}"
        used.add(tro.tro_id)
        next_number += 1
        result.new.append(tro.tro_id)

    result.missing = [e for j, e in enumerate(entries) if j not in entry_taken]
    for d, i, j in pairs:
        if j not in entry_taken:
            result.merged_into.setdefault(entries[j].tro_id, tros[i].tro_id)
    return result


def _anchor(tro, anchors: dict[str, tuple[float, float]]) -> tuple[float, float] | None:
    for sep in tro.separators:
        if sep in anchors:
            return anchors[sep]
    return None


def _entry_lines(tros: list, anchors: dict[str, tuple[float, float]]) -> list[str]:
    lines = []
    for tro in tros:
        pos = _anchor(tro, anchors)
        if pos is None:
            continue
        seps = ", ".join(tro.separators)
        lines.append(f"{tro.tro_id} = {pos[0]:.1f}, {pos[1]:.1f}"
                     f"    ; {tro.type_name} | {tro.label} | Sep {seps}")
    return lines


def render_ids(tros: list, anchors: dict[str, tuple[float, float]], source: str) -> str:
    """Komplette ID-Datei (erster Lauf oder renumber)."""
    head = [
        "; ============================================================",
        f"; TRO-IDs zu {source} - erzeugt von lib/tro_flow.py "
        f"am {datetime.now():%Y-%m-%d}",
        ";",
        "; Je TRO: ID = x, y (mm) seines Anker-Separators im CSV-Koordinatensystem.",
        "; Die IDs duerfen umbenannt werden (eindeutig, gueltiger SCL-Bezeichner);",
        "; tro_flow findet jeden TRO ueber die Lage wieder. Neue TROs werden",
        "; unten angehaengt, weggefallene bleiben stehen und werden gemeldet.",
        "; Alles neu nummerieren: tro_flow --renumber (ueberschreibt diese Datei).",
        "; Siehe doc/Konzept_Layout_Hardware_SCL.md, Abschnitt TRO-IDs.",
        "; ============================================================",
        "",
        f"[{SECTION}]",
    ]
    return "\n".join(head + _entry_lines(tros, anchors)) + "\n"


def render_new_entries(tros: list, anchors: dict[str, tuple[float, float]],
                       new_ids: list[str], source: str) -> str:
    """Anhang fuer neu gefundene TROs an eine bestehende ID-Datei."""
    neu = [t for t in tros if t.tro_id in set(new_ids)]
    head = [f"; neu aus {source} am {datetime.now():%Y-%m-%d}"]
    return "\n".join(head + _entry_lines(neu, anchors)) + "\n"


def other_sections(path: Path) -> str:
    """Alle Abschnitte ausser [tro] als Text (bleiben bei renumber erhalten)."""
    if not path.is_file():
        return ""
    header = re.compile(r"^\s*\[(.+)\]\s*$")
    keep: list[str] = []
    current = None
    for line in path.read_text(encoding="utf-8").splitlines():
        if m := header.match(line):
            current = m.group(1).strip()
        if current is not None and current != SECTION:
            keep.append(line)
    return "\n".join(keep).strip("\n")


def insert_entries(path: Path, block: str) -> None:
    """block am Ende des Abschnitts [tro] einfuegen, Rest der Datei unveraendert."""
    lines = path.read_text(encoding="utf-8").splitlines()
    header = re.compile(r"^\s*\[(.+)\]\s*$")
    start = next(i for i, line in enumerate(lines)
                 if (m := header.match(line)) and m.group(1).strip() == SECTION)
    end = next((i for i in range(start + 1, len(lines)) if header.match(lines[i])),
               len(lines))
    # vor Leerzeilen am Abschnittsende einfuegen
    while end > start + 1 and not lines[end - 1].strip():
        end -= 1
    lines[end:end] = block.rstrip("\n").splitlines()
    path.write_text("\n".join(lines) + "\n", encoding="utf-8")