# -*- coding: utf-8 -*- # Absichtlich ohne Shebang: der Windows-py-Launcher wuerde daraus "python3" # ableiten und scheitert, wenn keine PythonCore-Installation registriert ist. """ tro_extract.py - TROs aus der annotierten Zeichnung zurueck nach JSON lesen. Gegenrichtung zu tro_annotate.py: tro_annotate.py CSV + Ableitung -> Marker in der Zeichnung tro_extract.py Zeichnung -> JSON Gelesen wird ausschliesslich die Zeichnung, nicht der CSV-Export. Damit kommt zurueck, was tatsaechlich in der Zeichnung steht - also auch das, was in BricsCAD von Hand nachgezogen wurde (Marker verschoben, Typ ueber TRO_EDIT geaendert). Genau dafuer ist der Weg gedacht: die Zeichnung ist nach dem Review der Stand der Dinge, nicht der Export. Erkannt werden die Objekte, die tro_annotate.py mit seinen XDATA (AppID SPS_SKEL_TRO) gekennzeichnet hat: Blockreferenz TRO_SYM_ ein TRO, Attribute ID/TYPE/ITEMS/ CONFIDENCE/SEPARATORS (+ FB_BLOCK mit --fb) LWPOLYLINE "->" ein Flusspfeil = eine Verbindung Kreis/Linien "OFFEN:" ein offener Punkt (noch kein TRO) Die Koordinaten stehen doppelt in der Datei: `positionDxf` ist die Zeichnungskoordinate, `position` die Anlagenkoordinate. Umgerechnet wird mit dem Lagebezug aus _registration.json (von tro_annotate.py geschrieben); fehlt der, bleibt `position` leer statt geraten zu werden. Aufruf ueber bin/tro_extract.bat bzw. bin/tro_extract.sh. """ from __future__ import annotations import argparse import json import re import sys from collections import Counter from datetime import datetime from pathlib import Path import ezdxf from dxf_registration import Transform from material_flow import env_dir from tro_catalog import get_tro APPID = "SPS_SKEL_TRO" BLOCK_PREFIX = "TRO_SYM_" OPEN_PREFIX = "OFFEN:" # XDATA-Nutzlast eines Flusspfeils: "->" EDGE_RE = re.compile(r"^(?P[^>]+)->(?P.+)$") # "1x SSCC, 1x Scanner, 1x Separator" -> {"SSCC": 1, "Scanner": 1, ...} ITEM_RE = re.compile(r"(?P\d+)\s*x\s*(?P.+)") def xdata_payload(entity) -> str | None: """Zweiter String der XDATA von tro_annotate, sonst None. Aufbau: (1000, "tro_annotate"), (1000, ). Die Nutzlast ist je nach Objekt die TRO-ID, "->" oder "OFFEN:". """ try: tags = entity.get_xdata(APPID) except (ezdxf.DXFValueError, ezdxf.DXFTypeError, ezdxf.DXFError): return None strings = [str(value) for code, value in tags if code == 1000] if len(strings) < 2 or strings[0] != "tro_annotate": return None return strings[1] def parse_items(text: str) -> dict[str, int]: """Bauteilliste aus dem ITEMS-Attribut zurueckgewinnen.""" items: dict[str, int] = {} for part in (text or "").split(","): match = ITEM_RE.match(part.strip()) if match: items[match.group("item").strip()] = int(match.group("count")) return items def attrib(ref, tag: str) -> str: """Attributwert einer Blockreferenz (leerer String, wenn nicht da).""" try: return (ref.get_attrib_text(tag) or "").strip() except Exception: return "" def load_transform(dxf_file: Path, results: Path, warnings: list[str]) -> Transform | None: """Lagebezug zu dieser Zeichnung suchen. tro_annotate.py benennt die Datei nach dem CSV-Stamm (export_registration .json), die Zeichnung heisst aber _annotated.dxf - daher wird das Suffix "_annotated" zusaetzlich ohne Endung probiert. Zuletzt entscheidet der Inhalt: eine Registrierung, deren `dxf_file` auf diese Zeichnung passt, gilt auch dann, wenn der Dateiname anders lautet. """ stem = dxf_file.stem namen = [stem] if stem.endswith("_annotated"): namen.append(stem[: -len("_annotated")]) kandidaten = [results / f"{n}_registration.json" for n in namen] kandidaten += [dxf_file.with_name(f"{n}_registration.json") for n in namen] kandidaten += sorted(results.glob("*_registration.json")) for candidate in kandidaten: if not candidate.is_file(): continue try: roh = json.loads(candidate.read_text(encoding="utf-8")) transform = Transform(**roh) except (OSError, ValueError, TypeError) as exc: warnings.append(f"{candidate.name} nicht lesbar ({exc})") continue # Bei den aus dem Glob gefundenen Dateien pruefen, ob sie ueberhaupt # zu dieser Zeichnung gehoeren - sonst waere der Bezug frei erfunden. passt = candidate.stem.split("_registration")[0] in namen if not passt and Path(roh.get("dxf_file", "")).stem not in (stem, *namen): continue return transform warnings.append( f"kein Lagebezug gefunden (gesucht: " f"{', '.join(n + '_registration.json' for n in namen)}) - " f"'position' bleibt leer, 'positionDxf' ist trotzdem gefuellt" ) return None def extract(doc, transform: Transform | None, warnings: list[str]) -> dict: """TROs, Verbindungen und offene Punkte aus dem Modelbereich lesen.""" msp = doc.modelspace() tros: list[dict] = [] edges: list[dict] = [] open_points: list[dict] = [] seen_edges: set[tuple[str, str]] = set() for entity in msp: payload = xdata_payload(entity) if payload is None: continue if entity.dxftype() == "INSERT" and entity.dxf.name.startswith(BLOCK_PREFIX): tro_id = attrib(entity, "ID") or payload type_name = attrib(entity, "TYPE") or entity.dxf.name[len(BLOCK_PREFIX):] insert = entity.dxf.insert eintrag = { "id": tro_id, "type": type_name, "items": parse_items(attrib(entity, "ITEMS")), "confidence": attrib(entity, "CONFIDENCE"), "separators": [s.strip() for s in attrib(entity, "SEPARATORS").split(",") if s.strip()], "positionDxf": {"x": round(insert.x, 2), "y": round(insert.y, 2), "z": round(insert.z, 2)}, "layer": entity.dxf.layer, "block": entity.dxf.name, } # FB-Baustein: als Attribut nur vorhanden, wenn mit --fb annotiert # wurde. Sonst aus dem Katalog nachgeschlagen und als solches # gekennzeichnet - er steht dann nicht in der Zeichnung. fb = attrib(entity, "FB_BLOCK") if fb and fb != "-": eintrag["fbType"] = fb eintrag["fbTypeFrom"] = "dxf" else: definition = get_tro(type_name) if definition: eintrag["fbType"] = definition.get_fb_block() eintrag["fbTypeFrom"] = "tro_catalog" else: eintrag["fbType"] = None eintrag["fbTypeFrom"] = "unbekannter Typ" warnings.append( f"{tro_id}: Typ '{type_name}' steht nicht im TRO-Katalog" ) if transform is not None: x, y = transform.to_csv(insert.x, insert.y) eintrag["position"] = {"x": round(x, 2), "y": round(y, 2)} tros.append(eintrag) continue if payload.startswith(OPEN_PREFIX): name = payload[len(OPEN_PREFIX):] if entity.dxftype() != "CIRCLE": continue # Kreis, Kreuz und Text tragen dieselbe Nutzlast center = entity.dxf.center punkt = {"name": name, "positionDxf": {"x": round(center.x, 2), "y": round(center.y, 2)}} if transform is not None: x, y = transform.to_csv(center.x, center.y) punkt["position"] = {"x": round(x, 2), "y": round(y, 2)} open_points.append(punkt) continue match = EDGE_RE.match(payload) if match: # Schaft und Spitze eines Pfeils tragen dieselbe Nutzlast pair = (match.group("src"), match.group("dst")) if pair not in seen_edges: seen_edges.add(pair) edges.append({"from": pair[0], "to": pair[1]}) tros.sort(key=lambda t: t["id"]) edges.sort(key=lambda e: (e["from"], e["to"])) open_points.sort(key=lambda p: p["name"]) # Vorgaenger/Nachfolger je TRO - das ist die Frage, die man an so eine # Datei stellt ("was kommt vor diesem TRO, was danach"). bekannt = {t["id"] for t in tros} for tro in tros: tro["predecessors"] = sorted(e["from"] for e in edges if e["to"] == tro["id"]) tro["successors"] = sorted(e["to"] for e in edges if e["from"] == tro["id"]) for edge in edges: for seite in ("from", "to"): if edge[seite] not in bekannt: warnings.append( f"Verbindung {edge['from']} -> {edge['to']}: '{edge[seite]}' " f"ist kein TRO in dieser Zeichnung" ) return {"tros": tros, "connections": edges, "openPoints": open_points} def main(argv: list[str] | None = None) -> int: args = parse_args(argv) warnings: list[str] = [] results = env_dir("SKEL_RESULTS", "results") dxf_file = Path(args.dxf) if not dxf_file.is_absolute() and dxf_file.parent == Path("."): dxf_file = results / args.dxf if not dxf_file.is_file(): print(f"FEHLER: Zeichnung nicht gefunden: {dxf_file}", file=sys.stderr) return 1 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 transform = load_transform(dxf_file, results, warnings) daten = extract(doc, transform, warnings) if not daten["tros"]: print(f"FEHLER: keine TRO-Marker in {dxf_file} gefunden. Wurde sie mit " f"tro_annotate.py erzeugt?", file=sys.stderr) return 2 kopf = { "schema": "sps_skel-tro-extract-v1", "source": dxf_file.name, "extracted": datetime.now().strftime("%Y-%m-%d %H:%M"), "note": "Aus der Zeichnung gelesen, nicht aus dem CSV-Export - " "Aenderungen aus BricsCAD sind enthalten.", } if transform is not None: kopf["transform"] = {"dx": round(transform.dx, 3), "dy": round(transform.dy, 3), "rotation": transform.rotation, "note": "Anlagenkoordinate = Zeichnung + (dx, dy)"} out_file = results / (args.out or f"{dxf_file.stem}_tro.json") out_file.write_text( json.dumps({**kopf, **daten}, indent=2, ensure_ascii=False) + "\n", encoding="utf-8", ) typen = Counter(t["type"] for t in daten["tros"]) print("") print("================================================================") print("TRO-EXTRAKT") print("================================================================") print(f"Zeichnung = {dxf_file}") print(f"Lagebezug = " + (transform.describe() if transform else "- (nur DXF-Koordinaten)")) print(f"TROs = {len(daten['tros'])} " f"({', '.join(f'{k}: {v}' for k, v in sorted(typen.items()))})") print(f"Verbindungen = {len(daten['connections'])}") print(f"Offene Punkte = {len(daten['openPoints'])}") print(f"JSON = {out_file}") print("================================================================") if warnings: print("") print(f"Hinweise ({len(warnings)}):") for warning in warnings: print(f" ! {warning}") print("") return 0 def parse_args(argv: list[str] | None = None) -> argparse.Namespace: parser = argparse.ArgumentParser( prog="tro_extract", description="Liest die von tro_annotate.py gesetzten TRO-Marker aus " "einer Zeichnung zurueck und schreibt sie als JSON.", formatter_class=argparse.RawDescriptionHelpFormatter, epilog="Beispiele:\n" " tro_extract.bat\n" " tro_extract.bat --dxf export_annotated.dxf\n" " tro_extract.bat --dxf export_annotated.dxf --out tros.json\n" "\n" "Exit-Codes:\n" " 0 ok\n" " 1 Eingabe- oder Aufruffehler\n" " 2 keine TRO-Marker in der Zeichnung\n", ) parser.add_argument( "--dxf", default="export_annotated.dxf", metavar="NAME", help="Annotierte Zeichnung in %%SKEL_RESULTS%% (oder ein vollstaendiger " "Pfad). Standard: %(default)s", ) parser.add_argument( "--out", metavar="NAME", help="Name der JSON-Datei in %%SKEL_RESULTS%%. " "Standard: _tro.json", ) return parser.parse_args(argv) if __name__ == "__main__": sys.exit(main())