hw_import --dxf: gemeinsame Zeichnung (Layout + E-Planung) als Hardwarequelle

hw_import liest mit --dxf statt der Positionsdatei eine DXF, in der Layout
und E-Planung zusammen liegen (Dematic): jeder Block mit Attribut IO ist ein
E-Bauteil (IO/VERW/BEZEICHNUNG/KENNZEICHNUNG, Einfuegepunkt), schon in
Layout-Koordinaten (HardwareDocument.layoutCoordinates). --conveyor-from-motor
nimmt den Foerderer aus der Motornummer, wenn kein K0102/#106 dasteht
("Motor KR-MA0055" -> K0055); "ILS-VARIO" zaehlt als Strecke.

merge_hw_layout: bei layoutCoordinates kein Lagebezug (Identitaet, nichts
abgelegt); --rules findet einen blossen Namen in SKEL_CFG. Neue Regeldatei
cfg/e_planning_dematic.ini (Sensor/Stopper gleiche Nummer, Weichenventil =
Nummer des Stausensors, Abstaende an dematic_merge..dxf gemessen).

Fehler behoben: zwei Antriebs-TROs auf einem Layout-Foerderer (Dematic VF_6,
TRO04/TRO06) bekamen beide denselben Vario-Motor. Jetzt Widerspruch mit
Vorschlag; scl_context faellt dann nicht auf den Motor des Foerderers zurueck.

Dematic: 115 Namen im FB_Main (alle im TIA-Import), offene Stellen 730 -> 615.
cfg/dematic_merge_tro_ids.ini: TRO-IDs der Dematic-Anlage.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
2026-10-21 18:21:20 +02:00
parent d414bd6cf3
commit 2fbd2f119d
10 changed files with 467 additions and 38 deletions
+119 -16
View File
@@ -16,13 +16,19 @@ genuegen (Befund doc/E-Planung_Bruecke.md, "Nachpruefung 2026-10-15"):
Kabellaengen-Tools (optional) - je BMK die Lage in der E-Zeichnung, die
Verwendung (VERW, z. B. "IN SEP"), Klartext und E/A-Station.
* **Gemeinsame Zeichnung** ``*.dxf`` (statt der Positionsdatei) - Layout
und E-Planung in einer DXF (Dematic): jeder Block mit Attribut ``IO`` ist
ein E-Bauteil (IO = BMK, VERW, BEZEICHNUNG, KENNZEICHNUNG, Einfuegepunkt).
Die Lage steht schon in Layout-Koordinaten; merge_hw_layout ermittelt
dann keinen Lagebezug (``layoutCoordinates``).
EA-Liste und WSCAD-Import bringen bei vorhandener Positionsdatei nichts
dazu und werden nicht gelesen.
Ergebnis ``HardwareDocument`` (Schema sps_skel-hw-v1): je BMK ein Geraet mit
beiden Seiten, die Antriebe (Motor MA <-> Motorschutz FC, Geber GF) je
Foerderer der E-Planung (K0102; "#106" wird K0106), und die Gegenpruefung
beider Quellen.
Foerderer der E-Planung (K0102; "#106" wird K0106; mit --conveyor-from-motor
auch "MA0055" -> K0055), und die Gegenpruefung beider Quellen.
Die Zuordnung zum Layout macht merge_hw_layout.py.
Die xlsx-Datei wird ohne Fremdbibliothek gelesen (zipfile + XML): der
@@ -59,6 +65,7 @@ __all__ = [
"read_xlsx",
"load_tia",
"load_positionsdraw",
"load_dxf",
"pair_drives",
"conveyor_of",
"conveyor_kind_of",
@@ -79,10 +86,13 @@ _ADDRESS = re.compile(r"^%?([EIAQM])\s*(\d+)\.(\d)$", re.IGNORECASE)
_CONVEYOR_K = re.compile(r"(?<![A-Z0-9])K(\d{4})(?!\d)")
# ... bzw. "CONVEYOR #100", "CONVEYOR-#106", "CONVEYOR#110"
_CONVEYOR_HASH = re.compile(r"#\s*(\d{3})(?!\d)")
# ... bzw. ueber den Motor (Dematic, nur mit motor=True): "Motor KR-MA0055",
# "Rad dreht ILS-KR (MA0046)", "ILS-VARIO-MA0043" -> K0055 / K0046 / K0043
_CONVEYOR_MA = re.compile(r"(?<![A-Z0-9])MA(\d{4})(?!\d)")
# Art des Foerderers im Klartext: "KREISEL-K0102", "KR-K0104" -> Kreisel;
# "CONVEYOR #106", "KLINKENFOERDERER" -> Strecke
# "CONVEYOR #106", "KLINKENFOERDERER", "ILS-VARIO" -> Strecke
_KIND_KREISEL = re.compile(r"(?<![A-Z])(KREISEL|KR)(?![A-Z])")
_KIND_STRECKE = re.compile(r"CONVEYOR|CONVEYER|F(OE|Ö)RDERER")
_KIND_STRECKE = re.compile(r"CONVEYOR|CONVEYER|F(OE|Ö)RDERER|(?<![A-Z])VARIO(?![A-Z])")
# ---------------------------------------------------------------------------
@@ -207,6 +217,9 @@ class HardwareDocument(BaseModel):
tia_sha256: str = Field(default="", alias="tiaSha256")
positions_file: str = Field(default="", alias="positionsFile")
positions_sha256: str = Field(default="", alias="positionsSha256")
# True: die Lage kommt aus der gemeinsamen Zeichnung (load_dxf) und steht
# schon in Layout-Koordinaten - merge_hw_layout ermittelt keinen Lagebezug
layout_coordinates: bool = Field(default=False, alias="layoutCoordinates")
devices: list[HwDevice] = Field(default_factory=list)
drives: list[Drive] = Field(default_factory=list)
# Konstanten ohne passendes Signal (z. B. Projektkonstanten)
@@ -416,6 +429,44 @@ def load_positionsdraw(path: Path, warnings: list[str] | None = None) -> list[ED
return list(devices.values())
def load_dxf(path: Path, warnings: list[str] | None = None) -> list[EDevice]:
"""E-Bauteile aus einer gemeinsamen Zeichnung (Layout + E-Planung) lesen.
Jeder Block im Modellbereich mit einem BMK im Attribut ``IO`` ist ein
E-Bauteil; die Lage ist der Einfuegepunkt - schon in Layout-Koordinaten.
Bloecke ohne IO (Layout-Bauteile, Busverteiler, die Motorkennzeichnung
"ILS-Kreisel MA0053") zaehlen nicht: der Motor steht im Motorsymbol.
Steht ein BMK doppelt da, gewinnt der Eintrag mit VERW.
"""
import ezdxf
warnings = warnings if warnings is not None else []
doc = ezdxf.readfile(str(path))
devices: dict[str, EDevice] = {}
count: Counter = Counter()
for insert in doc.modelspace().query("INSERT"):
attrs = {a.dxf.tag.upper(): a.dxf.text.strip() for a in insert.attribs}
bmk = attrs.get("IO", "")
if not _BMK.match(bmk):
continue
count[bmk] += 1
dev = EDevice(
bmk=bmk, verw=attrs.get("VERW", ""), bezeichnung=attrs.get("BEZEICHNUNG", ""),
kennzeichnung=attrs.get("KENNZEICHNUNG", ""),
pos=(float(insert.dxf.insert.x), float(insert.dxf.insert.y)))
old = devices.get(bmk)
if old is None or (not old.verw and dev.verw):
devices[bmk] = dev
for bmk, n in sorted(count.items()):
if n > 1:
warnings.append(f"Zeichnung: {bmk} steht {n}x in der Zeichnung "
f"- verwendet wird ein Eintrag")
if not devices:
raise ValueError(f"{path.name}: kein Block mit BMK im Attribut 'IO' gefunden "
f"- ist das die Zeichnung mit der E-Planung?")
return list(devices.values())
def _positions_warnings(path: Path) -> list[str]:
"""Die Hinweise, die das Kabellaengen-Tool selbst in die Datei schreibt."""
data = json.loads(path.read_text(encoding="utf-8"))
@@ -433,13 +484,18 @@ def _positions_warnings(path: Path) -> list[str]:
# ---------------------------------------------------------------------------
def conveyor_of(*texts: str) -> str:
def conveyor_of(*texts: str, motor: bool = False) -> str:
"""Foerderer der E-Planung aus einem Klartext als "K<NNNN>", sonst "".
Die E-Planung schreibt denselben Foerderer zweifach: "CONVEYOR-K0103"
(Motorschutz FC0203) und "CONVEYOR-#103" (Motor MA0203). Beides ist die
vierstellige Foerderernummer des SCL-Namensschemas (DB_JamArea0103,
fbCarousel0102) und wird zu "K0103".
``motor=True`` (Dematic): ohne K/# nennt der Klartext nur den Motor
("Motor KR-MA0055", "Rad dreht ILS-KR (MA0046)"); dann ist die
Motornummer die Foerderernummer -> "K0055". Bei Mubea waere das falsch
(Motor MA0203 treibt K0103), deshalb nur auf Wunsch.
"""
for text in texts:
upper = (text or "").upper()
@@ -449,6 +505,11 @@ def conveyor_of(*texts: str) -> str:
m = _CONVEYOR_HASH.search(upper)
if m:
return f"K{int(m.group(1)):04d}"
if motor:
for text in texts:
m = _CONVEYOR_MA.search((text or "").upper())
if m:
return f"K{m.group(1)}"
return ""
@@ -497,10 +558,18 @@ def _constant_name(bmk: str, direction: str) -> str:
def build_hardware(tia_path: Path | None, positions_path: Path | None,
project: str = "") -> HardwareDocument:
"""Beide Quellen einlesen, je BMK zusammenfuehren und gegeneinander pruefen."""
if tia_path is None and positions_path is None:
raise ValueError("weder TIA-Import noch Positionsdatei angegeben")
project: str = "", dxf_path: Path | None = None,
conveyor_from_motor: bool = False) -> HardwareDocument:
"""Beide Quellen einlesen, je BMK zusammenfuehren und gegeneinander pruefen.
Die Lage kommt aus der Positionsdatei oder aus der gemeinsamen Zeichnung
``dxf_path`` (load_dxf), nicht aus beiden. ``conveyor_from_motor``: siehe
conveyor_of(motor=True).
"""
if tia_path is None and positions_path is None and dxf_path is None:
raise ValueError("weder TIA-Import noch Positionsdatei/Zeichnung angegeben")
if positions_path is not None and dxf_path is not None:
raise ValueError("Positionsdatei und Zeichnung zugleich - nur eine Quelle fuer die Lage")
warnings: list[str] = []
checks: list[str] = []
@@ -512,6 +581,9 @@ def build_hardware(tia_path: Path | None, positions_path: Path | None,
if positions_path is not None:
e_devices = load_positionsdraw(positions_path, warnings)
warnings.extend(_positions_warnings(positions_path))
elif dxf_path is not None:
e_devices = load_dxf(dxf_path, warnings)
positions_path = dxf_path
tag_of = {}
for t in tags:
@@ -546,7 +618,8 @@ def build_hardware(tia_path: Path | None, positions_path: Path | None,
comment=tag.comment if tag else "",
constant=cname if const else "",
constant_value=value,
conveyor=conveyor_of(e.bezeichnung if e else "", tag.comment if tag else ""),
conveyor=conveyor_of(e.bezeichnung if e else "", tag.comment if tag else "",
motor=conveyor_from_motor),
conveyor_kind=conveyor_kind_of(e.bezeichnung if e else "",
tag.comment if tag else ""),
sources=[s for s, ok in (("tia", tag is not None), ("positions", e is not None))
@@ -561,6 +634,7 @@ def build_hardware(tia_path: Path | None, positions_path: Path | None,
tia_sha256=file_fingerprint(tia_path) if tia_path else "",
positions_file=positions_path.name if positions_path else "",
positions_sha256=file_fingerprint(positions_path) if positions_path else "",
layout_coordinates=dxf_path is not None,
devices=devices,
drives=pair_drives(devices),
constants=[c for c in constants if c.name not in used_constants],
@@ -627,7 +701,10 @@ def render_report(doc: HardwareDocument) -> str:
add(f"# Hardwareplanung: {doc.project}")
add("")
add(f"- TIA-Import: `{doc.tia_file or '-'}`")
add(f"- Positionsdatei: `{doc.positions_file or '-'}`")
if doc.layout_coordinates:
add(f"- Gemeinsame Zeichnung (Lage in Layout-Koordinaten): `{doc.positions_file}`")
else:
add(f"- Positionsdatei: `{doc.positions_file or '-'}`")
add(f"- Erzeugt: {doc.created}")
add("")
add("## Geraete")
@@ -699,20 +776,30 @@ def project_name(tia_path: Path | None, positions_path: Path | None) -> str:
def main(argv: list[str] | None = None) -> int:
import ezdxf
from material_flow import env_dir, resolve_input
args = parse_args(argv)
tia_path = resolve_input(args.tia) if args.tia else None
positions_path = resolve_input(args.positions) if args.positions else None
for label, path in (("TIA-Import", tia_path), ("Positionsdatei", positions_path)):
dxf_path = resolve_input(args.dxf) if args.dxf else None
if positions_path is not None and dxf_path is not None:
print("FEHLER: --positions und --dxf zugleich - nur eine Quelle fuer die Lage",
file=sys.stderr)
return 1
for label, path in (("TIA-Import", tia_path), ("Positionsdatei", positions_path),
("Zeichnung", dxf_path)):
if path is not None and not path.is_file():
print(f"FEHLER: {label} nicht gefunden: {path}", file=sys.stderr)
return 1
project = args.project or project_name(tia_path, positions_path)
project = args.project or project_name(tia_path, positions_path or dxf_path)
try:
doc = build_hardware(tia_path, positions_path, project)
except (OSError, ValueError, KeyError, zipfile.BadZipFile, ET.ParseError) as exc:
doc = build_hardware(tia_path, positions_path, project, dxf_path,
args.conveyor_from_motor)
except (OSError, ValueError, KeyError, zipfile.BadZipFile, ET.ParseError,
ezdxf.DXFError) as exc:
print(f"FEHLER: {exc}", file=sys.stderr)
return 1
@@ -730,7 +817,10 @@ def main(argv: list[str] | None = None) -> int:
print("HARDWAREPLANUNG EINLESEN")
print("================================================================")
print(f"TIA-Import = {tia_path or '-'}")
print(f"Positionsdatei = {positions_path or '-'}")
if dxf_path is not None:
print(f"Zeichnung = {dxf_path} (Lage in Layout-Koordinaten)")
else:
print(f"Positionsdatei = {positions_path or '-'}")
print(f"Projekt = {project}")
print(f"Geraete = {len(doc.devices)} (TIA {n_tia}, Positionsdatei {n_pos}, "
f"beide {sum(1 for d in doc.devices if len(d.sources) == 2)})")
@@ -763,6 +853,8 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
" hw_import.bat --tia Mubea/mubea-1_TIA.xlsx "
"--positions Mubea/Mubea_positionsconv.json\n"
" hw_import.bat --tia mubea-1_TIA.xlsx --project Mubea\n"
" hw_import.bat --tia dematic_hw-2_TIA.xlsx --dxf dematic_merge..dxf "
"--project dematic --conveyor-from-motor\n"
"\n"
"Exit-Codes:\n"
" 0 ok (Befunde der Gegenpruefung stehen im Bericht)\n"
@@ -774,6 +866,17 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
parser.add_argument("--positions", metavar="NAME",
help="*_positionsconv.json bzw. *_positionsdraw.json in "
"%%SKEL_DATA%% (oder ein Pfad); ohne sie keine Lage")
parser.add_argument("--dxf", metavar="NAME",
help="statt --positions: gemeinsame Zeichnung (Layout + "
"E-Planung, *.dxf) in %%SKEL_DATA%% (oder ein Pfad); "
"jeder Block mit Attribut IO ist ein E-Bauteil, die Lage "
"steht schon in Layout-Koordinaten")
parser.add_argument("--conveyor-from-motor", dest="conveyor_from_motor",
action="store_true",
help="Foerderer auch aus der Motornummer im Klartext "
"(\"Motor KR-MA0055\" -> K0055), wenn kein K0102/#106 "
"dasteht (Dematic). Nicht fuer Mubea: dort treibt MA0203 "
"den Foerderer K0103.")
parser.add_argument("--project", metavar="NAME",
help="Projektname fuer <projekt>_hw.json. Standard: aus dem "
"Namen der Positionsdatei (Mubea_positionsconv.json -> Mubea)")
+48 -6
View File
@@ -371,6 +371,19 @@ def resolve_transform(points, devices, anchors, cfg_dir: Path, csv_name: str,
return transform
def _same_drawing(points: list[LayoutPoint], drawing: str) -> Transform:
"""Lagebezug einer gemeinsamen Zeichnung (hw_import --dxf): keiner.
Die E-Symbole stehen schon in Layout-Koordinaten. Ihr Versatz neben dem
Bauteil (Dematic ~600 mm) ist Zeichenabstand, keine Verschiebung - ein
eingepasster Bezug wuerde ihn nur verzerren. Er steckt in max_mm/soll_mm
der Regeln.
"""
n = sum(1 for p in points if p.kind == "Separator")
return Transform(dx=0.0, dy=0.0, source="zeichnung", anchor="gemeinsame Zeichnung",
matched=n, candidates=n, dxf_file=drawing)
# ---------------------------------------------------------------------------
# Slots
# ---------------------------------------------------------------------------
@@ -529,9 +542,20 @@ def _conveyor_kinds(hw: HardwareDocument) -> dict[str, str]:
def bind_drives(doc: TroDocument, conveyors: list[ConveyorBinding],
hw: HardwareDocument) -> list[Binding]:
"""Antriebs-Slots (Vario): Motor und Motorschutz des zugeordneten Foerderers."""
"""Antriebs-Slots (Vario): Motor und Motorschutz des zugeordneten Foerderers.
Tragen mehrere TROs einen Antriebs-Slot auf demselben Layout-Foerderer
(Dematic VF_6: zwei Vario-TROs, drei Vario-Motoren), hat jeder TRO seinen
eigenen Motor - welcher, sagt die Mehrheit je Foerderer nicht. Dann wird
der Antrieb ein Widerspruch (BMK als Vorschlag), Korrektur von Hand.
"""
devices = hw.by_bmk()
by_layout = {c.layout_id: c for c in conveyors if c.status == "zugeordnet"}
sharing: dict[str, list[str]] = {}
for tro in doc.tros:
for s in tro.slots:
if s.layout_id and any(r.rstrip("?") in _DRIVE_ROLES for r in s.roles):
sharing.setdefault(s.layout_id.split("-", 1)[0], []).append(tro.tro_id)
bindings = []
for tro in doc.tros:
for s in tro.slots:
@@ -541,17 +565,24 @@ def bind_drives(doc: TroDocument, conveyors: list[ConveyorBinding],
if conv is None:
continue
roles = [r.rstrip("?") for r in s.roles]
shared = sharing.get(s.layout_id.split("-", 1)[0], [])
for role, attr in _DRIVE_ROLES.items():
bmk = getattr(conv, attr)
if role not in roles or not bmk:
continue
dev = devices.get(bmk)
status = "zugeordnet" if dev and dev.in_tia else "widerspruch"
note = f"Antrieb von {conv.eplan_name}"
if len(shared) > 1:
status = "widerspruch"
note = (f"{conv.layout_name or conv.layout_id} traegt {len(shared)} "
f"Antriebs-TROs ({', '.join(shared)}) - Antrieb je TRO von Hand "
f"({OVERRIDES_FILE}); Vorschlag {conv.eplan_name}")
bindings.append(Binding(
tro=tro.tro_id, slot=s.slot, role=role, layout_id=s.layout_id,
bmk=bmk, constant=dev.constant if dev else "",
address=dev.address if dev else "", source="positionsdraw",
rule="mehrheit", status="zugeordnet" if dev and dev.in_tia else "widerspruch",
note=f"Antrieb von {conv.eplan_name}"))
rule="mehrheit", status=status, note=note))
return bindings
@@ -690,7 +721,9 @@ def build_anlage(doc: TroDocument, hw: HardwareDocument, rules: list[Rule],
assignments: list[Assignment] = []
conveyors: list[ConveyorBinding] = []
if hw.field_devices():
if transform is None:
if transform is None and hw.layout_coordinates:
transform = _same_drawing(points, hw.positions_file)
elif transform is None:
transform = resolve_transform(points, devices, anchors, cfg_dir, doc.source,
hw.positions_file, re_register, warnings,
hw.positions_sha256)
@@ -856,6 +889,14 @@ def _in_results(name: str, results: Path) -> Path:
return results / name
def _in_cfg(name: str, cfg_dir: Path) -> Path:
"""Regeldatei: ein blosser Name, der hier nicht liegt, steht in %SKEL_CFG%."""
path = Path(name)
if path.is_absolute() or path.parent != Path(".") or path.is_file():
return path
return cfg_dir / name
def main(argv: list[str] | None = None) -> int:
from material_flow import env_dir
@@ -864,7 +905,7 @@ def main(argv: list[str] | None = None) -> int:
cfg_dir = env_dir("SKEL_CFG", "cfg")
tro_path = _in_results(args.tro, results)
hw_path = _in_results(args.hw, results)
rules_path = Path(args.rules) if args.rules else cfg_dir / RULES_FILE
rules_path = _in_cfg(args.rules, cfg_dir) if args.rules else cfg_dir / RULES_FILE
overrides_path = Path(args.overrides) if args.overrides else cfg_dir / OVERRIDES_FILE
for label, path in (("TRO-JSON", tro_path), ("Hardware-JSON", hw_path),
("Regeln", rules_path)):
@@ -955,7 +996,8 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
help="Hardware-JSON <projekt>_hw.json in %%SKEL_RESULTS%% "
"(oder ein Pfad)")
parser.add_argument("--rules", metavar="DATEI",
help=f"Regeldatei. Standard: %%SKEL_CFG%%\\{RULES_FILE}")
help=f"Regeldatei (Name in %%SKEL_CFG%% oder ein Pfad), z. B. "
f"e_planning_dematic.ini. Standard: %%SKEL_CFG%%\\{RULES_FILE}")
parser.add_argument("--overrides", metavar="DATEI",
help=f"Handkorrekturen. Standard: %%SKEL_CFG%%\\{OVERRIDES_FILE} "
f"(fehlt sie, keine)")
+14 -3
View File
@@ -150,6 +150,13 @@ class HwSource:
geraet = self.device(tro_id, slot, role, layout)
return _bekannt(fmt.format(bmk=geraet[0])) if geraet else _offen(offen)
def conflict(self, tro_id: str, slot: str, role: str) -> bool:
"""True, wenn die Anlagen-JSON fuer die Slot-Rolle einen Widerspruch fuehrt."""
if self.anlage is None:
return False
b = self.anlage.binding(tro_id, slot, role)
return b is not None and b.status == "widerspruch"
def conveyor(self, host: str) -> ConveyorBinding | None:
"""Foerderer der Hardwareplanung (K0102) am Layout-Foerderer ("0010-R")."""
if self.anlage is None or not host:
@@ -643,7 +650,9 @@ def _drive(tro: TroBase, hw: HwSource) -> tuple[SclName, SclName]:
Erst der Antriebs-Slot (Rollen motor/motorProtection); fehlt er, der
Foerderer der Hardwareplanung am Host des TRO - der Vario-Antrieb ist
dessen Motor (MA/FC-Paar aus hw_import).
dessen Motor (MA/FC-Paar aus hw_import). Ein Widerspruch im Slot (z. B.
zwei Vario-TROs auf einem Foerderer) bleibt offen, statt den Motor des
Foerderers zu erben.
"""
motor = hw.output(tro.tro_id, "drive", "motor", offen="MA____")
schutz = hw.output(tro.tro_id, "drive", "motorProtection",
@@ -651,9 +660,11 @@ def _drive(tro: TroBase, hw: HwSource) -> tuple[SclName, SclName]:
hosts = list(getattr(tro, "hosts", []) or [])
conveyor = hw.conveyor(hosts[0]) if hosts else None
if conveyor is not None:
if motor.todo and conveyor.motor:
if (motor.todo and conveyor.motor
and not hw.conflict(tro.tro_id, "drive", "motor")):
motor = _bekannt(conveyor.motor)
if schutz.todo and conveyor.protection:
if (schutz.todo and conveyor.protection
and not hw.conflict(tro.tro_id, "drive", "motorProtection")):
schutz = _bekannt(f'"{conveyor.protection}"')
return motor, schutz