Files
sps_skel/lib/dxf_registration.py
T
s.ayadi 9ec1022fb1 tro_annotate: Lagebezug ueber die TeileId, Ablaufbeschreibung CSV/DXF -> SCL
dxf_registration.derive_from_ids: Zeichnungen aus dxfmakros tragen an jedem
Layout-Block das Attribut ID = TeileId des CSV. tro_annotate paart CSV-
Separatoren mit den Separator-Bloecken gleicher ID (sonst Scanner) und nimmt
den Median-Versatz - eindeutig statt Abstimmung, bei jedem Lauf neu, gilt
auch fuer eine verschobene Kopie der Zeichnung. Vorrang vor abgelegtem und
abgestimmtem Bezug; --offset bleibt. Parameter id_spread_mm, id_min_share in
cfg/parameter.ini [dxf_registration].

Dematic: gemeinsame Zeichnung 0/0, Mechanik-Zeichnung dematic.dxf
+466,1/-163549,3 mm. Die Abstimmung rastete vorher auf den Scanner-Bloecken
ein (Einfuegepunkt 360 mm neben der CSV-Position) und lag 360 mm daneben.

Gesperrte Ausgabedatei (in BricsCAD geoeffnet) gibt eine Meldung statt
Traceback. Neu: doc/Workflow_CSV_DXF_to_SCL.md - der ganze Ablauf vom
CSV-Export und der Zeichnung bis zur SCL.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-23 14:59:16 +02:00

443 lines
18 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 hashlib
import json
import math
from collections import Counter
from datetime import datetime
from pathlib import Path
from pydantic import BaseModel, ConfigDict, Field
from parameter import get_float, get_int
__all__ = [
"Transform",
"RegistrationError",
"ANCHOR_BLOCKS",
"derive_transform",
"derive_from_ids",
"verify_transform",
"load_transform",
"save_transform",
"registration_path",
"file_fingerprint",
]
# 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_*"),
}
# Zahlenwerte aus cfg/parameter.ini, Abschnitt [dxf_registration] - Bedeutung
# und Einheit stehen dort (lib/parameter.py).
_P = "dxf_registration"
VOTE_GRID = get_float(_P, "vote_grid")
VOTE_CANDIDATES = get_int(_P, "vote_candidates")
FIT_TOL = get_float(_P, "fit_tol")
DEFAULT_MATCH_TOL = get_float(_P, "default_match_tol")
MIN_INLIERS = get_int(_P, "min_inliers")
ROTATION_WARN_DEG = get_float(_P, "rotation_warn_deg")
ID_SPREAD_MM = get_float(_P, "id_spread_mm")
ID_MIN_SHARE = get_float(_P, "id_min_share")
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="")
# Pruefsumme der Zeichnung/E-Datei beim Ermitteln ("" = unbekannt). Der
# Schluessel enthaelt nur Dateinamen; ein neuer Stand gleichen Namens wird
# daran erkannt (siehe load_transform).
fingerprint: 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 derive_from_ids(
csv_by_id: dict[str, tuple[str, float, float]],
dxf_by_id: dict[str, tuple[str, float, float]],
csv_file: str = "",
dxf_file: str = "",
) -> Transform | None:
"""
Lagebezug ueber die TeileId: CSV-Objekt und DXF-Block mit derselben ID.
Mit dxfmakros gebaute Zeichnungen tragen an jedem Layout-Block das
Attribut ID - dieselbe Nummer, die der CSV-Export als TeileId fuehrt. Das
ist eine eindeutige Zuordnung, keine Abstimmung ueber Raster, und gilt
genauso fuer eine verschobene Kopie der Zeichnung (nur Mechanik als
Hintergrund) wie fuer die Zeichnung, aus der das CSV stammt.
Parameter:
csv_by_id: TeileId -> (Rolle, x, y) aus dem CSV (Rolle wie in
ANCHOR_BLOCKS: "Separator", "Scanner").
dxf_by_id: ID-Attribut -> (Rolle, x, y) der Bloecke im DXF.
Rueckgabe:
Transform (DXF -> CSV, ohne Drehung) oder None, wenn keine Rolle
genug uebereinstimmende Paare hat. Die Rollen werden in der
Reihenfolge von ANCHOR_BLOCKS versucht: der Separator-Block sitzt
fast genau auf der CSV-Position (Dematic: 60 mm), der Scanner-Block
360 mm daneben. Ausgedehnte Objekte (Kreisel, Strecken) zaehlen
nicht - ihr Einfuegepunkt liegt je nach Lage weit vom CSV-Mittelpunkt.
Eine Rolle gilt, wenn mindestens MIN_INLIERS Paare und ID_MIN_SHARE
aller Paare hoechstens ID_SPREAD_MM vom Median entfernt liegen.
"""
for rolle in ANCHOR_BLOCKS:
paare = [(c[1] - d[1], c[2] - d[2])
for i, c in csv_by_id.items()
if c[0] == rolle and (d := dxf_by_id.get(i)) is not None and d[0] == rolle]
if len(paare) < MIN_INLIERS:
continue
mdx = sorted(p[0] for p in paare)[len(paare) // 2]
mdy = sorted(p[1] for p in paare)[len(paare) // 2]
passend = [p for p in paare if math.hypot(p[0] - mdx, p[1] - mdy) <= ID_SPREAD_MM]
if len(passend) < MIN_INLIERS or len(passend) < ID_MIN_SHARE * len(paare):
continue
return Transform(
dx=mdx, dy=mdy, source="ids", csv_file=csv_file, dxf_file=dxf_file,
anchor=(f"TeileId = ID-Attribut: {len(paare)} {rolle}-Paare, "
f"{len(passend)} davon hoechstens {ID_SPREAD_MM:.0f} mm um den Median"),
created=datetime.now().strftime("%Y-%m-%d %H:%M"),
)
return None
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 file_fingerprint(path: Path) -> str:
"""SHA-256 einer Datei (Hex) - Kennung des Planungsstands."""
digest = hashlib.sha256()
with path.open("rb") as handle:
for chunk in iter(lambda: handle.read(1 << 20), b""):
digest.update(chunk)
return digest.hexdigest()
def load_transform(cfg_dir: Path, csv_file: str, dxf_file: str,
fingerprint: str = "") -> Transform | None:
"""Geprueften Lagebezug aus der Konfiguration lesen.
Mit ``fingerprint``: nur verwenden, wenn der abgelegte Bezug fuer genau
diesen Dateistand ermittelt wurde. Ein Eintrag ohne Pruefsumme (aeltere
Ablage) oder mit anderer gilt als veraltet -> None, der Aufrufer ermittelt
neu. Ohne ``fingerprint`` (z. B. tro_annotate) wie bisher nur ueber den
Dateinamen.
"""
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
if fingerprint and entry.get("fingerprint", "") != fingerprint:
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