From 7a8c25f3b970b595ef388ba8406992ff038b0188 Mon Sep 17 00:00:00 2001 From: Samer Ayadi Date: Thu, 30 Jul 2026 15:39:54 +0200 Subject: [PATCH] Add DXF registration with a provable coordinate transform The CSV export and the BricsCAD drawing sit in different coordinate systems. A silent offset here puts every annotation in the wrong place while still looking plausible, so this module derives the transform, proves it, and refuses when it cannot. Approach: vote for the translation over point pairs, refine the best candidates by Procrustes fit (rotation included), pick the one with the most exact matches, then report the residuals. Two things the naive version got wrong, both fixed here: Matching is constrained by role. The plant is built from regular grids - the store lines, their infeeds and their outfeeds each form a row of 20 points at 800 mm pitch. Voting over all points is therefore ambiguous: a wrong offset hits the infeed row exactly as well as the right one hits the separator row, and it won by a single vote (21 to 20), producing an offset 8.6 m out with a spurious 0.52 degree rotation. Separators are now matched only against separator blocks (S-SP, S-LP) and scanners only against SCAN, which makes the result unique. The fit uses exact matches only (50 mm). For extended objects the CSV reports the bounding-box centre while the block reports its insertion point; such pairs are 100-500 mm apart and would tilt the fit if they were included. They still count towards the reported residuals. For the Mubea plant this yields dx +50905.7 mm, dy +8414.5 mm, rotation exactly 0, from 20 exact matches - the same values a hand analysis produced. A verified transform is stored in SKEL_CFG/dxf_registration.json per CSV/DXF pair, so later runs are reproducible; a stored transform is still re-checked against the drawing on every run. Co-Authored-By: Claude Opus 5 (1M context) --- lib/dxf_registration.py | 357 ++++++++++++++++++++++++++++++++++++++++ 1 file changed, 357 insertions(+) create mode 100644 lib/dxf_registration.py 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