s.ayadi
2fbd2f119d
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>
888 lines
36 KiB
Python
888 lines
36 KiB
Python
# -*- coding: utf-8 -*-
|
|
# Absichtlich ohne Shebang: der Windows-py-Launcher wuerde daraus "python3"
|
|
# ableiten und scheitert, wenn keine PythonCore-Installation registriert ist.
|
|
"""
|
|
hw_import.py - Hardwareplanung einlesen: TIA-Import + Positionsdatei -> <projekt>_hw.json.
|
|
|
|
Stufe 2 des Ablaufs in doc/Konzept_Layout_Hardware_SCL.md. Zwei Quellen
|
|
genuegen (Befund doc/E-Planung_Bruecke.md, "Nachpruefung 2026-10-15"):
|
|
|
|
* **TIA-Import** ``*_TIA.xlsx`` - Blatt "PLC Tags" (Name, Datentyp,
|
|
Absolutadresse, Kommentar) und Blatt "Constants" (Index-Konstante
|
|
``cIn<BMK>``/``cOut<BMK>`` je Signal). Das ist genau das, was im SCL steht
|
|
("DB_Inputs".Sensors["cInBG0532"], "MB0533") und im TIA-Projekt existieren
|
|
muss - deshalb die erste Quelle und die Pruefliste fuer das SCL.
|
|
* **Positionsdatei** ``*_positionsconv.json`` / ``*_positionsdraw.json`` des
|
|
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; 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
|
|
TIA-Import ist eine einfache Tabelle, openpyxl waere eine Abhaengigkeit nur
|
|
dafuer.
|
|
|
|
Aufruf ueber bin/hw_import.bat bzw. bin/hw_import.sh.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import re
|
|
import sys
|
|
import zipfile
|
|
import xml.etree.ElementTree as ET
|
|
from collections import Counter
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
from pydantic import BaseModel, ConfigDict, Field
|
|
|
|
from dxf_registration import file_fingerprint
|
|
|
|
__all__ = [
|
|
"HW_SCHEMA",
|
|
"EDevice",
|
|
"Drive",
|
|
"TiaTag",
|
|
"TiaConstant",
|
|
"HwDevice",
|
|
"HardwareDocument",
|
|
"read_xlsx",
|
|
"load_tia",
|
|
"load_positionsdraw",
|
|
"load_dxf",
|
|
"pair_drives",
|
|
"conveyor_of",
|
|
"conveyor_kind_of",
|
|
"build_hardware",
|
|
]
|
|
|
|
HW_SCHEMA = "sps_skel-hw-v1"
|
|
|
|
_CONFIG = ConfigDict(extra="forbid", populate_by_name=True)
|
|
|
|
# "=A1+UH01-KF107" -> Ort/Steuerung "UH01", Geraet "KF107"
|
|
_KENNZEICHNUNG = re.compile(r"\+([A-Za-z0-9]+)-([A-Za-z0-9]+)")
|
|
# "MB0643" -> ("MB", "0643")
|
|
_BMK = re.compile(r"^([A-Za-z]+)(\d+)$")
|
|
# "%E053.2" / "%I53.2" / "%A020.0" / "%Q20.0"
|
|
_ADDRESS = re.compile(r"^%?([EIAQM])\s*(\d+)\.(\d)$", re.IGNORECASE)
|
|
# Foerderer im Klartext der E-Planung: "KREISEL-K0102", "KR-K0104", "K0105"
|
|
_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", "ILS-VARIO" -> Strecke
|
|
_KIND_KREISEL = re.compile(r"(?<![A-Z])(KREISEL|KR)(?![A-Z])")
|
|
_KIND_STRECKE = re.compile(r"CONVEYOR|CONVEYER|F(OE|Ö)RDERER|(?<![A-Z])VARIO(?![A-Z])")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Modell
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class EDevice(BaseModel):
|
|
"""Ein Sensor/Aktor/Schrankelement der E-Planung (Positionsdatei)."""
|
|
|
|
model_config = _CONFIG
|
|
|
|
bmk: str
|
|
verw: str = ""
|
|
bezeichnung: str = ""
|
|
kennzeichnung: str = ""
|
|
# Position in der E-Zeichnung (mm); None bei Schrankelementen ohne Lage
|
|
pos: tuple[float, float] | None = None
|
|
# True: aus "schaltschrank_elemente" (im Schrank, nicht im Feld)
|
|
cabinet: bool = False
|
|
|
|
@property
|
|
def controller(self) -> str:
|
|
"""Steuerung/Ort aus der Kennzeichnung ("UH01"), "" wenn unbekannt."""
|
|
m = _KENNZEICHNUNG.search(self.kennzeichnung)
|
|
return m.group(1) if m else ""
|
|
|
|
@property
|
|
def station(self) -> str:
|
|
"""E/A-Station aus der Kennzeichnung ("KF107"), "" wenn unbekannt."""
|
|
m = _KENNZEICHNUNG.search(self.kennzeichnung)
|
|
return m.group(2) if m else ""
|
|
|
|
@property
|
|
def prefix(self) -> str:
|
|
m = _BMK.match(self.bmk)
|
|
return m.group(1) if m else ""
|
|
|
|
@property
|
|
def number(self) -> int | None:
|
|
m = _BMK.match(self.bmk)
|
|
return int(m.group(2)) if m else None
|
|
|
|
|
|
class Drive(BaseModel):
|
|
"""Motor und sein Motorschutz (gleiche Nummer), dazu Foerderer und Geber."""
|
|
|
|
model_config = _CONFIG
|
|
|
|
motor: str
|
|
protection: str = ""
|
|
bezeichnung: str = ""
|
|
protection_bezeichnung: str = ""
|
|
# Foerderer der E-Planung aus dem Klartext ("K0102", "#106" -> "K0106")
|
|
conveyor: str = ""
|
|
# "Kreisel" | "Strecke" | "" aus dem Klartext (conveyor_kind_of)
|
|
conveyor_kind: str = Field(default="", alias="conveyorKind")
|
|
# Inkrementalgeber (GF...) desselben Foerderers
|
|
encoder: str = ""
|
|
|
|
|
|
class TiaTag(BaseModel):
|
|
"""Eine Zeile aus dem Blatt "PLC Tags" des TIA-Imports."""
|
|
|
|
model_config = _CONFIG
|
|
|
|
name: str
|
|
data_type: str = Field(default="", alias="dataType")
|
|
address: str = ""
|
|
comment: str = ""
|
|
|
|
|
|
class TiaConstant(BaseModel):
|
|
"""Eine Zeile aus dem Blatt "Constants" des TIA-Imports."""
|
|
|
|
model_config = _CONFIG
|
|
|
|
name: str
|
|
data_type: str = Field(default="", alias="dataType")
|
|
value: str = ""
|
|
comment: str = ""
|
|
|
|
|
|
class HwDevice(EDevice):
|
|
"""
|
|
Ein Geraet der Hardwareplanung: Positionsdatei und TIA-Import zusammen.
|
|
|
|
Die Felder aus EDevice kommen aus der Positionsdatei (leer, wenn das BMK
|
|
dort fehlt), Adresse/Konstante aus dem TIA-Import (leer, wenn es dort
|
|
fehlt). ``sources`` sagt, wo das BMK steht.
|
|
"""
|
|
|
|
# "in" | "out" | "" (Merker, unbekannt)
|
|
direction: str = ""
|
|
address: str = ""
|
|
data_type: str = Field(default="", alias="dataType")
|
|
comment: str = ""
|
|
# Index-Konstante fuer "DB_Inputs".Sensors[...] bzw. Ausgaenge
|
|
constant: str = ""
|
|
constant_value: int | None = Field(default=None, alias="constantValue")
|
|
# Foerderer der E-Planung aus Bezeichnung/Kommentar ("K0102"; "#106" -> "K0106")
|
|
conveyor: str = ""
|
|
# "Kreisel" | "Strecke" | "" - nur aus dem eigenen Klartext
|
|
conveyor_kind: str = Field(default="", alias="conveyorKind")
|
|
# "tia", "positions"
|
|
sources: list[str] = Field(default_factory=list)
|
|
|
|
@property
|
|
def in_tia(self) -> bool:
|
|
return "tia" in self.sources
|
|
|
|
|
|
class HardwareDocument(BaseModel):
|
|
"""Die Hardware-JSON ``<projekt>_hw.json``."""
|
|
|
|
model_config = ConfigDict(extra="forbid", populate_by_name=True)
|
|
|
|
schema_id: str = Field(default=HW_SCHEMA, alias="schema")
|
|
created: str = Field(default_factory=lambda: datetime.now().strftime("%Y-%m-%d %H:%M"))
|
|
project: str = ""
|
|
tia_file: str = Field(default="", alias="tiaFile")
|
|
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)
|
|
constants: list[TiaConstant] = Field(default_factory=list)
|
|
# Gegenpruefung TIA <-> Positionsdatei (je Befund eine Zeile)
|
|
checks: list[str] = Field(default_factory=list)
|
|
# Hinweise beim Einlesen (auch die der Positionsdatei selbst)
|
|
warnings: list[str] = Field(default_factory=list)
|
|
|
|
def by_bmk(self) -> dict[str, HwDevice]:
|
|
return {d.bmk: d for d in self.devices}
|
|
|
|
def field_devices(self) -> list[HwDevice]:
|
|
"""Geraete im Feld mit Lage in der E-Zeichnung."""
|
|
return [d for d in self.devices if d.pos is not None and not d.cabinet]
|
|
|
|
def to_json(self) -> str:
|
|
return self.model_dump_json(by_alias=True, indent=2) + "\n"
|
|
|
|
@classmethod
|
|
def from_json(cls, text: str) -> HardwareDocument:
|
|
return cls.model_validate_json(text)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# xlsx lesen (ohne openpyxl)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_NS = {
|
|
"m": "http://schemas.openxmlformats.org/spreadsheetml/2006/main",
|
|
"r": "http://schemas.openxmlformats.org/officeDocument/2006/relationships",
|
|
}
|
|
_T = "{%s}t" % _NS["m"]
|
|
|
|
|
|
def _column(ref: str) -> int:
|
|
"""'C12' -> 2 (Spalte, 0-basiert)."""
|
|
n = 0
|
|
for ch in re.match(r"[A-Z]+", ref).group():
|
|
n = n * 26 + ord(ch) - 64
|
|
return n - 1
|
|
|
|
|
|
def read_xlsx(path: Path) -> dict[str, list[list[str]]]:
|
|
"""Alle Blaetter einer xlsx-Datei als Zeilen von Texten ("" = leer).
|
|
|
|
Reicht fuer Tabellen-Exporte (TIA, EA-Liste): Werte, keine Formeln,
|
|
keine Formate. Leere Zeilen fallen weg.
|
|
"""
|
|
with zipfile.ZipFile(path) as z:
|
|
shared: list[str] = []
|
|
if "xl/sharedStrings.xml" in z.namelist():
|
|
root = ET.fromstring(z.read("xl/sharedStrings.xml"))
|
|
shared = ["".join(t.text or "" for t in si.iter(_T))
|
|
for si in root.findall("m:si", _NS)]
|
|
workbook = ET.fromstring(z.read("xl/workbook.xml"))
|
|
rels = ET.fromstring(z.read("xl/_rels/workbook.xml.rels"))
|
|
targets = {r.get("Id"): r.get("Target") for r in rels}
|
|
sheets: dict[str, list[list[str]]] = {}
|
|
for sheet in workbook.find("m:sheets", _NS):
|
|
target = targets[sheet.get("{%s}id" % _NS["r"])].lstrip("/")
|
|
target = target if target.startswith("xl/") else "xl/" + target
|
|
rows = []
|
|
for row in ET.fromstring(z.read(target)).iter("{%s}row" % _NS["m"]):
|
|
cells: dict[int, str] = {}
|
|
for c in row.findall("m:c", _NS):
|
|
v = c.find("m:v", _NS)
|
|
kind = c.get("t")
|
|
if kind == "s" and v is not None:
|
|
text = shared[int(v.text)]
|
|
elif kind == "inlineStr":
|
|
text = "".join(t.text or "" for t in c.iter(_T))
|
|
else:
|
|
text = v.text if v is not None and v.text is not None else ""
|
|
if text != "":
|
|
cells[_column(c.get("r"))] = text
|
|
if cells:
|
|
rows.append([cells.get(i, "") for i in range(max(cells) + 1)])
|
|
sheets[sheet.get("name")] = rows
|
|
return sheets
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# TIA-Import
|
|
# ---------------------------------------------------------------------------
|
|
|
|
# Spaltenkoepfe im englischen und deutschen TIA-Export
|
|
_TAG_HEADERS = {
|
|
"name": ("Name",),
|
|
"data_type": ("Data Type", "Datentyp"),
|
|
"address": ("Logical Address", "Logische Adresse", "Adresse"),
|
|
"comment": ("Comment", "Kommentar"),
|
|
}
|
|
_CONST_HEADERS = {
|
|
"name": ("Name",),
|
|
"data_type": ("Data Type", "Datentyp"),
|
|
"value": ("Value", "Wert"),
|
|
"comment": ("Comment", "Kommentar"),
|
|
}
|
|
|
|
|
|
def _columns(header: list[str], wanted: dict[str, tuple[str, ...]]) -> dict[str, int] | None:
|
|
"""Spaltennummern nach Kopfzeile; None, wenn "name" fehlt."""
|
|
lower = [h.strip().lower() for h in header]
|
|
found = {}
|
|
for key, names in wanted.items():
|
|
for n in names:
|
|
if n.lower() in lower:
|
|
found[key] = lower.index(n.lower())
|
|
break
|
|
return found if "name" in found else None
|
|
|
|
|
|
def _cell(row: list[str], index: int | None) -> str:
|
|
return row[index].strip() if index is not None and index < len(row) else ""
|
|
|
|
|
|
def load_tia(path: Path, warnings: list[str] | None = None
|
|
) -> tuple[list[TiaTag], list[TiaConstant]]:
|
|
"""Tags und Konstanten aus einem TIA-Import (xlsx).
|
|
|
|
Die Blaetter werden an der Kopfzeile erkannt, nicht am Namen: ein Blatt
|
|
mit Adressspalte sind Tags, eines mit Wertspalte Konstanten.
|
|
"""
|
|
warnings = warnings if warnings is not None else []
|
|
tags: list[TiaTag] = []
|
|
constants: list[TiaConstant] = []
|
|
for sheet, rows in read_xlsx(path).items():
|
|
if not rows:
|
|
continue
|
|
tag_cols = _columns(rows[0], _TAG_HEADERS)
|
|
const_cols = _columns(rows[0], _CONST_HEADERS)
|
|
if tag_cols and "address" in tag_cols:
|
|
for row in rows[1:]:
|
|
name = _cell(row, tag_cols["name"])
|
|
if name:
|
|
tags.append(TiaTag(
|
|
name=name, data_type=_cell(row, tag_cols.get("data_type")),
|
|
address=_cell(row, tag_cols.get("address")),
|
|
comment=_cell(row, tag_cols.get("comment"))))
|
|
elif const_cols and "value" in const_cols:
|
|
for row in rows[1:]:
|
|
name = _cell(row, const_cols["name"])
|
|
if name:
|
|
constants.append(TiaConstant(
|
|
name=name, data_type=_cell(row, const_cols.get("data_type")),
|
|
value=_cell(row, const_cols.get("value")),
|
|
comment=_cell(row, const_cols.get("comment"))))
|
|
else:
|
|
warnings.append(f"TIA-Import: Blatt '{sheet}' weder Tags noch Konstanten "
|
|
f"- uebersprungen")
|
|
if not tags:
|
|
raise ValueError(f"{path.name}: kein Blatt mit Tags (Spalte 'Logical Address' "
|
|
f"bzw. 'Logische Adresse') gefunden")
|
|
return tags, constants
|
|
|
|
|
|
def _direction(address: str) -> str:
|
|
m = _ADDRESS.match(address.strip())
|
|
if not m:
|
|
return ""
|
|
area = m.group(1).upper()
|
|
return "in" if area in "EI" else "out" if area in "AQ" else ""
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Positionsdatei (Kabellaengen-Tool)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _device(key: str, raw: dict, cabinet: bool) -> EDevice:
|
|
# Schluessel "BG0642@UNKNOWN" / "QA7000@A/B": vor dem @ steht das BMK.
|
|
bmk = str(raw.get("IO") or raw.get("B") or key.split("@", 1)[0]).strip()
|
|
pos = raw.get("pos")
|
|
return EDevice(
|
|
bmk=bmk,
|
|
verw=str(raw.get("VERW") or "").strip(),
|
|
bezeichnung=str(raw.get("BEZEICHNUNG") or "").strip(),
|
|
kennzeichnung=str(raw.get("KENNZEICHNUNG") or "").strip(),
|
|
pos=(float(pos[0]), float(pos[1])) if pos and len(pos) >= 2 else None,
|
|
cabinet=cabinet,
|
|
)
|
|
|
|
|
|
def load_positionsdraw(path: Path, warnings: list[str] | None = None) -> list[EDevice]:
|
|
"""E-Bauteile aus einer *_positionsdraw.json / *_positionsconv.json lesen.
|
|
|
|
Ein BMK kommt hoechstens einmal vor. Steht es doppelt da (im Schrank einmal
|
|
mit und einmal ohne Attribute), gewinnt der Eintrag mit VERW.
|
|
"""
|
|
warnings = warnings if warnings is not None else []
|
|
data = json.loads(path.read_text(encoding="utf-8"))
|
|
devices: dict[str, EDevice] = {}
|
|
for section, cabinet in (("sensors", False), ("schaltschrank_elemente", True)):
|
|
for key, raw in (data.get(section) or {}).items():
|
|
if not isinstance(raw, dict):
|
|
continue
|
|
dev = _device(key, raw, cabinet)
|
|
if not dev.bmk:
|
|
continue
|
|
old = devices.get(dev.bmk)
|
|
if old is None or (not old.verw and dev.verw):
|
|
devices[dev.bmk] = dev
|
|
for bmk, positions in (data.get("double_ids") or {}).items():
|
|
warnings.append(f"E-Planung: {bmk} steht {len(positions)}x in der Zeichnung "
|
|
f"- verwendet wird ein Eintrag")
|
|
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"))
|
|
out = []
|
|
for kind, entries in (data.get("warnings") or {}).items():
|
|
if isinstance(entries, dict):
|
|
for key, text in entries.items():
|
|
out.append(f"Positionsdatei ({kind}): {key.split('@', 1)[0]}: "
|
|
f"{str(text).split(':', 1)[0]}")
|
|
return out
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Zusammenfuehren
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
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()
|
|
m = _CONVEYOR_K.search(upper)
|
|
if m:
|
|
return f"K{m.group(1)}"
|
|
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 ""
|
|
|
|
|
|
def conveyor_kind_of(*texts: str) -> str:
|
|
"""Art des genannten Foerderers aus einem Klartext: "Kreisel", "Strecke" oder ""."""
|
|
for text in texts:
|
|
upper = (text or "").upper()
|
|
if _KIND_KREISEL.search(upper):
|
|
return "Kreisel"
|
|
if _KIND_STRECKE.search(upper):
|
|
return "Strecke"
|
|
return ""
|
|
|
|
|
|
def pair_drives(devices: list[EDevice]) -> list[Drive]:
|
|
"""Motor MAnnnn <-> Motorschutz FCnnnn (gleiche Nummer), Geber GF je Foerderer."""
|
|
fc = {d.number: d for d in devices if d.prefix.upper() == "FC" and d.number is not None}
|
|
encoders: dict[str, str] = {}
|
|
for d in sorted(devices, key=lambda d: d.bmk):
|
|
if d.prefix.upper() == "GF":
|
|
conv = getattr(d, "conveyor", "") or conveyor_of(d.bezeichnung)
|
|
if conv:
|
|
encoders.setdefault(conv, d.bmk)
|
|
drives = []
|
|
for d in sorted(devices, key=lambda d: d.bmk):
|
|
if d.prefix.upper() != "MA" or d.number is None:
|
|
continue
|
|
schutz = fc.get(d.number)
|
|
conv = (getattr(d, "conveyor", "") or conveyor_of(d.bezeichnung)
|
|
or (conveyor_of(schutz.bezeichnung) if schutz else ""))
|
|
drives.append(Drive(
|
|
motor=d.bmk,
|
|
protection=schutz.bmk if schutz else "",
|
|
bezeichnung=d.bezeichnung,
|
|
protection_bezeichnung=schutz.bezeichnung if schutz else "",
|
|
conveyor=conv,
|
|
conveyor_kind=conveyor_kind_of(d.bezeichnung,
|
|
schutz.bezeichnung if schutz else ""),
|
|
encoder=encoders.get(conv, "") if conv else "",
|
|
))
|
|
return drives
|
|
|
|
|
|
def _constant_name(bmk: str, direction: str) -> str:
|
|
return f"cIn{bmk}" if direction == "in" else f"cOut{bmk}" if direction == "out" else ""
|
|
|
|
|
|
def build_hardware(tia_path: Path | None, positions_path: Path | None,
|
|
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] = []
|
|
|
|
tags: list[TiaTag] = []
|
|
constants: list[TiaConstant] = []
|
|
if tia_path is not None:
|
|
tags, constants = load_tia(tia_path, warnings)
|
|
e_devices: list[EDevice] = []
|
|
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:
|
|
if t.name in tag_of:
|
|
checks.append(f"TIA: Tag {t.name} doppelt")
|
|
tag_of[t.name] = t
|
|
const_of = {c.name: c for c in constants}
|
|
e_of = {d.bmk: d for d in e_devices}
|
|
|
|
devices: list[HwDevice] = []
|
|
used_constants: set[str] = set()
|
|
for bmk in sorted(set(tag_of) | set(e_of)):
|
|
tag, e = tag_of.get(bmk), e_of.get(bmk)
|
|
direction = _direction(tag.address) if tag else ""
|
|
cname = _constant_name(bmk, direction)
|
|
const = const_of.get(cname)
|
|
if const:
|
|
used_constants.add(cname)
|
|
value = None
|
|
if const is not None:
|
|
try:
|
|
value = int(const.value)
|
|
except ValueError:
|
|
checks.append(f"TIA: Konstante {cname} hat keinen ganzzahligen Wert "
|
|
f"'{const.value}'")
|
|
base = e.model_dump() if e else {"bmk": bmk}
|
|
devices.append(HwDevice(
|
|
**base,
|
|
direction=direction,
|
|
address=tag.address if tag else "",
|
|
data_type=tag.data_type if tag else "",
|
|
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 "",
|
|
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))
|
|
if ok],
|
|
))
|
|
|
|
_check(devices, constants, used_constants, tags, e_devices, tia_path, positions_path,
|
|
checks)
|
|
return HardwareDocument(
|
|
project=project,
|
|
tia_file=tia_path.name if tia_path else "",
|
|
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],
|
|
checks=checks,
|
|
warnings=warnings,
|
|
)
|
|
|
|
|
|
def _check(devices: list[HwDevice], constants: list[TiaConstant], used_constants: set[str],
|
|
tags: list[TiaTag], e_devices: list[EDevice], tia_path: Path | None,
|
|
positions_path: Path | None, checks: list[str]) -> None:
|
|
"""Gegenpruefung TIA-Import <-> Positionsdatei (Ergebnis in checks)."""
|
|
if tia_path is not None and positions_path is not None:
|
|
only_e = [d.bmk for d in devices if d.sources == ["positions"]]
|
|
if only_e:
|
|
checks.append(f"nur in der Positionsdatei, nicht im TIA-Import "
|
|
f"({len(only_e)}): {', '.join(only_e)}")
|
|
# Feldgeraete (Sensoren/Ventile/Motoren) gehoeren in die Zeichnung;
|
|
# Taster, Leuchten, Not-Halt (SF/PF) stehen dort meist nicht.
|
|
only_tia = [d.bmk for d in devices if d.sources == ["tia"]
|
|
and d.prefix.upper() in ("BG", "MB", "MA", "BX")]
|
|
if only_tia:
|
|
checks.append(f"Feldgeraet nur im TIA-Import, nicht in der Positionsdatei "
|
|
f"({len(only_tia)}): {', '.join(only_tia)}")
|
|
differ = [d.bmk for d in devices if len(d.sources) == 2 and d.bezeichnung
|
|
and d.comment and d.bezeichnung.strip() != d.comment.strip()]
|
|
if differ:
|
|
checks.append(f"TIA-Kommentar weicht von der BEZEICHNUNG ab ({len(differ)}): "
|
|
f"{', '.join(differ)}")
|
|
if tia_path is not None:
|
|
no_const = [d.bmk for d in devices if d.in_tia and d.direction and not d.constant]
|
|
if no_const:
|
|
checks.append(f"Signal ohne Index-Konstante cIn/cOut ({len(no_const)}): "
|
|
f"{', '.join(no_const)}")
|
|
orphan = [c.name for c in constants
|
|
if c.name not in used_constants and re.match(r"^c(In|Out)[A-Z]+\d+$", c.name)]
|
|
if orphan:
|
|
checks.append(f"Index-Konstante ohne Signal ({len(orphan)}): {', '.join(orphan)}")
|
|
addresses = Counter(t.address for t in tags if t.address)
|
|
double = [a for a, n in addresses.items() if n > 1]
|
|
if double:
|
|
checks.append(f"Adresse mehrfach vergeben: {', '.join(sorted(double))}")
|
|
for direction in ("in", "out"):
|
|
values = Counter(d.constant_value for d in devices
|
|
if d.direction == direction and d.constant_value is not None)
|
|
double = sorted(v for v, n in values.items() if n > 1)
|
|
if double:
|
|
checks.append(f"Index-Konstanten ({direction}) mit gleichem Wert: "
|
|
f"{', '.join(map(str, double))}")
|
|
no_ctrl = [d.bmk for d in e_devices if not d.cabinet and not d.controller]
|
|
if no_ctrl:
|
|
checks.append(f"Feldgeraet ohne Steuerung in KENNZEICHNUNG ({len(no_ctrl)}): "
|
|
f"{', '.join(no_ctrl)}")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Bericht
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def render_report(doc: HardwareDocument) -> str:
|
|
out: list[str] = []
|
|
add = out.append
|
|
add(f"# Hardwareplanung: {doc.project}")
|
|
add("")
|
|
add(f"- TIA-Import: `{doc.tia_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")
|
|
add("")
|
|
prefixes = Counter(d.prefix or "?" for d in doc.devices)
|
|
add("| Praefix | Anzahl | im Feld mit Lage | mit Index-Konstante |")
|
|
add("|---|---|---|---|")
|
|
for p, n in sorted(prefixes.items()):
|
|
devs = [d for d in doc.devices if (d.prefix or "?") == p]
|
|
add(f"| {p} | {n} | {sum(1 for d in devs if d.pos and not d.cabinet)} "
|
|
f"| {sum(1 for d in devs if d.constant)} |")
|
|
add("")
|
|
add("## Antriebe (Motor <-> Motorschutz, Geber)")
|
|
add("")
|
|
if doc.drives:
|
|
add("| Foerderer | Motor | Bezeichnung | Motorschutz | Geber |")
|
|
add("|---|---|---|---|---|")
|
|
for dr in doc.drives:
|
|
add(f"| {dr.conveyor or '-'} | {dr.motor} | {dr.bezeichnung} "
|
|
f"| {dr.protection or '-'} | {dr.encoder or '-'} |")
|
|
else:
|
|
add("- keine Motoren")
|
|
add("")
|
|
conveyors = Counter(d.conveyor for d in doc.field_devices() if d.conveyor)
|
|
add("## Foerderer der E-Planung (Feldgeraete, die ihn im Klartext nennen)")
|
|
add("")
|
|
if conveyors:
|
|
for name, n in sorted(conveyors.items()):
|
|
add(f"- {name}: {n}")
|
|
else:
|
|
add("- keine")
|
|
add("")
|
|
add("## Gegenpruefung")
|
|
add("")
|
|
for c in doc.checks or ["keine Befunde"]:
|
|
add(f"- {c}")
|
|
add("")
|
|
if doc.constants:
|
|
add("## Konstanten ohne Signal")
|
|
add("")
|
|
add(", ".join(c.name for c in doc.constants))
|
|
add("")
|
|
if doc.warnings:
|
|
add("## Hinweise")
|
|
add("")
|
|
for w in doc.warnings:
|
|
add(f"- {w}")
|
|
add("")
|
|
return "\n".join(out)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# CLI
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_POSITION_SUFFIXES = ("_positionsconv", "_positionsdraw", "_positions")
|
|
|
|
|
|
def project_name(tia_path: Path | None, positions_path: Path | None) -> str:
|
|
"""Projektname aus den Dateinamen: Mubea_positionsconv.json -> Mubea."""
|
|
if positions_path is not None:
|
|
stem = positions_path.stem
|
|
for suffix in _POSITION_SUFFIXES:
|
|
if stem.lower().endswith(suffix):
|
|
return stem[: -len(suffix)]
|
|
return stem
|
|
stem = tia_path.stem if tia_path else "anlage"
|
|
return stem[:-4] if stem.lower().endswith("_tia") else stem
|
|
|
|
|
|
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
|
|
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 or dxf_path)
|
|
|
|
try:
|
|
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
|
|
|
|
results = env_dir("SKEL_RESULTS", "results")
|
|
results.mkdir(parents=True, exist_ok=True)
|
|
json_file = results / f"{project}_hw.json"
|
|
md_file = results / f"{project}_hw.md"
|
|
json_file.write_text(doc.to_json(), encoding="utf-8")
|
|
md_file.write_text(render_report(doc), encoding="utf-8")
|
|
|
|
n_tia = sum(1 for d in doc.devices if d.in_tia)
|
|
n_pos = sum(1 for d in doc.devices if "positions" in d.sources)
|
|
print("")
|
|
print("================================================================")
|
|
print("HARDWAREPLANUNG EINLESEN")
|
|
print("================================================================")
|
|
print(f"TIA-Import = {tia_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)})")
|
|
print(f"Feld mit Lage = {len(doc.field_devices())}")
|
|
print(f"Index-Konstanten = {sum(1 for d in doc.devices if d.constant)} "
|
|
f"(+ {len(doc.constants)} ohne Signal)")
|
|
print(f"Antriebe = {len(doc.drives)} "
|
|
f"({sum(1 for d in doc.drives if d.protection)} mit Motorschutz, "
|
|
f"{sum(1 for d in doc.drives if d.conveyor)} mit Foerderer)")
|
|
for c in doc.checks:
|
|
print(f" ? {c}")
|
|
for w in doc.warnings:
|
|
print(f" ! {w}")
|
|
print(f"JSON = {json_file}")
|
|
print(f"Bericht = {md_file}")
|
|
print("================================================================")
|
|
print("")
|
|
return 0
|
|
|
|
|
|
def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(
|
|
prog="hw_import",
|
|
description="Liest die Hardwareplanung ein - TIA-Import (PLC Tags + "
|
|
"Constants) und die Positionsdatei des Kabellaengen-Tools - "
|
|
"und schreibt <projekt>_hw.json fuer merge_hw_layout plus einen "
|
|
"Bericht <projekt>_hw.md.",
|
|
formatter_class=argparse.RawDescriptionHelpFormatter,
|
|
epilog="Beispiele:\n"
|
|
" 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"
|
|
" 1 Eingabe- oder Lesefehler\n",
|
|
)
|
|
parser.add_argument("--tia", required=True, metavar="NAME",
|
|
help="TIA-Import *_TIA.xlsx in %%SKEL_DATA%% (oder ein Pfad, "
|
|
"auch mit Unterordner)")
|
|
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)")
|
|
return parser.parse_args(argv)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|