diff --git a/lib/dxf_registration.py b/lib/dxf_registration.py new file mode 100644 index 0000000..765dd9e --- /dev/null +++ b/lib/dxf_registration.py @@ -0,0 +1,357 @@ +# -*- 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