Schoenenberger_Systeme_GmbH/sps_skel
8
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add TRO annotation tool writing an annotated copy of the drawing

Browse Source 
Takes the original DXF, the CSV export and the derived TROs and writes a copy of
the drawing with the TRO information overlaid. The original is never touched.

Per TRO a block marker with attributes, not loose geometry: one block definition
per type (TRO_SYM_<type>), inserted at the plant coordinate, carrying TRO_ID,
TRO_TYPE, FB_BLOCK visibly and ITEMS, CONFIDENCE, SEPARATORS as hidden data.
That makes the result queryable and editable inside BricsCAD, and a corrected
TRO_ID can later feed back into the generator.

Layers follow the colour groups of tro_catalog (TRO_MARKER_sep, _vario, _store,
...) with the group colour as true colour and entities BYLAYER, so the layer
manager filters by TRO type. Flow arrows and the legend get their own layers.

Re-runs are idempotent through XDATA, not layer names: every generated entity is
tagged with the SPS_SKEL_TRO appid and a re-run deletes exactly those. Deleting
by layer would also destroy anything a user had put on the same layer.

Switches follow the other tools; --check reports the registration and what would
be drawn without writing, --flow adds the TRO-level arrows, --fb shows the
function block, --offset and --re-register override the stored transform. DWG is
rejected with a clear message since ezdxf cannot read it.

Verified against the real 23.8 MB drawing: 9 markers on the correct layers with
readable attributes, 15 flow arrows, and a re-run replaces its own 46 entities
instead of duplicating them.

Adds ezdxf to requirements and stores the verified Mubea transform in cfg/.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Samer Ayadi
2026-07-30 15:39:54 +02:00
parent 7a8c25f3b9
commit 977b5c826e
5 changed files with 663 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/tro_annotate.py
+560
View File
@@ -0,0 +1,560 @@
# -*- coding: utf-8 -*-
# Absichtlich ohne Shebang: der Windows-py-Launcher wuerde daraus "python3"
# ableiten und scheitert, wenn keine PythonCore-Installation registriert ist.
"""
tro_annotate.py - TRO-Informationen in die BricsCAD-Zeichnung einblenden.
Nimmt die Originalzeichnung (DXF), den CSV-Export und die daraus ermittelten
TROs und schreibt eine **Kopie** der Zeichnung mit zusaetzlicher Beschriftung:
* je TRO ein farbiger Marker als Block mit Attributen
* TRO-Kennung, TRO-Typ, optional der Siemens-Funktionsbaustein
* optional Pfeile fuer den gerichteten Materialfluss zwischen den TROs
* optional eine Legende
Die Originalzeichnung wird nie veraendert. Alle Ergaenzungen liegen auf eigenen
Layern (Vorgabe: Praefix "TRO_") und tragen XDATA der Anwendungskennung
SPS_SKEL_TRO. Ein erneuter Lauf entfernt genau diese Objekte und baut sie neu -
gelaeuscht wird nach XDATA, nicht nach Layername, damit eigene Objekte auf
denselben Layern unangetastet bleiben.
Marker als Block mit Attributen
-------------------------------
Je TRO-Typ wird eine Blockdefinition angelegt (TRO_SYM_<Typ>) und an der
Anlagenkoordinate eingefuegt. Die Daten stehen als Attribute am Block:
TRO_ID TRO03
TRO_TYPE 1Sep
FB_BLOCK FB_ILS_MTRO_1Sep
ITEMS 1x Separator
CONFIDENCE mittel
Damit sind sie in BricsCAD auswertbar und aenderbar - und ein korrigiertes
TRO_ID kann spaeter zurueck in die Generierung laufen.
Lagebezug
---------
CSV und Zeichnung stehen in verschiedenen Koordinatensystemen. Der Versatz wird
von lib/dxf_registration.py ermittelt, nachgewiesen und in SKEL_CFG abgelegt;
spaetere Laeufe verwenden den abgelegten Wert. --re-register erzwingt eine neue
Ermittlung, --offset setzt ihn von Hand.
Ausgabe (in SKEL_RESULTS)
-------------------------
<name>_annotated.dxf Kopie der Zeichnung mit Beschriftung
<name>_registration.json verwendeter Lagebezug samt Nachweis
Aufruf ueber bin/tro_annotate.bat bzw. bin/tro_annotate.sh.
Benoetigt ezdxf (siehe requirements.txt). DWG wird nicht unterstuetzt - die
Zeichnung muss als DXF vorliegen.
"""
from __future__ import annotations
import argparse
import collections
import csv as csv_module
import json
import math
import sys
from pathlib import Path
import ezdxf
from ezdxf.enums import TextEntityAlignment
from dxf_registration import (
ANCHOR_BLOCKS,
RegistrationError,
Transform,
derive_transform,
load_transform,
save_transform,
verify_transform,
)
from material_flow import build_graph, env_dir, read_elements, resolve_input
from tro_catalog import STYLES, get_tro
from tro_flow import Analysis, analyse_tros
# ---------------------------------------------------------------------------
# Konstanten
# ---------------------------------------------------------------------------
APPID = "SPS_SKEL_TRO"
LAYER_PREFIX = "TRO_"
BLOCK_PREFIX = "TRO_SYM_"
# Zeichnungsgroessen in mm, wenn nicht anders angegeben
DEFAULT_MARKER = 300.0
DEFAULT_TEXT = 220.0
# Restfehler, ab dem der Lagebezug abgelehnt wird
DEFAULT_MAX_RESIDUAL = 600.0
def hex_to_rgb(value: str) -> tuple[int, int, int]:
"""'#e8d4f0' -> (232, 212, 240)"""
text = value.lstrip("#")
return (int(text[0:2], 16), int(text[2:4], 16), int(text[4:6], 16))
# ---------------------------------------------------------------------------
# Lagebezug
# ---------------------------------------------------------------------------
def anchor_points_csv(elements) -> dict[str, list[tuple[float, float]]]:
"""Punktartige CSV-Objekte je Rolle (Separator, Scanner)."""
points: dict[str, list[tuple[float, float]]] = collections.defaultdict(list)
for element in elements:
if element.kind in ANCHOR_BLOCKS and element.position:
points[element.kind].append((element.x, element.y))
return dict(points)
def anchor_points_dxf(msp) -> dict[str, list[tuple[float, float]]]:
"""Einfuegepunkte der Anker-Bloecke je Rolle."""
wanted = {name: role for role, names in ANCHOR_BLOCKS.items() for name in names}
points: dict[str, list[tuple[float, float]]] = collections.defaultdict(list)
for ref in msp.query("INSERT"):
role = wanted.get(ref.dxf.name)
if role:
points[role].append((ref.dxf.insert.x, ref.dxf.insert.y))
return dict(points)
def resolve_transform(
args, elements, msp, cfg_dir: Path, csv_name: str, dxf_name: str,
warnings: list[str],
) -> Transform:
"""Lagebezug bestimmen: Kommandozeile, Konfiguration oder Ermittlung."""
csv_points = anchor_points_csv(elements)
dxf_points = anchor_points_dxf(msp)
if args.offset:
try:
dx, dy = (float(v) for v in args.offset.split(","))
except ValueError:
raise RegistrationError(
f"--offset '{args.offset}' nicht lesbar, erwartet 'dx,dy'"
) from None
transform = Transform(dx=dx, dy=dy, source="cli", csv_file=csv_name,
dxf_file=dxf_name, anchor="von Hand gesetzt")
verify_transform(transform, csv_points, dxf_points)
return transform
if not args.re_register:
stored = load_transform(cfg_dir, csv_name, dxf_name)
if stored:
# Auch der abgelegte Bezug wird gegen die Zeichnung nachgerechnet
verify_transform(stored, csv_points, dxf_points)
return stored
transform = derive_transform(csv_points, dxf_points, csv_name, dxf_name)
saved = save_transform(cfg_dir, transform)
warnings.append(f"Lagebezug neu ermittelt und abgelegt: {saved}")
return transform
# ---------------------------------------------------------------------------
# Zeichnung vorbereiten
# ---------------------------------------------------------------------------
def ensure_appid(doc) -> None:
if APPID not in doc.appids:
doc.appids.add(APPID)
def remove_previous(msp) -> int:
"""Objekte frueherer Laeufe entfernen - erkannt an ihren XDATA."""
doomed = []
for entity in msp:
try:
if entity.get_xdata(APPID):
doomed.append(entity)
except (ezdxf.DXFValueError, ezdxf.DXFTypeError):
continue
for entity in doomed:
msp.delete_entity(entity)
return len(doomed)
def tag(entity, tro_id: str = "") -> None:
"""Objekt als von diesem Programm erzeugt kennzeichnen."""
entity.set_xdata(APPID, [(1000, "tro_annotate"), (1000, tro_id or "-")])
def build_layers(doc, prefix: str) -> dict[str, str]:
"""Layer je Farbgruppe plus Label-, Fluss- und Legendenlayer."""
names: dict[str, str] = {}
for style in STYLES:
name = f"{prefix}MARKER_{style.group}"
names[style.group] = name
if name not in doc.layers:
doc.layers.add(name, dxfattribs={
"true_color": ezdxf.rgb2int(hex_to_rgb(style.fill)),
})
for key, colour in (("label", "#404040"), ("flow", "#2f5597"),
("legend", "#808080")):
name = f"{prefix}{key.upper()}"
names[key] = name
if name not in doc.layers:
doc.layers.add(name, dxfattribs={
"true_color": ezdxf.rgb2int(hex_to_rgb(colour)),
})
return names
def build_symbol(doc, type_name: str, marker: float, text: float, show_fb: bool) -> str:
"""Blockdefinition fuer einen TRO-Typ anlegen (Marker + Attributdefinitionen)."""
name = f"{BLOCK_PREFIX}{type_name}"
if name in doc.blocks:
return name
blk = doc.blocks.new(name=name)
# Marker: Ring mit Mittelpunkt, Farbe kommt vom Layer des INSERT
blk.add_circle((0, 0), marker, dxfattribs={"color": 256})
blk.add_circle((0, 0), marker * 0.18, dxfattribs={"color": 256})
blk.add_line((-marker, 0), (marker, 0), dxfattribs={"color": 256})
blk.add_line((0, -marker), (0, marker), dxfattribs={"color": 256})
gap = marker * 1.35
rows = [
("TRO_ID", type_name and "TRO??", text * 1.35, 0),
("TRO_TYPE", type_name, text, 0),
]
if show_fb:
rows.append(("FB_BLOCK", "FB", text * 0.85, 0))
offset = 0.0
for tag_name, default, height, _ in rows:
blk.add_attdef(
tag_name,
insert=(gap, offset),
height=height,
dxfattribs={"style": "Standard", "color": 256, "text": default},
)
offset -= height * 1.55
# nicht sichtbare Attribute: reine Daten
for tag_name in ("ITEMS", "CONFIDENCE", "SEPARATORS"):
blk.add_attdef(
tag_name,
insert=(gap, offset),
height=text * 0.7,
dxfattribs={"style": "Standard", "color": 256, "text": "",
"invisible": 1},
)
offset -= text
return name
# ---------------------------------------------------------------------------
# Beschriftung zeichnen
# ---------------------------------------------------------------------------
def place_markers(
msp, doc, analysis: Analysis, transform: Transform, layers: dict[str, str],
marker: float, text: float, show_fb: bool, warnings: list[str],
) -> dict[str, tuple[float, float]]:
"""Je TRO einen Blockmarker mit Attributen einfuegen."""
placed: dict[str, tuple[float, float]] = {}
for tro in analysis.tros:
if not tro.position:
warnings.append(f"{tro.tro_id}: keine Koordinate - nicht beschriftet")
continue
x, y = transform.to_dxf(tro.position[0], tro.position[1])
placed[tro.tro_id] = (x, y)
definition = get_tro(tro.type_name)
group = definition.get_group() if definition else "ext"
block = build_symbol(doc, tro.type_name, marker, text, show_fb)
values = {
"TRO_ID": tro.tro_id,
"TRO_TYPE": tro.type_name,
"ITEMS": ", ".join(f"{c}x {i}" for i, c in sorted(tro.items.items())),
"CONFIDENCE": tro.confidence,
"SEPARATORS": ", ".join(tro.separators),
}
if show_fb:
values["FB_BLOCK"] = tro.fb_block or "-"
ref = msp.add_blockref(block, (x, y), dxfattribs={
"layer": layers.get(group, layers["ext"]),
})
ref.add_auto_attribs(values)
tag(ref, tro.tro_id)
return placed
def draw_flow(
msp, analysis: Analysis, placed: dict[str, tuple[float, float]],
layers: dict[str, str], marker: float,
) -> int:
"""Pfeile fuer den gerichteten Materialfluss zwischen den TROs."""
head = marker * 0.9
drawn = 0
for src, dst in analysis.edges:
if src not in placed or dst not in placed:
continue
x1, y1 = placed[src]
x2, y2 = placed[dst]
length = math.dist((x1, y1), (x2, y2))
if length <= 2 * marker * 1.1:
continue
ux, uy = (x2 - x1) / length, (y2 - y1) / length
# An beiden Enden um den Marker kuerzen, damit der Pfeil ihn nicht deckt
sx, sy = x1 + ux * marker * 1.1, y1 + uy * marker * 1.1
ex, ey = x2 - ux * marker * 1.1, y2 - uy * marker * 1.1
shaft = msp.add_lwpolyline([(sx, sy), (ex, ey)],
dxfattribs={"layer": layers["flow"]})
tag(shaft, f"{src}->{dst}")
# Pfeilspitze als gefuellte Flaeche
bx, by = ex - ux * head, ey - uy * head
px, py = -uy * head * 0.38, ux * head * 0.38
point = msp.add_solid(
[(ex, ey), (bx + px, by + py), (bx - px, by - py)],
dxfattribs={"layer": layers["flow"]},
)
tag(point, f"{src}->{dst}")
drawn += 1
return drawn
def draw_legend(
msp, analysis: Analysis, placed: dict[str, tuple[float, float]],
layers: dict[str, str], marker: float, text: float, source: str,
) -> None:
"""Legende links unterhalb der beschrifteten Objekte."""
if not placed:
return
x0 = min(p[0] for p in placed.values()) - marker * 12
y0 = min(p[1] for p in placed.values()) - marker * 6
row = text * 2.2
title = msp.add_text(
f"TRO-Beschriftung - {source}",
height=text * 1.5,
dxfattribs={"layer": layers["legend"], "style": "Standard"},
)
title.set_placement((x0, y0 + row))
tag(title)
used = {t.type_name for t in analysis.tros}
groups = []
for style in STYLES:
types = sorted(n for n in used
if (get_tro(n).get_group() if get_tro(n) else "ext") == style.group)
if types:
groups.append((style, types))
for index, (style, types) in enumerate(groups):
y = y0 - index * row
box = msp.add_solid(
[(x0, y), (x0 + text, y), (x0 + text, y - text), (x0, y - text)],
dxfattribs={"layer": layers[style.group]},
)
tag(box)
label = msp.add_text(
f"{style.label}: {', '.join(types)}",
height=text,
dxfattribs={"layer": layers["legend"], "style": "Standard"},
)
label.set_placement((x0 + text * 1.8, y - text))
tag(label)
# ---------------------------------------------------------------------------
# Bericht
# ---------------------------------------------------------------------------
def report(
analysis: Analysis, transform: Transform, csv_file: Path, dxf_file: Path,
out_file: Path | None, reg_file: Path | None, counts: dict, warnings: list[str],
) -> None:
print("")
print("================================================================")
print("TRO-BESCHRIFTUNG")
print("================================================================")
print(f"CSV = {csv_file}")
print(f"Zeichnung = {dxf_file}")
print(f"Lagebezug = {transform.describe()}")
print(f" Herkunft = {transform.source}"
+ (f", ermittelt {transform.created}" if transform.created else ""))
print(f" Nachweis = {transform.anchor}")
print(f" Restfehler = max {transform.residual_max:.1f} mm, "
f"mittel {transform.residual_mean:.1f} mm "
f"({transform.matched}/{transform.candidates} Punkte)")
print(f"TROs = {len(analysis.tros)}, beschriftet: {counts['marker']}")
print(f"Flusspfeile = {counts['flow']}")
print(f"Entfernte Objekte= {counts['removed']} (aus frueherem Lauf)")
print(f"Layer = {counts['layers']}, Bloecke: {counts['blocks']}")
print(f"DXF = {out_file if out_file else '- (--check)'}")
print(f"Lagebezug-Datei = {reg_file if reg_file else '-'}")
print("================================================================")
if warnings:
print("")
print(f"Hinweise ({len(warnings)}):")
for warning in warnings:
print(f" ! {warning}")
print("")
# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------
def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
parser = argparse.ArgumentParser(
prog="tro_annotate",
description="Blendet die ermittelten TROs in eine Kopie der "
"BricsCAD-Zeichnung ein (DXF).",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="Beispiele:\n"
" tro_annotate.bat --file mubea.csv --dxf 500573_60_1.dxf --check\n"
" tro_annotate.bat --file mubea.csv --dxf 500573_60_1.dxf --flow --fb\n"
"\n"
"Exit-Codes:\n"
" 0 ok\n"
" 1 Eingabe- oder Aufruffehler\n"
" 4 Lagebezug nicht nachweisbar\n",
)
parser.add_argument("--file", default="export.csv", metavar="NAME",
help="CSV-Export in %%SKEL_DATA%% oder ein Pfad. "
"Standard: %(default)s")
parser.add_argument("--dxf", required=True, metavar="NAME",
help="Originalzeichnung als DXF (Pfad oder Name in "
"%%SKEL_DATA%%).")
parser.add_argument("--out", metavar="NAME",
help="Name der Ausgabedatei in %%SKEL_RESULTS%%. "
"Standard: <csv>_annotated.dxf")
parser.add_argument("--flow", action="store_true",
help="Pfeile fuer den Materialfluss zwischen den TROs "
"zeichnen.")
parser.add_argument("--fb", action="store_true",
help="Den Siemens-Funktionsbaustein als sichtbares "
"Attribut am Marker anzeigen.")
parser.add_argument("--legend", action="store_true",
help="Legende der Farbgruppen einzeichnen.")
parser.add_argument("--marker-size", type=float, default=DEFAULT_MARKER,
metavar="MM", help="Markerradius in mm. Standard: %(default)s")
parser.add_argument("--text-height", type=float, default=DEFAULT_TEXT,
metavar="MM", help="Texthoehe in mm. Standard: %(default)s")
parser.add_argument("--layer-prefix", default=LAYER_PREFIX, metavar="P",
help="Praefix der erzeugten Layer. Standard: %(default)s")
parser.add_argument("--offset", metavar="DX,DY",
help="Lagebezug von Hand setzen statt ermitteln.")
parser.add_argument("--re-register", action="store_true",
help="Lagebezug neu ermitteln, auch wenn in %%SKEL_CFG%% "
"einer abgelegt ist.")
parser.add_argument("--max-residual", type=float, default=DEFAULT_MAX_RESIDUAL,
metavar="MM",
help="Groesster zugelassener Restfehler des Lagebezugs. "
"Standard: %(default)s")
parser.add_argument("--check", action="store_true",
help="Nur pruefen und berichten, keine Datei schreiben.")
return parser.parse_args(argv)
def main(argv: list[str] | None = None) -> int:
args = parse_args(argv)
warnings: list[str] = []
csv_file = resolve_input(args.file)
if not csv_file.is_file():
print(f"FEHLER: CSV nicht gefunden: {csv_file}", file=sys.stderr)
return 1
dxf_file = Path(args.dxf)
if not dxf_file.is_absolute() and dxf_file.parent == Path("."):
dxf_file = env_dir("SKEL_DATA", "data") / args.dxf
if not dxf_file.is_file():
print(f"FEHLER: Zeichnung nicht gefunden: {dxf_file}", file=sys.stderr)
return 1
if dxf_file.suffix.lower() == ".dwg":
print("FEHLER: DWG wird nicht unterstuetzt - bitte als DXF exportieren.",
file=sys.stderr)
return 1
results = env_dir("SKEL_RESULTS", "results")
cfg_dir = env_dir("SKEL_CFG", "cfg")
results.mkdir(parents=True, exist_ok=True)
# TROs aus dem CSV
read_warnings: list[str] = []
try:
elements = read_elements(csv_file, read_warnings)
except (OSError, ValueError, csv_module.Error) as exc:
print(f"FEHLER: {csv_file} nicht lesbar: {exc}", file=sys.stderr)
return 1
graph = build_graph(elements, csv_file.name, read_warnings)
analysis = analyse_tros(graph, elements)
if not analysis.tros:
print(f"FEHLER: keine TROs aus {csv_file} ableitbar.", file=sys.stderr)
return 1
# Zeichnung
try:
doc = ezdxf.readfile(str(dxf_file))
except (OSError, ezdxf.DXFError) as exc:
print(f"FEHLER: {dxf_file} nicht lesbar: {exc}", file=sys.stderr)
return 1
msp = doc.modelspace()
# Lagebezug
try:
transform = resolve_transform(args, elements, msp, cfg_dir,
csv_file.name, dxf_file.name, warnings)
except RegistrationError as exc:
print(f"FEHLER: Lagebezug nicht bestimmbar: {exc}", file=sys.stderr)
return 4
if transform.residual_max > args.max_residual:
print(f"FEHLER: Restfehler des Lagebezugs {transform.residual_max:.1f} mm "
f"ueber der Grenze von {args.max_residual:.1f} mm.", file=sys.stderr)
print(f" {transform.describe()}", file=sys.stderr)
print(" Mit --offset von Hand setzen oder --max-residual anheben.",
file=sys.stderr)
return 4
counts = {"marker": 0, "flow": 0, "removed": 0, "layers": 0, "blocks": 0}
if args.check:
counts["marker"] = sum(1 for t in analysis.tros if t.position)
report(analysis, transform, csv_file, dxf_file, None, None, counts, warnings)
return 0
ensure_appid(doc)
counts["removed"] = remove_previous(msp)
layers = build_layers(doc, args.layer_prefix)
counts["layers"] = len({v for v in layers.values()})
placed = place_markers(msp, doc, analysis, transform, layers,
args.marker_size, args.text_height, args.fb, warnings)
counts["marker"] = len(placed)
counts["blocks"] = sum(1 for b in doc.blocks if b.name.startswith(BLOCK_PREFIX))
if args.flow:
counts["flow"] = draw_flow(msp, analysis, placed, layers, args.marker_size)
if args.legend:
draw_legend(msp, analysis, placed, layers, args.marker_size,
args.text_height, csv_file.name)
out_file = results / (args.out or f"{csv_file.stem}_annotated.dxf")
doc.saveas(str(out_file))
reg_file = results / f"{csv_file.stem}_registration.json"
reg_file.write_text(
json.dumps(transform.model_dump(), indent=2, ensure_ascii=False) + "\n",
encoding="utf-8",
)
report(analysis, transform, csv_file, dxf_file, out_file, reg_file,
counts, warnings)
return 0
if __name__ == "__main__":
sys.exit(main())