332 lines
13 KiB
Python
332 lines
13 KiB
Python
# -*- 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_<Typ> ein TRO, Attribute ID/TYPE/ITEMS/
|
|
CONFIDENCE/SEPARATORS (+ FB_BLOCK mit --fb)
|
|
LWPOLYLINE "<von>-><nach>" ein Flusspfeil = eine Verbindung
|
|
Kreis/Linien "OFFEN:<Name>" 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 <stem>_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: "<von>-><nach>"
|
|
EDGE_RE = re.compile(r"^(?P<src>[^>]+)->(?P<dst>.+)$")
|
|
|
|
# "1x SSCC, 1x Scanner, 1x Separator" -> {"SSCC": 1, "Scanner": 1, ...}
|
|
ITEM_RE = re.compile(r"(?P<count>\d+)\s*x\s*(?P<item>.+)")
|
|
|
|
|
|
def xdata_payload(entity) -> str | None:
|
|
"""Zweiter String der XDATA von tro_annotate, sonst None.
|
|
|
|
Aufbau: (1000, "tro_annotate"), (1000, <Nutzlast>). Die Nutzlast ist je
|
|
nach Objekt die TRO-ID, "<von>-><nach>" oder "OFFEN:<Name>".
|
|
"""
|
|
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 <stamm>_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: <Zeichnung>_tro.json",
|
|
)
|
|
return parser.parse_args(argv)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|