366 lines
14 KiB
Python
366 lines
14 KiB
Python
# -*- 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, als Muster
|
|
# (fnmatch, Gross-/Kleinschreibung egal). Es gibt zwei Namenswelten, und beide
|
|
# kommen im Projekt vor:
|
|
# - Original-Kundenzeichnung (z.B. 500573_60_1.dxf): S-SP / S-LP / SCAN
|
|
# - von den dxfmakros-Makros erzeugte Zeichnung (z.B. Mubea.dxf): die flachen
|
|
# Blocknamen mit Dimensionssuffix, also Separator_SP_2D/_3D, Scanner_2D/_3D
|
|
# Bewusst NICHT aufgenommen: Staustrecke_Separator_SP_*. Das ist ein Streckenmodul
|
|
# mit eingebautem Separator - sein Einfuegepunkt ist der Modulursprung, nicht der
|
|
# Separator, was einen zweiten, um genau diesen Versatz verschobenen Cluster in
|
|
# die Abstimmung braechte.
|
|
ANCHOR_BLOCKS: dict[str, tuple[str, ...]] = {
|
|
"Separator": ("S-SP", "S-LP", "Separator_SP*"),
|
|
"Scanner": ("SCAN", "Scanner_*"),
|
|
}
|
|
|
|
# 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
|