sps_skel/lib/dxf_registration.py

# -*- coding: utf-8 -*-
# Absichtlich ohne Shebang: der Windows-py-Launcher wuerde daraus "python3"
# ableiten und scheitert, wenn keine PythonCore-Installation registriert ist.
"""
dxf_registration.py - Lagebezug zwischen CSV-Export und DXF-Zeichnung.

Der CSV-Export und die BricsCAD-Zeichnung stehen in unterschiedlichen
Koordinatensystemen. Damit eine Beschriftung an der richtigen Stelle landet,
muss dieser Versatz bekannt sein - und zwar nachweisbar. Ein stiller Fehler
hier setzt *jede* Markierung falsch, ohne dass es auffaellt.

Vorgehen
--------
1. Punktwolken **je Rolle** bilden: Separatoren des CSV gegen die
   Separator-Bloecke des DXF, Scanner gegen Scanner-Bloecke. Nur Punkte
   derselben Rolle werden einander zugeordnet.
2. Verschiebung abstimmen ("voting"): fuer jedes Punktpaar gleicher Rolle den
   Versatz berechnen, in ein Raster einsortieren, Stimmen zaehlen.
3. Mehrere Kandidaten pruefen: die haeufigsten Rasterplaetze werden einzeln
   ausgeglichen und bewertet. Der Kandidat mit den meisten *exakten* Treffern
   gewinnt.
4. Nachweis: Restfehler je Inlier, Maximum und Mittel. Ueber einer Schwelle
   wird die Registrierung verworfen statt verwendet.

Warum Rollen und mehrere Kandidaten
-----------------------------------
Die Anlage besteht aus regelmaessigen Rastern: die Speicherlinien, ihre Zulaeufe
und ihre Ablaeufe bilden je eine Reihe von 20 Punkten im 800-mm-Abstand. Eine
reine Abstimmung ueber alle Punkte ist deshalb mehrdeutig - ein falscher Versatz
trifft die Zulaufreihe genauso gut wie der richtige die Separatorenreihe und kann
die Abstimmung um eine einzige Stimme gewinnen. Erst die Einschraenkung auf
gleiche Rollen und die Bewertung mehrerer Kandidaten macht das Ergebnis
eindeutig.

Der Ausgleich laeuft nur ueber **exakte** Treffer (FIT_TOL). Bei ausgedehnten
Objekten meldet der CSV-Export den Mittelpunkt der Huellbox, der Block im DXF
seinen Einfuegepunkt; solche Paare liegen 100-500 mm auseinander und wuerden
eine Scheindrehung erzeugen, wenn man sie mit ausgleicht.

Die Transformation gilt als **DXF -> CSV** (so wird sie ermittelt und geprueft).
Fuer das Beschriften wird die Umkehrung gebraucht: :meth:`Transform.to_dxf`.

Persistenz
----------
Eine gepruefte Transformation wird in SKEL_CFG/dxf_registration.json abgelegt,
je Kombination aus CSV- und DXF-Dateiname. Spaetere Laeufe verwenden sie
unveraendert, damit Ergebnisse reproduzierbar sind; --re-register erzwingt eine
neue Ermittlung.
"""

from __future__ import annotations

import json
import math
from collections import Counter
from datetime import datetime
from pathlib import Path

from pydantic import BaseModel, ConfigDict, Field

__all__ = [
    "Transform",
    "RegistrationError",
    "ANCHOR_BLOCKS",
    "derive_transform",
    "verify_transform",
    "load_transform",
    "save_transform",
    "registration_path",
]

# Rollen fuer den Lagebezug: CSV-Objektart -> Blocknamen im DXF.
# Erweiterbar, wenn eine Zeichnung andere Bloecke verwendet.
ANCHOR_BLOCKS: dict[str, tuple[str, ...]] = {
    "Separator": ("S-SP", "S-LP"),
    "Scanner": ("SCAN",),
}

# Rastergroesse der Abstimmung in mm
VOTE_GRID = 100.0
# So viele Kandidaten der Abstimmung werden ausgeglichen und bewertet
VOTE_CANDIDATES = 6
# Nur Paare bis zu diesem Abstand gehen in den Ausgleich ein (exakte Treffer)
FIT_TOL = 50.0
# Punkte gelten als zugeordnet, wenn sie nach der Verschiebung so nah liegen
DEFAULT_MATCH_TOL = 600.0
# So viele exakte Treffer muessen es mindestens sein
MIN_INLIERS = 5
# Ab dieser Drehung wird gewarnt (Grad)
ROTATION_WARN_DEG = 0.05


class RegistrationError(RuntimeError):
    """Der Lagebezug liess sich nicht nachweisbar bestimmen."""


class Transform(BaseModel):
    """
    Starre Transformation DXF -> CSV: erst drehen, dann verschieben.

        csv = R(rotation) * dxf + (dx, dy)
    """

    model_config = ConfigDict(validate_assignment=True, extra="forbid")

    dx: float = Field(description="Verschiebung in X in mm (DXF -> CSV)")
    dy: float = Field(description="Verschiebung in Y in mm (DXF -> CSV)")
    rotation: float = Field(default=0.0, description="Drehung in Grad (DXF -> CSV)")

    source: str = Field(default="auto", description="auto | cfg | cli")
    anchor: str = Field(default="", description="Wie der Bezug gefunden wurde")
    matched: int = Field(default=0, description="Zugeordnete Punktpaare (Inlier)")
    candidates: int = Field(default=0, description="Punktartige CSV-Objekte")
    residual_max: float = Field(default=0.0, description="Groesster Restfehler in mm")
    residual_mean: float = Field(default=0.0, description="Mittlerer Restfehler in mm")
    csv_file: str = Field(default="")
    dxf_file: str = Field(default="")
    created: str = Field(default="")

    # -- Anwenden ----------------------------------------------------------

    def to_csv(self, x: float, y: float) -> tuple[float, float]:
        """Punkt aus dem DXF in das CSV-Koordinatensystem bringen."""
        rad = math.radians(self.rotation)
        cos, sin = math.cos(rad), math.sin(rad)
        return (x * cos - y * sin + self.dx, x * sin + y * cos + self.dy)

    def to_dxf(self, x: float, y: float) -> tuple[float, float]:
        """Punkt aus dem CSV in das DXF-Koordinatensystem bringen (Umkehrung)."""
        rad = math.radians(-self.rotation)
        cos, sin = math.cos(rad), math.sin(rad)
        px, py = x - self.dx, y - self.dy
        return (px * cos - py * sin, px * sin + py * cos)

    # -- Darstellung -------------------------------------------------------

    def describe(self) -> str:
        rot = "" if abs(self.rotation) < 1e-9 else f", Drehung {self.rotation:+.4f} Grad"
        return (f"dx {self.dx:+.1f} mm, dy {self.dy:+.1f} mm{rot} "
                f"[{self.source}, {self.matched}/{self.candidates} Punkte, "
                f"Restfehler max {self.residual_max:.1f} mm]")


# ---------------------------------------------------------------------------
# Ermittlung
# ---------------------------------------------------------------------------


Points = dict[str, list[tuple[float, float]]]


def _vote_offsets(csv_points: Points, dxf_points: Points) -> list[tuple[float, float, int]]:
    """Haeufigste Versaetze bestimmen - nur Punktpaare gleicher Rolle."""
    votes: Counter = Counter()
    for role, cpts in csv_points.items():
        dpts = dxf_points.get(role, [])
        for cx, cy in cpts:
            for dx_, dy_ in dpts:
                key = (round((cx - dx_) / VOTE_GRID), round((cy - dy_) / VOTE_GRID))
                votes[key] += 1
    if not votes:
        raise RegistrationError(
            "keine Punktpaare gleicher Rolle vorhanden - passen die Blocknamen "
            f"in ANCHOR_BLOCKS ({', '.join(sorted(ANCHOR_BLOCKS))}) zur Zeichnung?"
        )
    return [(kx * VOTE_GRID, ky * VOTE_GRID, n)
            for (kx, ky), n in votes.most_common(VOTE_CANDIDATES)]


def _match(
    csv_points: Points,
    dxf_points: Points,
    transform: Transform,
    tol: float,
) -> list[tuple[tuple[float, float], tuple[float, float], float]]:
    """Zu jedem CSV-Punkt den naechsten transformierten DXF-Punkt gleicher Rolle."""
    pairs = []
    for role, cpts in csv_points.items():
        dpts = dxf_points.get(role, [])
        for cx, cy in cpts:
            best, best_d = None, float("inf")
            for dpt in dpts:
                tx, ty = transform.to_csv(*dpt)
                d = math.dist((cx, cy), (tx, ty))
                if d < best_d:
                    best, best_d = dpt, d
            if best is not None and best_d <= tol:
                pairs.append(((cx, cy), best, best_d))
    return pairs


def _fit(
    pairs: list[tuple[tuple[float, float], tuple[float, float], float]]
) -> tuple[float, float, float]:
    """
    Ausgleich per Procrustes: Drehung und Verschiebung DXF -> CSV.

    Ohne Skalierung - die Zeichnung und der Export haben denselben Massstab,
    eine Skalierung waere ein Fehler und soll sich als Restfehler zeigen.
    """
    n = len(pairs)
    cx = sum(p[0][0] for p in pairs) / n
    cy = sum(p[0][1] for p in pairs) / n
    dx = sum(p[1][0] for p in pairs) / n
    dy = sum(p[1][1] for p in pairs) / n

    num = sum((p[1][0] - dx) * (p[0][1] - cy) - (p[1][1] - dy) * (p[0][0] - cx)
              for p in pairs)
    den = sum((p[1][0] - dx) * (p[0][0] - cx) + (p[1][1] - dy) * (p[0][1] - cy)
              for p in pairs)
    angle = math.atan2(num, den) if (num or den) else 0.0
    cos, sin = math.cos(angle), math.sin(angle)
    return (cx - (dx * cos - dy * sin), cy - (dx * sin + dy * cos),
            math.degrees(angle))


def _refine(
    start: tuple[float, float],
    csv_points: Points,
    dxf_points: Points,
) -> tuple[Transform, list]:
    """Einen Kandidaten ausgleichen: zweimal zuordnen und ausgleichen."""
    transform = Transform(dx=start[0], dy=start[1])
    pairs: list = []
    for _ in range(2):
        pairs = _match(csv_points, dxf_points, transform, FIT_TOL)
        if len(pairs) < MIN_INLIERS:
            return transform, pairs
        fdx, fdy, angle = _fit(pairs)
        transform.dx, transform.dy = fdx, fdy
        transform.rotation = 0.0 if abs(angle) < 1e-9 else angle
    return transform, pairs


def derive_transform(
    csv_points: Points,
    dxf_points: Points,
    csv_file: str = "",
    dxf_file: str = "",
    tol: float = DEFAULT_MATCH_TOL,
) -> Transform:
    """
    Lagebezug ermitteln: abstimmen, mehrere Kandidaten ausgleichen, besten waehlen.

    csv_points / dxf_points sind je Rolle ("Separator", "Scanner") eine Liste
    von Punkten; zugeordnet wird nur innerhalb derselben Rolle.
    """
    total = sum(len(p) for p in csv_points.values())
    if not total:
        raise RegistrationError("CSV enthaelt keine punktartigen Objekte")
    if not sum(len(p) for p in dxf_points.values()):
        raise RegistrationError(
            "DXF enthaelt keine passenden Anker-Bloecke "
            f"({', '.join(n for names in ANCHOR_BLOCKS.values() for n in names)})"
        )

    scored = []
    for vdx, vdy, votes in _vote_offsets(csv_points, dxf_points):
        candidate, pairs = _refine((vdx, vdy), csv_points, dxf_points)
        if len(pairs) < MIN_INLIERS:
            continue
        mean = sum(d for _, _, d in pairs) / len(pairs)
        # Mehr exakte Treffer schlaegt kleineren Restfehler
        scored.append((len(pairs), -mean, votes, candidate, pairs))

    if not scored:
        raise RegistrationError(
            f"kein Kandidat erreichte {MIN_INLIERS} exakte Treffer "
            f"(Toleranz {FIT_TOL:.0f} mm) - Lagebezug nicht nachweisbar"
        )

    scored.sort(key=lambda s: (s[0], s[1]), reverse=True)
    best_n, _, votes, transform, pairs = scored[0]

    transform.source = "auto"
    transform.csv_file = csv_file
    transform.dxf_file = dxf_file
    ambiguous = ""
    if len(scored) > 1 and scored[1][0] == best_n:
        ambiguous = (f"; ACHTUNG mehrdeutig - Kandidat 2 hat ebenfalls {best_n} "
                     f"exakte Treffer (Restfehler {-scored[1][1]:.1f} mm)")
    transform.anchor = (
        f"{best_n} exakte Treffer (<= {FIT_TOL:.0f} mm) ueber die Rollen "
        f"{', '.join(f'{r}:{len(p)}' for r, p in sorted(csv_points.items()))}; "
        f"{votes} Stimmen im {VOTE_GRID:.0f}-mm-Raster, "
        f"{len(scored)} Kandidaten geprueft{ambiguous}"
    )

    verify_transform(transform, csv_points, dxf_points, tol)
    transform.created = datetime.now().strftime("%Y-%m-%d %H:%M")
    return transform


def verify_transform(
    transform: Transform,
    csv_points: Points,
    dxf_points: Points,
    tol: float = DEFAULT_MATCH_TOL,
) -> list[float]:
    """Restfehler der Zuordnung berechnen und in die Transformation schreiben."""
    pairs = _match(csv_points, dxf_points, transform, tol)
    residuals = sorted(d for _, _, d in pairs)
    transform.matched = len(pairs)
    transform.candidates = sum(len(p) for p in csv_points.values())
    transform.residual_max = max(residuals) if residuals else 0.0
    transform.residual_mean = (sum(residuals) / len(residuals)) if residuals else 0.0
    return residuals


# ---------------------------------------------------------------------------
# Persistenz in SKEL_CFG
# ---------------------------------------------------------------------------

REGISTRATION_FILE = "dxf_registration.json"


def registration_path(cfg_dir: Path) -> Path:
    return cfg_dir / REGISTRATION_FILE


def _key(csv_file: str, dxf_file: str) -> str:
    return f"{Path(csv_file).name}|{Path(dxf_file).name}"


def load_transform(cfg_dir: Path, csv_file: str, dxf_file: str) -> Transform | None:
    """Geprueften Lagebezug aus der Konfiguration lesen."""
    path = registration_path(cfg_dir)
    if not path.is_file():
        return None
    try:
        data = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError):
        return None
    entry = data.get(_key(csv_file, dxf_file))
    if not entry:
        return None
    transform = Transform.model_validate(entry)
    transform.source = "cfg"
    return transform


def save_transform(cfg_dir: Path, transform: Transform) -> Path:
    """Lagebezug in der Konfiguration ablegen."""
    cfg_dir.mkdir(parents=True, exist_ok=True)
    path = registration_path(cfg_dir)
    data = {}
    if path.is_file():
        try:
            data = json.loads(path.read_text(encoding="utf-8"))
        except (OSError, json.JSONDecodeError):
            data = {}
    stored = transform.model_dump()
    stored["source"] = "auto"
    data[_key(transform.csv_file, transform.dxf_file)] = stored
    path.write_text(json.dumps(data, indent=2, ensure_ascii=False) + "\n",
                    encoding="utf-8")
    return path