s.ayadi
7242e815f8
- lib/dxf_scan_components.py: erkennt ILS-/Omniflo-Komponenten in fremden
Projektzeichnungen und schreibt sie im Testdaten-JSON-Schema aus
- lib/dxf_abbild.py: getreue Abbildung der Bauteile einer Fremdzeichnung
(Attribute, Weltkoordinaten, Unterkomponenten) ohne Interpretation
- tests/test_hundm05.{lsp,py}, testdata/hundm05.json, conftest-Fixtures und
alltests.json-Eintrag fuer den Kreisel-Abschnitt aus ST500592_05.dxf
- menu: TEST_HUNDM05 im Testmenue, Connection_Insert/Edit in SSG_LIB.cui
- Doku: hartkodierte Pfade durch (getenv "DXFMAKRO") ersetzt
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
1111 lines
47 KiB
Python
1111 lines
47 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
dxf_scan_components.py - Erkennt ILS-/Omniflo-Komponenten in einem FREMDEN DXF.
|
|
|
|
Gedacht fuer Projekt-Zeichnungen, die NICHT von unseren Makros erzeugt wurden
|
|
(also ohne die Wrapper-Bloecke KREISEL_*, VF_*, GF_*). Das Skript sucht die
|
|
Bauteile ueber Blocknamen-Muster, rechnet ihre Position in Weltkoordinaten um
|
|
und schreibt das Ergebnis im JSON-Schema aus tests/testdata/object_data.md -
|
|
also so, dass die Anlage per JSON-Treiber mit den Produktionsmakros
|
|
nachgebaut werden kann (Vorbild: tests/testdata/mubea.json, das bisher von
|
|
Hand aus einem DXF abgeleitet wurde).
|
|
|
|
Aufruf:
|
|
python dxf_scan_components.py <dxf> <data_dir> <output.json>
|
|
[--fenster X0 Y0 X1 Y1]
|
|
|
|
Beispiel (ganze Zeichnung):
|
|
python lib/dxf_scan_components.py results/ST500592_05.dxf data \
|
|
tests/testdata/hundm05.json
|
|
|
|
Beispiel (nur ein Anlagenabschnitt - hier der Kreisel-Bereich):
|
|
python lib/dxf_scan_components.py results/ST500592_05.dxf data \
|
|
tests/testdata/hundm05.json --fenster 216000 144000 228000 156000
|
|
|
|
Ohne --fenster liefert eine Zeichnung dieser Groesse ueber 1300 Eintraege -
|
|
zuviel fuer einen Testfall, der in BricsCAD noch diagnostizierbar bleibt.
|
|
|
|
Besonderheit: die Zieldateien sind binaere DXF im GB-Bereich. ezdxf.readfile()
|
|
wuerde die komplette Zeichnung als DOM in den Speicher legen (bei 1 GB Datei
|
|
weit mehr RAM als verfuegbar), darum liest dieses Skript die Tags per mmap
|
|
streamend selbst - konstanter Speicherbedarf, ein Durchlauf ca. 1 Minute.
|
|
Eine geoeffnete Zeichnung (BricsCAD-Dateisperre) stoert den Lesezugriff nicht.
|
|
"""
|
|
|
|
import argparse
|
|
import io
|
|
import json
|
|
import math
|
|
import mmap
|
|
import os
|
|
import re
|
|
import struct
|
|
import sys
|
|
|
|
from ezdxf.lldxf.types import BINARY_DATA, INT16, INT32, INT64, DOUBLE, BYTES
|
|
from ezdxf.math import Matrix44, OCS
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
from export_blockpatterns import matches_any # noqa: E402
|
|
|
|
PREFIX = "[dxf_scan]"
|
|
|
|
# Maximaler Abstand, bis zu dem Antriebs- und Umlenkstation noch als ein und
|
|
# derselbe Foerderer gewertet werden (mm).
|
|
MAX_PAAR_ABSTAND = 60000.0
|
|
|
|
# Bis zu dieser Hoehendifferenz (mm) gilt ein Antrieb/Umlenk-Paar als
|
|
# waagrechte Foerderstrecke ("gerade") und NICHT als VarioFoerderer. Der
|
|
# Vario-Solver braucht eine Hoehendifferenz, um einen Bogenwinkel aus der
|
|
# Kandidatenliste (3..51 Grad) zu waehlen; bei ~0 mm ist der Eintrag unbaubar.
|
|
# 10 mm deckt Rundungs- und Zeichenungenauigkeiten der Fremdzeichnung ab.
|
|
GERADE_MAX_DELTA_H = 10.0
|
|
|
|
# Toleranz fuer die Dedup-Rundung (Nachkommastellen).
|
|
RUND = 2
|
|
|
|
# Fenster, in dem die Einfuegehoehe eines Anlagenbauteils liegen muss (mm).
|
|
# Z-Werte ausserhalb sind praktisch immer ein Zeichen fuer eine falsch
|
|
# aufgeloeste Transformation (z.B. eine nicht beruecksichtigte OCS) und nicht
|
|
# fuer echte Geometrie - eine Foerderanlage hat keine Bauteile 15 m unter dem
|
|
# Hallenboden. Wird nur fuer die Plausibilitaetsmeldung genutzt, nicht zum
|
|
# Filtern: aussortiert wird nichts, aber es wird auch nichts still geschrieben.
|
|
Z_PLAUSIBEL = (-1000.0, 20000.0)
|
|
|
|
# Groesste plausible XY-Ausdehnung einer Anlage (mm).
|
|
XY_MAX_AUSDEHNUNG = 500000.0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Hilfsfunktionen
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def load_json(path):
|
|
"""JSON laden; UTF-8 mit cp1252-Fallback (wie lib/export_csv.py)."""
|
|
try:
|
|
with open(path, "r", encoding="utf-8") as fh:
|
|
return json.load(fh)
|
|
except UnicodeDecodeError:
|
|
with open(path, "r", encoding="cp1252") as fh:
|
|
return json.load(fh)
|
|
|
|
|
|
def decode_str(raw):
|
|
"""DXF-String dekodieren - die Zeichnungen mischen UTF-8 und cp1252."""
|
|
try:
|
|
return raw.decode("utf-8")
|
|
except UnicodeDecodeError:
|
|
return raw.decode("cp1252", errors="replace")
|
|
|
|
|
|
def iter_binary_tags(mm):
|
|
"""Streamt (code, value)-Tags aus einem binaeren DXF (AC1021+).
|
|
|
|
Nachbau von ezdxf.lldxf.tagger.binary_tags_loader, aber ueber ein
|
|
mmap-Objekt statt eines bytes-Objekts: der Original-Loader braucht
|
|
data.index(), das mmap nicht kennt, waehrend ein Proxy-Objekt wiederum am
|
|
Buffer-Protokoll von struct.unpack_from scheitert. Die Typtabellen
|
|
(BINARY_DATA/INT16/...) kommen weiterhin aus ezdxf.
|
|
"""
|
|
if mm[:22] != b"AutoCAD Binary DXF\r\n\x1a\x00":
|
|
raise ValueError("keine binaere DXF-Datei")
|
|
i = 22
|
|
n = len(mm)
|
|
unpack = struct.unpack_from
|
|
while i < n:
|
|
code = mm[i] | (mm[i + 1] << 8) # AC1021+: immer 2-Byte-Gruppencode
|
|
i += 2
|
|
if code in BINARY_DATA:
|
|
length = mm[i]
|
|
i += 1
|
|
value = mm[i:i + length]
|
|
i += length
|
|
elif code in INT16:
|
|
value = unpack("<h", mm, i)[0]
|
|
i += 2
|
|
elif code in DOUBLE:
|
|
value = unpack("<d", mm, i)[0]
|
|
i += 8
|
|
elif code in INT32:
|
|
value = unpack("<i", mm, i)[0]
|
|
i += 4
|
|
elif code in INT64:
|
|
value = unpack("<q", mm, i)[0]
|
|
i += 8
|
|
elif code in BYTES:
|
|
value = mm[i]
|
|
i += 1
|
|
else: # nullterminierter String
|
|
end = mm.find(b"\x00", i)
|
|
value = decode_str(mm[i:end])
|
|
i = end + 1
|
|
yield code, value
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Familien-Erkennung (Blockname -> Bauteilfamilie)
|
|
# ---------------------------------------------------------------------------
|
|
# Reihenfolge zaehlt: der erste Treffer gewinnt. Muster ohne Umlaute, dafuer
|
|
# mit Wildcards - die Blocknamen der Fremdzeichnungen enthalten Umlaute und
|
|
# Sonderzeichen (z.B. "3D_Schraegfoerderer_Antrieb", "TEF_Bogen_90Grad_...").
|
|
FAMILIEN = [
|
|
("kreisel_an", ["AN8", "AN8_*"]),
|
|
("kreisel_sp", ["SP8", "SP8_*"]),
|
|
("foerderer_antrieb", ["3D_TEF_Antrieb*", "3D_FF_Antrieb*",
|
|
"3D_Schr*rderer_Antrieb*", "Antrieb FF"]),
|
|
("foerderer_umlenk", ["3D_TEF_Umlenk*", "3D_FF_Umlenk*",
|
|
"3D_Schr*rderer_Umlenk*", "Umlenkung FF"]),
|
|
("omniflo_bogen", ["APB *", "APB_*", "APBW *", "APBW_*"]),
|
|
# ILS-Modulweiche VOR omniflo_weiche pruefen: "Weiche 2 Wege" ist eine
|
|
# ILS-Weiche (unsere Bloecke Modulweiche_RG_*/Modulweiche_RR_*) und wuerde
|
|
# sonst vom allgemeinen Muster "Weiche *" als Omniflo-Weiche eingesammelt.
|
|
# Als Omniflo-Eintrag ist sie zwangsweise ohne Sivasnr, weil der Blockname
|
|
# keine Radius-/Winkel-/Masskennzahlen enthaelt, aus denen sich ein
|
|
# Katalogtreffer ableiten liesse.
|
|
("ils_weiche", ["Weiche 2 Wege", "Weiche 2 Wege*"]),
|
|
("omniflo_weiche", ["Weiche *", "Weichen-TEF*"]),
|
|
("bogen", ["3D_ILS_Bogen_*", "3D_FF_Bogen_*", "TEF_Bogen_*",
|
|
"tef30*", "tef90*"]),
|
|
("schleuselement", ["Ausschleuselement*", "Einschleuselement*",
|
|
"EE_Neu*", "AE_Neu*"]),
|
|
("sensor", ["S-LP", "S-SP", "SCAN", "SCAN_*", "Scanner*",
|
|
"Separator*", "Stopper*", "Stopperschleuse*"]),
|
|
]
|
|
|
|
# Foerderer-System je Blockname - es wird nur innerhalb desselben Systems
|
|
# gepaart (ein TEF-Antrieb gehoert nie zu einer Fingerfoerderer-Umlenkung).
|
|
SYSTEM_MUSTER = [
|
|
("TEF", ["3D_TEF_*"]),
|
|
("FF", ["3D_FF_*", "Antrieb FF", "Umlenkung FF"]),
|
|
("SCHRAEG", ["3D_Schr*"]),
|
|
]
|
|
|
|
# Blocknamen der Fremdzeichnung -> Blocknamen, die ils-insert-sensor kennt
|
|
# (ssg-ils-block-datei loest diese mit 2D/3D-Suffix auf). Ohne Eintrag bleibt
|
|
# der Originalname stehen - der Nachbau muss dafuer dann einen Block liefern.
|
|
SENSOR_BLOCK_MAP = {
|
|
"SCAN": "Scanner",
|
|
"Separator": "Separator_SP",
|
|
"Separator 200000070": "Separator_SP",
|
|
"Separator LP 3D": "S-LP",
|
|
}
|
|
|
|
# Varianten-Defaults fuer Angaben, die im Fremd-Blocknamen fehlen. Unsere
|
|
# Bibliothek unterscheidet Seite (links/rechts), Winkel und bei Kurven die
|
|
# Lage (aussen/innen); die Fremdzeichnung benennt das nur teilweise
|
|
# ("3D_FF_Antrieb_rechts" ja, "Antrieb FF" nein). Ohne eine Festlegung liesse
|
|
# sich kein Block aufloesen, darum diese Defaults - jede so gesetzte Angabe
|
|
# wird im comment-Feld des Eintrags als geraten ausgewiesen, damit sie im
|
|
# JSON gezielt nachgezogen werden kann.
|
|
DEFAULT_SEITE = "rechts"
|
|
DEFAULT_BOGEN_LAGE = "aussen"
|
|
DEFAULT_KURVEN_WINKEL = 30
|
|
DEFAULT_SCHLEUS_WINKEL = 90
|
|
|
|
|
|
def winkel_aus_name(bname, default):
|
|
"""Ersten Winkel (Grad) aus einem Blocknamen lesen.
|
|
|
|
Rueckgabe: (winkel, geraten) - geraten=True, wenn der Name keinen Winkel
|
|
enthaelt und der Default eingesetzt wurde.
|
|
"""
|
|
m = re.search(r"(\d{1,3})\s*(?:Grad|°)", bname)
|
|
if not m:
|
|
m = re.search(r"_(\d{1,3})(?:_|$)", bname)
|
|
if m:
|
|
return int(m.group(1)), False
|
|
return default, True
|
|
|
|
|
|
def mappe_block(bname):
|
|
"""Fremd-Blockname auf einen Block aus data/ils/ abbilden.
|
|
|
|
Rueckgabe: (zielblock, geraten). "geraten" listet die Angaben, die im
|
|
Fremdnamen NICHT stehen und darum aus den DEFAULT_*-Konstanten kommen;
|
|
eine leere Liste heisst: vollstaendig aus dem Namen abgeleitet. Ist der
|
|
Name gar nicht zuzuordnen, kommt er unveraendert zurueck (geraten=None) -
|
|
dann entscheidet der LISP-Treiber (Fallback auf Separator_SP bzw.
|
|
Platzhalter).
|
|
"""
|
|
if bname in SENSOR_BLOCK_MAP:
|
|
return SENSOR_BLOCK_MAP[bname], []
|
|
low = bname.lower()
|
|
geraten = []
|
|
|
|
def seite_aus():
|
|
if "rechts" in low:
|
|
return "rechts"
|
|
if "links" in low:
|
|
return "links"
|
|
geraten.append("Seite=%s" % DEFAULT_SEITE)
|
|
return DEFAULT_SEITE
|
|
|
|
# Antriebs-/Umlenkstation -> Vario_Motorstation / Vario_Umlenkstation
|
|
if "antrieb" in low:
|
|
return "Vario_Motorstation_500mm_%s" % seite_aus(), geraten
|
|
if "umlenk" in low:
|
|
return "Vario_Umlenkstation_500mm_%s" % seite_aus(), geraten
|
|
|
|
# Kurven/Boegen -> Vario_Kurve_<seite>_<winkel>_TEF_<aussen|innen>
|
|
if "bogen" in low or low.startswith("tef"):
|
|
winkel, geschaetzt = winkel_aus_name(bname, DEFAULT_KURVEN_WINKEL)
|
|
if geschaetzt:
|
|
geraten.append("Winkel=%d" % winkel)
|
|
if "außen" in low or "aussen" in low:
|
|
lage = "aussen"
|
|
elif "innen" in low:
|
|
lage = "innen"
|
|
else:
|
|
lage = DEFAULT_BOGEN_LAGE
|
|
geraten.append("Lage=%s" % lage)
|
|
return ("Vario_Kurve_%s_%d_TEF_%s" % (seite_aus(), winkel, lage),
|
|
geraten)
|
|
|
|
# Aus-/Einschleuselement -> AS_Element / ES_Element
|
|
if "ausschleus" in low or low.startswith("ae_"):
|
|
winkel, geschaetzt = winkel_aus_name(bname, DEFAULT_SCHLEUS_WINKEL)
|
|
if geschaetzt:
|
|
geraten.append("Winkel=%d" % winkel)
|
|
return "AS_Element_%d_%s" % (winkel, seite_aus()), geraten
|
|
if "einschleus" in low or low.startswith("ee_"):
|
|
winkel, geschaetzt = winkel_aus_name(bname, DEFAULT_SCHLEUS_WINKEL)
|
|
if geschaetzt:
|
|
geraten.append("Winkel=%d" % winkel)
|
|
return "ES_Element_%d_%s" % (winkel, seite_aus()), geraten
|
|
|
|
return bname, None
|
|
|
|
|
|
def lade_bekannte_bloecke(data_dir):
|
|
"""Blocknamen aus data/ils/ (ohne _2D/_3D-Suffix) als Menge.
|
|
|
|
Dient der Kontrolle, ob ein von mappe_block gebildeter Name wirklich
|
|
existiert - eine erfundene Variante wuerde im Testlauf sonst still zum
|
|
Platzhalter.
|
|
"""
|
|
ils_dir = os.path.join(data_dir, "ils")
|
|
namen = set()
|
|
if not os.path.isdir(ils_dir):
|
|
return namen
|
|
for datei in os.listdir(ils_dir):
|
|
stamm, endung = os.path.splitext(datei)
|
|
if endung.lower() != ".dwg":
|
|
continue
|
|
for suffix in ("_2D", "_3D"):
|
|
if stamm.endswith(suffix):
|
|
stamm = stamm[:-len(suffix)]
|
|
break
|
|
namen.add(stamm)
|
|
return namen
|
|
|
|
# Blocknamen, die zwar technisch aussehen, aber bewusst ignoriert werden
|
|
# (Ausstattung/Visualisierung, kein Anlagenbauteil).
|
|
IGNORIEREN = ["Trolley*", "MT Trolley*", "Minitrolley*", "*Trolley*"]
|
|
|
|
|
|
def familie_von(bname):
|
|
"""Bauteilfamilie zu einem Blocknamen, oder None."""
|
|
if matches_any(bname, IGNORIEREN):
|
|
return None
|
|
for name, muster in FAMILIEN:
|
|
if matches_any(bname, muster):
|
|
return name
|
|
return None
|
|
|
|
|
|
def system_von(bname):
|
|
"""Foerderer-System (TEF/FF/SCHRAEG) zu einem Blocknamen."""
|
|
for name, muster in SYSTEM_MUSTER:
|
|
if matches_any(bname, muster):
|
|
return name
|
|
return "?"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# DXF einlesen und Bloecke in Weltkoordinaten aufloesen
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def lese_inserts(dxf_path):
|
|
"""Liest alle INSERTs (Modelspace und Blockdefinitionen) plus Basispunkte.
|
|
|
|
Rueckgabe: (ms_inserts, block_inserts, block_base, statistik)
|
|
ms_inserts - Liste der INSERTs der ENTITIES-Sektion
|
|
block_inserts- dict blockname -> Liste der darin verschachtelten INSERTs
|
|
block_base - dict blockname -> Basispunkt [x, y, z]
|
|
statistik - dict mit Zaehlern fuer die Diagnoseausgabe
|
|
"""
|
|
ms_inserts = []
|
|
block_inserts = {}
|
|
block_base = {}
|
|
statistik = {"inserts": 0, "ocs_gekippt": 0, "papierbereich": 0}
|
|
|
|
section = None
|
|
warte_auf_sectionname = False
|
|
aktuell = None # aktueller Entity-Typ
|
|
aktueller_block = None # Name der gerade gelesenen Blockdefinition
|
|
rec = None # INSERT-Datensatz im Aufbau
|
|
attrib_tag = None # Code 2 des laufenden ATTRIB
|
|
attrib_wert = None # Code 1 des laufenden ATTRIB (kommt VOR dem Tag)
|
|
|
|
def neu_insert(im_modelspace, parent):
|
|
return {"name": None, "layer": None, "p": [0.0, 0.0, 0.0],
|
|
"rot": 0.0, "s": [1.0, 1.0, 1.0], "ps": 0, "attribs": {},
|
|
"ext": [0.0, 0.0, 1.0], "_ms": im_modelspace,
|
|
"_parent": parent}
|
|
|
|
def ablegen(datensatz):
|
|
if datensatz["name"] is None:
|
|
return
|
|
# Erst hier ist die Extrusionsrichtung komplett gelesen (210/220/230
|
|
# kommen als drei einzelne Tags), darum wird die OCS-Statistik beim
|
|
# Ablegen gefuehrt und nicht beim Lesen der einzelnen Codes.
|
|
if datensatz["ext"] != [0.0, 0.0, 1.0]:
|
|
statistik["ocs_gekippt"] += 1
|
|
if datensatz["_ms"]:
|
|
if datensatz["ps"] != 1: # Papierbereich ueberspringen
|
|
ms_inserts.append(datensatz)
|
|
elif datensatz["_parent"]:
|
|
block_inserts.setdefault(datensatz["_parent"], []).append(datensatz)
|
|
|
|
fh = io.open(dxf_path, "rb")
|
|
try:
|
|
mm = mmap.mmap(fh.fileno(), 0, access=mmap.ACCESS_READ)
|
|
except ValueError:
|
|
fh.close()
|
|
raise
|
|
try:
|
|
for code, value in iter_binary_tags(mm):
|
|
if code == 0:
|
|
# Ein ATTRIB endet immer am naechsten 0-Tag. Erst hier steht
|
|
# sicher fest, dass Tag UND Wert gelesen sind - in einem ATTRIB
|
|
# kommt laut DXF-Spezifikation Code 1 (Wert) VOR Code 2 (Tag),
|
|
# eine Zuordnung schon beim Lesen von Code 1 waere also immer
|
|
# leer (in dieser Zeichnung: 400 von 400 Stichproben in der
|
|
# Reihenfolge 1,2 - saemtliche Attribute gingen verloren).
|
|
if rec is not None and attrib_tag and attrib_wert:
|
|
rec["attribs"][attrib_tag] = attrib_wert
|
|
attrib_tag = None
|
|
attrib_wert = None
|
|
if value in ("ATTRIB", "SEQEND"):
|
|
aktuell = value
|
|
continue
|
|
if rec is not None:
|
|
ablegen(rec)
|
|
rec = None
|
|
if value == "SECTION":
|
|
warte_auf_sectionname = True
|
|
aktuell = None
|
|
elif value == "ENDSEC":
|
|
section = None
|
|
aktuell = None
|
|
aktueller_block = None
|
|
elif value == "ENDBLK":
|
|
aktueller_block = None
|
|
aktuell = value
|
|
else:
|
|
aktuell = value
|
|
if value == "BLOCK":
|
|
aktueller_block = None
|
|
elif value == "INSERT":
|
|
rec = neu_insert(section == "ENTITIES", aktueller_block)
|
|
statistik["inserts"] += 1
|
|
elif code == 2:
|
|
if warte_auf_sectionname:
|
|
section = value
|
|
warte_auf_sectionname = False
|
|
elif aktuell == "BLOCK":
|
|
aktueller_block = value
|
|
block_base.setdefault(value, [0.0, 0.0, 0.0])
|
|
elif aktuell == "INSERT" and rec is not None:
|
|
rec["name"] = value
|
|
elif aktuell == "ATTRIB":
|
|
attrib_tag = value
|
|
elif aktuell == "BLOCK" and aktueller_block and code in (10, 20, 30):
|
|
block_base[aktueller_block][{10: 0, 20: 1, 30: 2}[code]] = value
|
|
elif aktuell == "INSERT" and rec is not None:
|
|
if code == 8:
|
|
rec["layer"] = value
|
|
elif code in (10, 20, 30):
|
|
rec["p"][{10: 0, 20: 1, 30: 2}[code]] = value
|
|
elif code in (41, 42, 43):
|
|
rec["s"][{41: 0, 42: 1, 43: 2}[code]] = value
|
|
elif code == 50:
|
|
rec["rot"] = value
|
|
elif code == 67:
|
|
rec["ps"] = value
|
|
if value == 1:
|
|
statistik["papierbereich"] += 1
|
|
elif code in (210, 220, 230):
|
|
# Extrusionsrichtung = Normale des OCS, in dem p/rot/s
|
|
# angegeben sind (siehe insert_matrix).
|
|
rec["ext"][{210: 0, 220: 1, 230: 2}[code]] = value
|
|
elif aktuell == "ATTRIB" and code == 1 and attrib_wert is None:
|
|
# Nur den ersten Code 1 nehmen; die Zuordnung zum Tag erfolgt
|
|
# am Ende des ATTRIB (siehe oben beim 0-Tag).
|
|
attrib_wert = value
|
|
if rec is not None:
|
|
ablegen(rec)
|
|
finally:
|
|
mm.close()
|
|
fh.close()
|
|
return ms_inserts, block_inserts, block_base, statistik
|
|
|
|
|
|
def insert_matrix(rec, block_base):
|
|
"""Transformationsmatrix eines INSERT (inkl. Basispunkt des Zielblocks).
|
|
|
|
Einfuegepunkt, Rotation und Skalierung eines INSERT sind NICHT in
|
|
Weltkoordinaten angegeben, sondern im Objektkoordinatensystem (OCS), das
|
|
von der Extrusionsrichtung (Gruppencode 210/220/230) aufgespannt wird.
|
|
Bei der Normalen (0,0,1) ist OCS == WCS, und der Unterschied fiele nicht
|
|
auf; in dieser Zeichnung liegen aber rund 14 Prozent der INSERTs auf
|
|
gekippten Ebenen - haeufig (0,0,-1) sowie (+-1,0,0), also senkrecht
|
|
stehende Einbauten. Ohne die abschliessende OCS->WCS-Abbildung wird deren
|
|
Y als Z gedeutet: Positionen und Rotation sind dann falsch, ohne dass man
|
|
es dem Ergebnis ansieht. Die OCS-Achsen liefert ezdxf ueber den
|
|
Arbitrary-Axis-Algorithmus.
|
|
"""
|
|
base = block_base.get(rec["name"], (0.0, 0.0, 0.0))
|
|
lokal = Matrix44.scale(*rec["s"])
|
|
lokal @= Matrix44.z_rotate(math.radians(rec["rot"]))
|
|
lokal @= Matrix44.translate(*rec["p"])
|
|
ext = rec.get("ext", [0.0, 0.0, 1.0])
|
|
# Nullvektor waere eine defekte Extrusionsrichtung - OCS() wuerde daran
|
|
# scheitern; in dem Fall bleibt es bei OCS == WCS.
|
|
if list(ext) != [0.0, 0.0, 1.0] and any(ext):
|
|
ocs = OCS(ext)
|
|
lokal @= Matrix44.ucs(ocs.ux, ocs.uy, ocs.uz)
|
|
return Matrix44.translate(-base[0], -base[1], -base[2]) @ lokal
|
|
|
|
|
|
def im_fenster(x, y, fenster):
|
|
"""Liegt (x, y) im XY-Fenster (x0, y0, x1, y1)? Ohne Fenster: immer ja."""
|
|
if not fenster:
|
|
return True
|
|
x0, y0, x1, y1 = fenster
|
|
return x0 <= x <= x1 and y0 <= y <= y1
|
|
|
|
|
|
def sammle_vorkommen(ms_inserts, block_inserts, block_base, max_tiefe=12):
|
|
"""Loest alle INSERTs rekursiv in Weltkoordinaten auf.
|
|
|
|
Aequivalent zu ezdxf virtual_entities(), aber ohne DOM: die Matrizen
|
|
werden selbst komponiert. Gespiegelte Inserts (negative Skalierung)
|
|
liefern dadurch ebenfalls die korrekte Weltrotation. Zurueckgegeben
|
|
werden nur Vorkommen, deren Blockname einer Familie zugeordnet ist.
|
|
|
|
Hier wird NICHT auf einen Abschnitt gefiltert - das passiert erst nach
|
|
der Paarbildung (siehe filtere_fenster). Ein Filter an dieser Stelle
|
|
wuerde die Paare zerreissen: der Kreisel dieser Zeichnung ist 23,6 m
|
|
lang, seine Spannstation liegt ausserhalb eines 12-m-Fensters, und ein
|
|
AN8 ohne SP8-Partner ergibt keinen Kreisel mehr, sondern einen
|
|
bedeutungslosen Einzel-Insert. Dasselbe gilt fuer Antrieb/Umlenkung.
|
|
"""
|
|
treffer = []
|
|
unbekannt = {}
|
|
|
|
def lauf(recs, mutter, tiefe, pfad):
|
|
if tiefe > max_tiefe:
|
|
return
|
|
for rec in recs:
|
|
mat = insert_matrix(rec, block_base) @ mutter
|
|
ursprung = mat.transform((0.0, 0.0, 0.0))
|
|
xachse = mat.transform((1.0, 0.0, 0.0)) - ursprung
|
|
rot = math.degrees(math.atan2(xachse.y, xachse.x)) % 360.0
|
|
fam = familie_von(rec["name"])
|
|
if fam:
|
|
treffer.append({
|
|
"familie": fam,
|
|
"block": rec["name"],
|
|
"layer": rec["layer"] or "",
|
|
# "+ 0.0" normalisiert negative Null: round(-0.0, 2)
|
|
# bleibt -0.0 und landete als "-0.0" im JSON, was beim
|
|
# Lesen und Vergleichen von Hand nur verwirrt.
|
|
"x": round(ursprung.x, RUND) + 0.0,
|
|
"y": round(ursprung.y, RUND) + 0.0,
|
|
"z": round(ursprung.z, RUND) + 0.0,
|
|
"rot": round(rot, RUND) + 0.0,
|
|
"tiefe": tiefe,
|
|
"attribs": rec["attribs"],
|
|
})
|
|
elif rec["name"] and not rec["name"].startswith("*"):
|
|
unbekannt[rec["name"]] = unbekannt.get(rec["name"], 0) + 1
|
|
kinder = block_inserts.get(rec["name"])
|
|
if kinder and rec["name"] not in pfad:
|
|
lauf(kinder, mat, tiefe + 1, pfad | {rec["name"]})
|
|
|
|
lauf(ms_inserts, Matrix44(), 0, frozenset())
|
|
|
|
# Echte Duplikate entfernen - die Zeichnung enthaelt deckungsgleiche
|
|
# Mehrfach-Inserts (z.B. SP8 zweimal auf derselben Position).
|
|
gesehen = set()
|
|
eindeutig = []
|
|
verworfen = 0
|
|
for t in treffer:
|
|
key = (t["block"], t["x"], t["y"], t["z"], t["rot"])
|
|
if key in gesehen:
|
|
verworfen += 1
|
|
continue
|
|
gesehen.add(key)
|
|
eindeutig.append(t)
|
|
return eindeutig, unbekannt, verworfen
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Omniflo: Sivasnr aus dem Blocknamen ableiten
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def zahlen_aus(text):
|
|
"""Alle Zahlen eines Textes als float-Liste (Komma = Dezimaltrenner)."""
|
|
return [float(t.replace(",", ".")) for t in re.findall(r"\d+(?:[.,]\d+)?", text)]
|
|
|
|
|
|
def basis_sivasnr(eintraege):
|
|
"""Waehlt aus Katalogtreffern den Grundeintrag (ohne TEF-Kombination)."""
|
|
for e in eintraege:
|
|
if isinstance(e["Sivasnr"], int):
|
|
return str(e["Sivasnr"])
|
|
return str(eintraege[0]["Sivasnr"]) if eintraege else None
|
|
|
|
|
|
def sivasnr_bogen(bname, katalog):
|
|
"""Sivasnr eines Omniflo-Bogens aus dem Blocknamen ermitteln."""
|
|
direkt = re.search(r"\b(\d{9})\b", bname)
|
|
if direkt:
|
|
return direkt.group(1), "Sivasnr direkt aus Blockname"
|
|
nums = zahlen_aus(bname)
|
|
if len(nums) < 3:
|
|
return None, "Blockname enthaelt zu wenige Kennzahlen"
|
|
radius, winkel = nums[1], nums[2]
|
|
kandidaten = [e for e in katalog
|
|
if e["Radius"] == radius and e["KurvenWinkel"] == winkel]
|
|
if not kandidaten and len(nums) > 3 and float(winkel).is_integer():
|
|
# "67 5" im Blocknamen meint 67.5 - der Punkt geht beim Blocknamen verloren
|
|
winkel2 = winkel + nums[3] / 10.0
|
|
kandidaten = [e for e in katalog
|
|
if e["Radius"] == radius and e["KurvenWinkel"] == winkel2]
|
|
if kandidaten:
|
|
winkel = winkel2
|
|
nums = nums[:3] + nums[4:]
|
|
if len(kandidaten) > 1 and len(nums) >= 5:
|
|
breite, laenge = nums[3], nums[4]
|
|
enger = [e for e in kandidaten
|
|
if e["Breite"] == breite and e["Länge"] == laenge]
|
|
if enger:
|
|
kandidaten = enger
|
|
if not kandidaten:
|
|
return None, "kein Katalogtreffer fuer R=%g/Winkel=%g" % (radius, winkel)
|
|
return (basis_sivasnr(kandidaten),
|
|
"Katalogtreffer R=%g/Winkel=%g" % (radius, winkel))
|
|
|
|
|
|
def sivasnr_weiche(bname, katalog):
|
|
"""Sivasnr einer Omniflo-Weiche aus dem Blocknamen ermitteln."""
|
|
direkt = re.search(r"\b(\d{9})\b", bname)
|
|
if direkt:
|
|
return direkt.group(1), "Sivasnr direkt aus Blockname"
|
|
nums = zahlen_aus(bname)
|
|
if len(nums) < 3:
|
|
return None, "Blockname enthaelt zu wenige Kennzahlen"
|
|
winkel, breite, laenge = nums[0], nums[1], nums[2]
|
|
richtung = None
|
|
if re.search(r"\bR\b", bname):
|
|
richtung = 2
|
|
elif re.search(r"\bL\b", bname):
|
|
richtung = 1
|
|
schaltung = None
|
|
m = re.search(r"mit\s+([MP])\b", bname, re.I)
|
|
if m:
|
|
schaltung = m.group(1).upper()
|
|
kandidaten = [e for e in katalog if e["KurvenWinkel"] == winkel
|
|
and e["Breite"] == breite and e["Länge"] == laenge]
|
|
if richtung is not None:
|
|
kandidaten = [e for e in kandidaten if e["KurvenRichtung"] == richtung]
|
|
if schaltung:
|
|
kandidaten = [e for e in kandidaten if e["Schaltungstyp"] == schaltung]
|
|
if not kandidaten:
|
|
return None, "kein Katalogtreffer fuer Winkel=%g/%gx%g" % (winkel, breite, laenge)
|
|
return (basis_sivasnr(kandidaten),
|
|
"Katalogtreffer Winkel=%g/%gx%g" % (winkel, breite, laenge))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Paarbildung und Schema-Abbildung
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def paare_bilden(antriebe, umlenkungen):
|
|
"""Gegenseitig naechste Nachbarn paaren (nur innerhalb desselben Systems).
|
|
|
|
Rueckgabe: (paare, rest_antriebe, rest_umlenkungen)
|
|
"""
|
|
def dist(a, b):
|
|
return math.hypot(b["x"] - a["x"], b["y"] - a["y"])
|
|
|
|
paare = []
|
|
offen_a = list(antriebe)
|
|
offen_u = list(umlenkungen)
|
|
while True:
|
|
bester = None
|
|
for a in offen_a:
|
|
kand = [u for u in offen_u if system_von(u["block"]) == system_von(a["block"])]
|
|
if not kand:
|
|
continue
|
|
u = min(kand, key=lambda k: dist(a, k))
|
|
d = dist(a, u)
|
|
if d > MAX_PAAR_ABSTAND:
|
|
continue
|
|
# Gegenprobe: ist a auch fuer u der naechste Antrieb?
|
|
kand_a = [x for x in offen_a if system_von(x["block"]) == system_von(u["block"])]
|
|
if min(kand_a, key=lambda k: dist(k, u)) is not a:
|
|
continue
|
|
if bester is None or d < bester[2]:
|
|
bester = (a, u, d)
|
|
if bester is None:
|
|
break
|
|
a, u, d = bester
|
|
offen_a.remove(a)
|
|
offen_u.remove(u)
|
|
paare.append((a, u, d))
|
|
return paare, offen_a, offen_u
|
|
|
|
|
|
def foerderer_eintrag(a, u, dist_xy, nummer, gf_winkel_min):
|
|
"""Ein Antrieb/Umlenk-Paar auf Gerade, VarioFoerderer (3.1) oder
|
|
Gefaellestrecke (2.1) abbilden.
|
|
|
|
Die Aufteilung folgt der Produktionsregel aus gf-analysiere-kette
|
|
(Lisp/Gefaellestrecke.lsp): Endpunkt hoeher -> VarioFoerderer; tiefer mit
|
|
mindestens dem Gefaellewinkel -> Gefaellestrecke; tiefer mit weniger
|
|
Neigung -> VarioFoerderer "Ab".
|
|
|
|
Vorgeschaltet der Sonderfall ohne Hoehendifferenz: ein VarioFoerderer ist
|
|
per Definition ein Steig- oder Fallfoerderer, sein Solver
|
|
(berechne-standard -> berechne-alle-winkel) muss einen Bogenwinkel aus der
|
|
Kandidatenliste 3..51 Grad finden. Bei deltaH ~ 0 gibt es keinen, der
|
|
Eintrag waere unbaubar. Solche Paare sind waagrechte Foerderstrecken und
|
|
werden als eigener Typ "gerade" ausgewiesen.
|
|
"""
|
|
dz = u["z"] - a["z"]
|
|
hz = round(math.degrees(math.atan2(u["y"] - a["y"], u["x"] - a["x"])) % 360.0, RUND)
|
|
neigung = round(math.degrees(math.atan2(abs(dz), dist_xy)), RUND) if dist_xy else 0.0
|
|
herkunft = ("aus DXF: %s -> %s (Layer %s / %s), XY-Abstand %.2f, "
|
|
"Neigung %.2f Grad" % (a["block"], u["block"], a["layer"],
|
|
u["layer"], dist_xy, neigung))
|
|
|
|
if abs(dz) < GERADE_MAX_DELTA_H:
|
|
return {
|
|
"test_id": "GERADE_HundM05_%02d" % nummer,
|
|
"typ": "gerade",
|
|
"laenge": round(dist_xy, RUND),
|
|
"hz": hz,
|
|
"startpunkt": [a["x"], a["y"], a["z"]],
|
|
"endpunkt": [u["x"], u["y"], u["z"]],
|
|
"comment": herkunft + "; waagrechte Foerderstrecke (Hoehen"
|
|
"differenz %.2f mm < %g mm) - KEIN VarioFoerderer, da "
|
|
"dessen Winkel-Solver eine Hoehendifferenz braucht."
|
|
% (abs(dz), GERADE_MAX_DELTA_H),
|
|
}
|
|
|
|
if dz < 0 and neigung >= gf_winkel_min:
|
|
return {
|
|
"test_id": "GF_HundM05_%02d" % nummer,
|
|
"modus": 1,
|
|
"deltaL_mm": round(dist_xy, RUND),
|
|
"winkel_grad": neigung,
|
|
"hz_grad": hz,
|
|
"start_mm": [a["x"], a["y"], a["z"]],
|
|
"as_seite": "links",
|
|
"es_seite": "links",
|
|
"as_winkel": "90",
|
|
"es_winkel": "90",
|
|
"comment": herkunft + "; Gefaellestrecke, da fallend und Neigung "
|
|
">= %g Grad. as_/es_seite und -winkel sind Defaults "
|
|
"(im DXF nicht ableitbar)." % gf_winkel_min,
|
|
}
|
|
|
|
return {
|
|
"test_id": "VF_HundM05_%02d" % nummer,
|
|
"typ": "standard",
|
|
"richtung": "Auf" if dz > 0 else "Ab",
|
|
"deltaL": round(dist_xy, RUND),
|
|
"deltaH": round(abs(dz), RUND),
|
|
"startpunkt": [a["x"], a["y"], a["z"]],
|
|
"seite": "rechts",
|
|
"hz": hz,
|
|
"comment": herkunft + "; VarioFoerderer. 'winkel' bewusst nicht "
|
|
"gesetzt - der Solver waehlt den kleinsten gueltigen "
|
|
"Kandidaten. 'seite' ist Default (im DXF nicht ableitbar).",
|
|
}
|
|
|
|
|
|
def sensor_eintrag(t, bekannte_bloecke=None, hinweise=None):
|
|
"""Beliebiges Bauteil auf das generische Insert-Schema (5.) abbilden.
|
|
|
|
"bekannte_bloecke" (siehe lade_bekannte_bloecke) dient der Kontrolle, ob
|
|
der abgebildete Blockname in data/ils/ wirklich existiert; fehlende
|
|
Varianten werden in "hinweise" gemeldet statt im Testlauf still zu einem
|
|
Platzhalter zu werden.
|
|
"""
|
|
block, geraten = mappe_block(t["block"])
|
|
kommentar = "aus DXF: Block '%s', Layer '%s'" % (t["block"], t["layer"])
|
|
if block != t["block"]:
|
|
kommentar += ", auf Makro-Block '%s' abgebildet" % block
|
|
if geraten:
|
|
kommentar += ("; ACHTUNG geraten (im DXF nicht ableitbar): %s - bei "
|
|
"Bedarf hier im JSON korrigieren" % ", ".join(geraten))
|
|
if bekannte_bloecke and block not in bekannte_bloecke:
|
|
kommentar += ("; Block '%s' ist in data/ils/ NICHT vorhanden - "
|
|
"erzeugt im Testlauf einen Platzhalter" % block)
|
|
if hinweise is not None:
|
|
hinweise.append("Block '%s' (aus '%s') fehlt in data/ils/"
|
|
% (block, t["block"]))
|
|
if t["attribs"]:
|
|
kommentar += ", Attribute %s" % json.dumps(t["attribs"], ensure_ascii=False)
|
|
return {
|
|
"function": "insert",
|
|
"block": block,
|
|
"x": t["x"],
|
|
"y": t["y"],
|
|
"z": t["z"],
|
|
"rotation": t["rot"],
|
|
"comment": kommentar,
|
|
}
|
|
|
|
|
|
def baue_eintraege(treffer, defaults, boegen, weichen,
|
|
bekannte_bloecke=None):
|
|
"""Erkannte Vorkommen auf die Schemata aus object_data.md abbilden."""
|
|
nach_familie = {}
|
|
for t in treffer:
|
|
nach_familie.setdefault(t["familie"], []).append(t)
|
|
|
|
eintraege = []
|
|
hinweise = []
|
|
|
|
# --- Kreisel: AN8 (Antriebsstation) + SP8 (Spannstation) --------------
|
|
durchmesser = defaults.get("kreisel", {}).get("durchmesser", 800.0)
|
|
an_liste = sorted(nach_familie.get("kreisel_an", []), key=lambda t: (t["x"], t["y"]))
|
|
sp_liste = list(nach_familie.get("kreisel_sp", []))
|
|
for i, an in enumerate(an_liste, start=1):
|
|
if not sp_liste:
|
|
hinweise.append("AN8 bei (%.2f, %.2f) ohne SP8-Partner" % (an["x"], an["y"]))
|
|
eintraege.append(sensor_eintrag(an, bekannte_bloecke, hinweise))
|
|
continue
|
|
sp = min(sp_liste, key=lambda s: math.hypot(s["x"] - an["x"], s["y"] - an["y"]))
|
|
sp_liste.remove(sp)
|
|
dist = math.hypot(sp["x"] - an["x"], sp["y"] - an["y"])
|
|
rotation = math.degrees(math.atan2(sp["y"] - an["y"], sp["x"] - an["x"])) % 360.0
|
|
eintraege.append({
|
|
"id": "Kreisel_HundM05_%02d" % i,
|
|
"function": "connect",
|
|
"start_x": an["x"], "start_y": an["y"], "start_z": an["z"],
|
|
"end_x": sp["x"], "end_y": sp["y"], "end_z": sp["z"],
|
|
"typ": "STANDARD",
|
|
"hoehe": an["z"],
|
|
"comment": "aus DXF: AN8 (Layer %s) -> SP8 (Layer %s), Abstand "
|
|
"%.2f; daraus folgt produktionsseitig abstand = %.2f - "
|
|
"%.1f = %.2f und rotation = %.2f Grad. ACHTUNG hoehe: "
|
|
"AN8 und SP8 sind in dieser Zeichnung flache 2D-Symbole "
|
|
"(nur Linien/Kreis/Text auf Z=0) und auf Z=0 eingefuegt "
|
|
"- die Einfuegehoehe des Kreisels steht NICHT im DXF "
|
|
"und muss hier von Hand gesetzt werden (vgl. "
|
|
"mubea.json, wo z ebenfalls Handarbeit ist)."
|
|
% (an["layer"], sp["layer"], dist, dist, durchmesser,
|
|
dist - durchmesser, rotation),
|
|
})
|
|
for sp in sp_liste:
|
|
hinweise.append("SP8 bei (%.2f, %.2f) ohne AN8-Partner" % (sp["x"], sp["y"]))
|
|
eintraege.append(sensor_eintrag(sp, bekannte_bloecke, hinweise))
|
|
|
|
# --- Foerderer: Antriebs- und Umlenkstation paaren ---------------------
|
|
gf_winkel_min = defaults.get("gefaelle", {}).get("default_winkel", 3.0)
|
|
paare, rest_a, rest_u = paare_bilden(nach_familie.get("foerderer_antrieb", []),
|
|
nach_familie.get("foerderer_umlenk", []))
|
|
paare.sort(key=lambda p: (p[0]["x"], p[0]["y"]))
|
|
for i, (a, u, d) in enumerate(paare, start=1):
|
|
eintraege.append(foerderer_eintrag(a, u, d, i, gf_winkel_min))
|
|
for t in rest_a + rest_u:
|
|
hinweise.append("Foerderer-Station ohne Partner: %s bei (%.2f, %.2f)"
|
|
% (t["block"], t["x"], t["y"]))
|
|
eintraege.append(sensor_eintrag(t, bekannte_bloecke, hinweise))
|
|
|
|
# --- Omniflo Boegen und Weichen ---------------------------------------
|
|
for familie, katalog, typ in (("omniflo_bogen", boegen, "bogen"),
|
|
("omniflo_weiche", weichen, "weiche")):
|
|
posten = sorted(nach_familie.get(familie, []), key=lambda t: (t["x"], t["y"]))
|
|
for i, t in enumerate(posten, start=1):
|
|
if typ == "bogen":
|
|
sivasnr, grund = sivasnr_bogen(t["block"], katalog)
|
|
else:
|
|
sivasnr, grund = sivasnr_weiche(t["block"], katalog)
|
|
if sivasnr is None:
|
|
hinweise.append("Omniflo %s ohne Sivasnr: '%s' (%s)"
|
|
% (typ, t["block"], grund))
|
|
eintraege.append({
|
|
"id": sivasnr or ("%s_HundM05_%02d" % (typ.upper(), i)),
|
|
"type": typ,
|
|
"sivasnr": sivasnr,
|
|
"x": t["x"], "y": t["y"], "z": t["z"],
|
|
"hoehe": t["z"],
|
|
"drehung": t["rot"],
|
|
"comment": "aus DXF: Block '%s', Layer '%s'; %s"
|
|
% (t["block"], t["layer"], grund),
|
|
})
|
|
|
|
# --- ILS-Modulweichen -------------------------------------------------
|
|
# Welche der zwoelf Modulweiche-Varianten (RG links/rechts bzw. RR, je
|
|
# Passiv/Pneumatik/Zylinder) gemeint ist, steht im Blocknamen "Weiche 2
|
|
# Wege" nicht drin und ist aus der Zeichnung auch sonst nicht ableitbar.
|
|
# Darum bleibt der Originalname stehen und die Variante wird als Hinweis
|
|
# gemeldet, statt hier eine zu erfinden.
|
|
ils_weichen = sorted(nach_familie.get("ils_weiche", []),
|
|
key=lambda t: (t["x"], t["y"]))
|
|
if ils_weichen:
|
|
hinweise.append(
|
|
"%d ILS-Modulweiche(n): Variante nicht aus der Zeichnung "
|
|
"ableitbar - Blockname bleibt stehen, gewuenschte Variante "
|
|
"(Modulweiche_RG_links/rechts bzw. Modulweiche_RR, je Passiv/"
|
|
"Pneumatik/Zylinder) muss im JSON von Hand gesetzt werden"
|
|
% len(ils_weichen))
|
|
eintraege.extend(sensor_eintrag(t, bekannte_bloecke, hinweise)
|
|
for t in ils_weichen)
|
|
|
|
# --- Rest: Sensoren, Boegen, Schleuselemente --------------------------
|
|
for familie in ("bogen", "schleuselement", "sensor"):
|
|
posten = sorted(nach_familie.get(familie, []),
|
|
key=lambda t: (t["block"], t["x"], t["y"]))
|
|
eintraege.extend(sensor_eintrag(t, bekannte_bloecke, hinweise)
|
|
for t in posten)
|
|
|
|
return eintraege, hinweise
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Plausibilitaetspruefung der erzeugten Eintraege
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def punkte_von(eintrag):
|
|
"""Alle Koordinatentripel eines Eintrags - je Schema anders benannt."""
|
|
punkte = []
|
|
if "x" in eintrag:
|
|
punkte.append((eintrag["x"], eintrag["y"], eintrag.get("z", 0.0)))
|
|
if "start_x" in eintrag:
|
|
punkte.append((eintrag["start_x"], eintrag["start_y"],
|
|
eintrag.get("start_z", 0.0)))
|
|
if "end_x" in eintrag:
|
|
punkte.append((eintrag["end_x"], eintrag["end_y"],
|
|
eintrag.get("end_z", 0.0)))
|
|
# startpunkt: VarioFoerderer (3.1/3.2) und Gerade; start_mm:
|
|
# Gefaellestrecke (2.1); endpunkt: nur Gerade. Wird hier ein Feld eines
|
|
# neuen Eintragsschemas vergessen, liefert die Funktion eine leere Liste -
|
|
# filtere_fenster wuerde den Eintrag dann ungeprueft behalten, darum muss
|
|
# diese Liste vollstaendig bleiben.
|
|
for feld in ("startpunkt", "start_mm", "endpunkt"):
|
|
p = eintrag.get(feld)
|
|
if isinstance(p, (list, tuple)) and len(p) >= 3:
|
|
punkte.append((p[0], p[1], p[2]))
|
|
return punkte
|
|
|
|
|
|
def filtere_fenster(eintraege, fenster):
|
|
"""Behaelt die Eintraege, die den Abschnitt (x0, y0, x1, y1) beruehren.
|
|
|
|
Gefiltert wird bewusst erst hier - nach der Paarbildung in
|
|
baue_eintraege - und nicht schon auf Ebene der einzelnen Vorkommen: ein
|
|
Kreisel oder Foerderer besteht aus zwei Stationen, die weiter
|
|
auseinanderliegen koennen als das Fenster breit ist. Ein Eintrag gilt als
|
|
im Abschnitt, wenn IRGENDEINER seiner Punkte darin liegt; er darf also
|
|
ueber den Fensterrand hinausragen (der Kreisel dieser Zeichnung tut das:
|
|
Antriebsstation innen, Spannstation 23,6 m weiter aussen).
|
|
|
|
Rueckgabe: (behaltene Eintraege, Anzahl verworfener, Anzahl ohne Punkte)
|
|
Eintraege, aus denen punkte_von keine Koordinate lesen kann, werden
|
|
behalten - aber gezaehlt und gemeldet: das ist immer eine Luecke in
|
|
punkte_von und keine Eigenschaft der Zeichnung.
|
|
"""
|
|
if not fenster:
|
|
return eintraege, 0, 0
|
|
behalten = []
|
|
ohne_punkte = 0
|
|
for e in eintraege:
|
|
punkte = punkte_von(e)
|
|
if not punkte:
|
|
ohne_punkte += 1
|
|
behalten.append(e)
|
|
elif any(im_fenster(p[0], p[1], fenster) for p in punkte):
|
|
behalten.append(e)
|
|
return behalten, len(eintraege) - len(behalten), ohne_punkte
|
|
|
|
|
|
def arrays_einzeilig(text):
|
|
"""Zahlen-Arrays im JSON-Text auf eine Zeile ziehen.
|
|
|
|
ZWINGEND fuer den LISP-Treiber: ssg-parse-json-array (ssg_core.lsp) liest
|
|
die Datei ZEILENWEISE und ssg-cfg-parse-array erwartet das komplette Array
|
|
in einem String. json.dump(indent=2) bricht Arrays aber um:
|
|
|
|
"start_mm": [
|
|
219540.85,
|
|
...
|
|
],
|
|
|
|
Davon sieht der Parser nur "start_mm": [ - das ergibt nil, und der Treiber
|
|
baut das Element am Ursprung statt an seiner Position. mubea.json war von
|
|
Hand geschrieben und hatte die Arrays deshalb immer einzeilig
|
|
("start_mm": [9510, 12300, 1941]). Greift nur bei rein numerischen
|
|
Arrays - andere kommen in dieser Ausgabe nicht vor - und ist idempotent.
|
|
"""
|
|
def zusammen(m):
|
|
return "[" + " ".join(m.group(1).split()) + "]"
|
|
return re.sub(r"\[([\s\d.,+\-eE]*?)\]", zusammen, text)
|
|
|
|
|
|
def pruefe_plausibilitaet(eintraege):
|
|
"""Prueft Hoehen und Ausdehnung der Eintraege auf Unmoeglichkeiten.
|
|
|
|
Rueckgabe: Liste von Warntexten (leer = alles im Rahmen). Es wird nichts
|
|
aussortiert - die Pruefung soll nur verhindern, dass eine JSON mit
|
|
offensichtlich falsch transformierten Koordinaten still geschrieben und
|
|
danach fuer einen Testfall verwendet wird.
|
|
"""
|
|
warnungen = []
|
|
punkte = [p for e in eintraege for p in punkte_von(e)]
|
|
if not punkte:
|
|
return ["keine Koordinaten in den Eintraegen gefunden"]
|
|
|
|
zu_tief = [p for p in punkte if p[2] < Z_PLAUSIBEL[0]]
|
|
zu_hoch = [p for p in punkte if p[2] > Z_PLAUSIBEL[1]]
|
|
if zu_tief or zu_hoch:
|
|
warnungen.append(
|
|
"%d von %d Koordinaten liegen ausserhalb des plausiblen "
|
|
"Z-Fensters %.0f..%.0f mm (%d zu tief, min %.1f; %d zu hoch, "
|
|
"max %.1f) - Hinweis auf eine falsch aufgeloeste Transformation"
|
|
% (len(zu_tief) + len(zu_hoch), len(punkte), Z_PLAUSIBEL[0],
|
|
Z_PLAUSIBEL[1], len(zu_tief), min(p[2] for p in punkte),
|
|
len(zu_hoch), max(p[2] for p in punkte)))
|
|
|
|
breite = max(p[0] for p in punkte) - min(p[0] for p in punkte)
|
|
hoehe = max(p[1] for p in punkte) - min(p[1] for p in punkte)
|
|
if breite > XY_MAX_AUSDEHNUNG or hoehe > XY_MAX_AUSDEHNUNG:
|
|
warnungen.append(
|
|
"XY-Ausdehnung %.0f x %.0f mm ueberschreitet die plausible "
|
|
"Anlagengroesse von %.0f mm" % (breite, hoehe, XY_MAX_AUSDEHNUNG))
|
|
return warnungen
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(
|
|
description="Erkennt ILS-/Omniflo-Komponenten in einem fremden DXF "
|
|
"und schreibt sie im JSON-Schema aus "
|
|
"tests/testdata/object_data.md.")
|
|
parser.add_argument("dxf", help="Quellzeichnung (binaeres DXF)")
|
|
parser.add_argument("data_dir", help="Datenverzeichnis mit json/")
|
|
parser.add_argument("output", help="Ziel-JSON")
|
|
parser.add_argument("--fenster", nargs=4, type=float,
|
|
metavar=("X0", "Y0", "X1", "Y1"),
|
|
help="Nur Bauteile in diesem XY-Fenster (mm) "
|
|
"ausgeben - fuer den Nachbau eines einzelnen "
|
|
"Anlagenabschnitts statt der ganzen Zeichnung")
|
|
args = parser.parse_args()
|
|
dxf_path, data_dir, output_path = args.dxf, args.data_dir, args.output
|
|
fenster = None
|
|
if args.fenster:
|
|
x0, y0, x1, y1 = args.fenster
|
|
fenster = (min(x0, x1), min(y0, y1), max(x0, x1), max(y0, y1))
|
|
|
|
if not os.path.isfile(dxf_path):
|
|
print("%s FEHLER: DXF nicht gefunden: %s" % (PREFIX, dxf_path))
|
|
sys.exit(1)
|
|
|
|
json_dir = os.path.join(data_dir, "json")
|
|
defaults = load_json(os.path.join(json_dir, "component_defaults.json"))
|
|
boegen = load_json(os.path.join(json_dir, "omniflo_boegen.json"))
|
|
weichen = load_json(os.path.join(json_dir, "omniflo_weichen.json"))
|
|
bekannte_bloecke = lade_bekannte_bloecke(data_dir)
|
|
print("%s %d Blocknamen in %s/ils gefunden" % (PREFIX,
|
|
len(bekannte_bloecke),
|
|
data_dir))
|
|
|
|
groesse = os.path.getsize(dxf_path) / (1024.0 * 1024.0)
|
|
print("%s lese %s (%.1f MB) ..." % (PREFIX, dxf_path, groesse))
|
|
ms_inserts, block_inserts, block_base, statistik = lese_inserts(dxf_path)
|
|
print("%s %d INSERTs im Modelspace, %d Blockdefinitionen mit Inhalt"
|
|
% (PREFIX, len(ms_inserts), len(block_inserts)))
|
|
print("%s %d INSERTs gesamt, davon %d mit gekippter OCS (%.1f%%), "
|
|
"%d im Papierbereich (ignoriert)"
|
|
% (PREFIX, statistik["inserts"], statistik["ocs_gekippt"],
|
|
100.0 * statistik["ocs_gekippt"] / max(statistik["inserts"], 1),
|
|
statistik["papierbereich"]))
|
|
|
|
treffer, unbekannt, verworfen = sammle_vorkommen(ms_inserts, block_inserts,
|
|
block_base)
|
|
print("%s %d erkannte Bauteil-Vorkommen (%d deckungsgleiche Duplikate "
|
|
"verworfen)" % (PREFIX, len(treffer), verworfen))
|
|
mit_attrib = sum(1 for t in treffer if t["attribs"])
|
|
print("%s %d davon mit Blockattributen" % (PREFIX, mit_attrib))
|
|
|
|
zaehler = {}
|
|
for t in treffer:
|
|
zaehler[t["familie"]] = zaehler.get(t["familie"], 0) + 1
|
|
for familie in sorted(zaehler):
|
|
print("%s %-20s %d" % (PREFIX, familie, zaehler[familie]))
|
|
|
|
eintraege, hinweise = baue_eintraege(treffer, defaults, boegen, weichen,
|
|
bekannte_bloecke)
|
|
|
|
if fenster:
|
|
vorher = len(eintraege)
|
|
eintraege, weggefallen, ohne_punkte = filtere_fenster(eintraege,
|
|
fenster)
|
|
print("%s Abschnitt X %.0f..%.0f / Y %.0f..%.0f mm: %d von %d "
|
|
"Eintraegen behalten (%d ausserhalb)"
|
|
% (PREFIX, fenster[0], fenster[2], fenster[1], fenster[3],
|
|
len(eintraege), vorher, weggefallen))
|
|
if ohne_punkte:
|
|
print("%s ! %d Eintraege ohne lesbare Koordinaten wurden "
|
|
"ungeprueft behalten - punkte_von ergaenzen"
|
|
% (PREFIX, ohne_punkte))
|
|
|
|
warnungen = pruefe_plausibilitaet(eintraege)
|
|
if warnungen:
|
|
print("%s PLAUSIBILITAET - die Ausgabe ist mit Vorsicht zu "
|
|
"verwenden:" % PREFIX)
|
|
for w in warnungen:
|
|
print("%s ! %s" % (PREFIX, w))
|
|
else:
|
|
print("%s Plausibilitaetspruefung (Z-Fenster, XY-Ausdehnung): ok"
|
|
% PREFIX)
|
|
|
|
ziel_dir = os.path.dirname(os.path.abspath(output_path))
|
|
if ziel_dir and not os.path.isdir(ziel_dir):
|
|
os.makedirs(ziel_dir)
|
|
with open(output_path, "w", encoding="utf-8") as fh:
|
|
fh.write(arrays_einzeilig(json.dumps(eintraege, ensure_ascii=False,
|
|
indent=2)))
|
|
fh.write("\n")
|
|
print("%s %d Eintraege geschrieben -> %s" % (PREFIX, len(eintraege), output_path))
|
|
|
|
if hinweise:
|
|
# Die Hinweise entstehen bei der Paarbildung, also VOR dem
|
|
# Abschnittsfilter - sie beziehen sich damit auf die ganze Zeichnung
|
|
# und nicht nur auf den ausgegebenen Abschnitt.
|
|
print("%s Hinweise (%d, ganze Zeichnung%s):"
|
|
% (PREFIX, len(hinweise),
|
|
" - nicht nur der Abschnitt" if fenster else ""))
|
|
for h in hinweise:
|
|
print("%s - %s" % (PREFIX, h))
|
|
|
|
if unbekannt:
|
|
top = sorted(unbekannt.items(), key=lambda kv: -kv[1])[:25]
|
|
print("%s nicht zugeordnete Blocknamen (Top %d von %d) - ggf. "
|
|
"FAMILIEN ergaenzen:" % (PREFIX, len(top), len(unbekannt)))
|
|
for name, anzahl in top:
|
|
print("%s %-50s %d" % (PREFIX, name, anzahl))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|