# -*- 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 -> _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``/``cOut`` 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"(? K0055 / K0046 / K0043 _CONVEYOR_MA = re.compile(r"(? Kreisel; # "CONVEYOR #106", "KLINKENFOERDERER", "ILS-VARIO" -> Strecke _KIND_KREISEL = re.compile(r"(? 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 ``_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", 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 _hw.json fuer merge_hw_layout plus einen " "Bericht _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 _hw.json. Standard: aus dem " "Namen der Positionsdatei (Mubea_positionsconv.json -> Mubea)") return parser.parse_args(argv) if __name__ == "__main__": sys.exit(main())