mistangl/sps_skel
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

hw_import + merge_hw_layout: Hardwareplanung einlesen und mit dem Layout zusammenfuehren

Browse Source 
Stufen 2 und 3 aus doc/Konzept_Layout_Hardware_SCL.md:

- lib/hw_import.py: TIA-Import (*_TIA.xlsx, ohne openpyxl gelesen) +
  Positionsdatei (*_positionsconv.json) -> <projekt>_hw.json: je BMK
  Adresse, Index-Konstante, VERW, Station, Lage, Foerderer (K0102, #106 ->
  K0106); Antriebe MA/FC/GF; Gegenpruefung beider Quellen.
- lib/merge_hw_layout.py: <csv>_tro.json + <projekt>_hw.json ->
  <csv>_anlage.json (model.AnlageDocument): bindings je TRO-ID + Slot +
  Rolle mit Konstante und Adresse, Foerderer per Mehrheit (Mubea: K0102 =
  Kreisel3, K0104 = Kreisel2, K0105 = Kreisel1), Handkorrekturen
  cfg/hw_overrides.ini. Zuordnungslogik aus e_planning.py hierher
  verschoben; e_planning re-exportiert sie, Ergebnis unveraendert.
- TRO-JSON traegt optional `parts` (Lage/Huellbox, tro_flow --json),
  PinStore bekommt Slots line<n>.
- Gespeicherter Lagebezug mit SHA-256 der Positionsdatei; Unterordner in
  Eingabenamen relativ zu SKEL_DATA.
- Doku: Python_Scripts.md 5d/5e, Konzept, E-Planung_Bruecke (Abgleich
  mubea_layout.dxf / CSV / E-Planung).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Samer Ayadi
2026-10-15 16:16:10 +02:00
co-authored by Claude Opus 5.5
parent c4122b4497
commit df24cf5adb
21 changed files with 3226 additions and 442 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/dxf_registration.py
+27 -2
View File
@@ -50,6 +50,7 @@ neue Ermittlung.
from __future__ import annotations
import hashlib
import json
import math
from collections import Counter
@@ -69,6 +70,7 @@ __all__ = [
"load_transform",
"save_transform",
"registration_path",
"file_fingerprint",
]
# Rollen fuer den Lagebezug: CSV-Objektart -> Blocknamen im DXF, als Muster
@@ -123,6 +125,10 @@ class Transform(BaseModel):
csv_file: str = Field(default="")
dxf_file: str = Field(default="")
created: str = Field(default="")
# Pruefsumme der Zeichnung/E-Datei beim Ermitteln ("" = unbekannt). Der
# Schluessel enthaelt nur Dateinamen; ein neuer Stand gleichen Namens wird
# daran erkannt (siehe load_transform).
fingerprint: str = Field(default="")
# -- Anwenden ----------------------------------------------------------
@@ -329,8 +335,25 @@ def _key(csv_file: str, dxf_file: str) -> str:
return f"{Path(csv_file).name}|{Path(dxf_file).name}"
def load_transform(cfg_dir: Path, csv_file: str, dxf_file: str) -> Transform | None:
"""Geprueften Lagebezug aus der Konfiguration lesen."""
def file_fingerprint(path: Path) -> str:
"""SHA-256 einer Datei (Hex) - Kennung des Planungsstands."""
digest = hashlib.sha256()
with path.open("rb") as handle:
for chunk in iter(lambda: handle.read(1 << 20), b""):
digest.update(chunk)
return digest.hexdigest()
def load_transform(cfg_dir: Path, csv_file: str, dxf_file: str,
fingerprint: str = "") -> Transform | None:
"""Geprueften Lagebezug aus der Konfiguration lesen.
Mit ``fingerprint``: nur verwenden, wenn der abgelegte Bezug fuer genau
diesen Dateistand ermittelt wurde. Ein Eintrag ohne Pruefsumme (aeltere
Ablage) oder mit anderer gilt als veraltet -> None, der Aufrufer ermittelt
neu. Ohne ``fingerprint`` (z. B. tro_annotate) wie bisher nur ueber den
Dateinamen.
"""
path = registration_path(cfg_dir)
if not path.is_file():
return None
@@ -341,6 +364,8 @@ def load_transform(cfg_dir: Path, csv_file: str, dxf_file: str) -> Transform | N
entry = data.get(_key(csv_file, dxf_file))
if not entry:
return None
if fingerprint and entry.get("fingerprint", "") != fingerprint:
return None
transform = Transform.model_validate(entry)
transform.source = "cfg"
return transform
lib/e_planning.py
+27 -386
View File
@@ -31,8 +31,15 @@ scl_skeleton.py --eplan liest die JSON und setzt die gefundenen Namen in die
SCL ein (SclName mit todo=False). Was keine Regel trifft, bleibt dort
TODO(E-Planung) - geraten wird nicht.
Schichten: e_planning importiert material_flow (CSV) und dxf_registration
(Lagebezug); scl_context importiert nur das Modell EplanDocument.
Seit 2026-10-15 liegen das E-Modell (EDevice, load_positionsdraw,
pair_drives) in hw_import.py und die Zuordnung (Regeln, raeumlich,
Nummernregel, Lagebezug) in merge_hw_layout.py - der neue Weg ueber die
Anlagen-JSON (doc/Konzept_Layout_Hardware_SCL.md). e_planning bleibt fuer
scl_skeleton --eplan, bis scl_skeleton die Anlagen-JSON liest, und
re-exportiert die verschobenen Namen.
Schichten: e_planning importiert material_flow (CSV), hw_import und
merge_hw_layout; scl_context importiert nur das Modell EplanDocument.
Aufruf ueber bin/e_planning.bat bzw. bin/e_planning.sh.
"""
@@ -40,27 +47,25 @@ Aufruf ueber bin/e_planning.bat bzw. bin/e_planning.sh.
from __future__ import annotations
import argparse
import configparser
import fnmatch
import json
import math
import re
import sys
from dataclasses import dataclass, field
from datetime import datetime
from pathlib import Path
from pydantic import BaseModel, ConfigDict, Field
from dxf_registration import (
RegistrationError,
Transform,
derive_transform,
load_transform,
save_transform,
verify_transform,
from dxf_registration import RegistrationError, Transform, file_fingerprint
from merge_hw_layout import (
LAYOUT_KINDS,
RULES_FILE,
Assignment,
LayoutPoint,
Rule,
_match,
assign,
load_rules,
resolve_transform,
)
from parameter import get_float
from hw_import import Drive, EDevice, load_positionsdraw, pair_drives
__all__ = [
"EPLAN_SCHEMA",
@@ -74,21 +79,12 @@ __all__ = [
"load_rules",
"assign",
"pair_drives",
"layout_points",
"resolve_transform",
"LAYOUT_KINDS",
]
EPLAN_SCHEMA = "sps_skel-eplan-v1"
RULES_FILE = "e_planning.ini"
# Zulaessige Abweichung vom Sollabstand, wenn eine Regel kein toleranz_mm
# angibt - aus cfg/parameter.ini [e_planning] (lib/parameter.py).
DEFAULT_TOLERANCE_MM = get_float("e_planning", "toleranz_mm")
_CONFIG = ConfigDict(extra="forbid")
# "=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+)$")
# ---------------------------------------------------------------------------
@@ -96,112 +92,6 @@ _BMK = re.compile(r"^([A-Za-z]+)(\d+)$")
# ---------------------------------------------------------------------------
class EDevice(BaseModel):
"""Ein Sensor/Aktor/Schrankelement der E-Planung."""
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
@dataclass
class LayoutPoint:
"""Ein punktartiges Bauteil des Layouts in CSV-Koordinaten."""
layout_id: str
kind: str
x: float
y: float
name: str = ""
@dataclass
class Rule:
"""Eine Zuordnungsregel aus e_planning.ini (Abschnitt "<Art>.<Rolle>")."""
role: str # "Separator.sensor"
layout: list[str] # Muster fuer die Layout-Art
verw: list[str] # Muster fuer VERW
max_mm: float | None = None # raeumlich: Hoechstabstand
number_from: str = "" # Nummernregel: Rolle des Partners ...
number_offset: int = 0 # ... und Versatz der Nummer
prefix: str = "" # BMK-Praefix, leer = egal
# Sollabstand: das Symbol liegt immer gleich weit neben dem Bauteil.
# Dann zaehlt die Abweichung vom Soll, nicht der kleinste Abstand.
target_mm: float | None = None
tolerance_mm: float = DEFAULT_TOLERANCE_MM
def score(self, dist: float) -> float | None:
"""Rangfolge eines Kandidaten (kleiner = besser), None = nicht zulaessig."""
if self.max_mm is not None and dist > self.max_mm:
return None
if self.target_mm is None:
return dist
dev = abs(dist - self.target_mm)
return dev if dev <= self.tolerance_mm else None
class Assignment(BaseModel):
"""Ein E-Bauteil in einer Rolle an einem Layout-Bauteil."""
model_config = _CONFIG
layout_id: str
layout_kind: str
role: str
bmk: str
verw: str = ""
bezeichnung: str = ""
controller: str = ""
station: str = ""
# Abstand nach dem Lagebezug in mm (auch bei der Nummernregel zur Kontrolle)
distance: float
# "raeumlich" | "nummer"
rule: str
# Abstand des naechstbesten Kandidaten derselben Rolle (None: keiner)
runner_up: float | None = None
class Drive(BaseModel):
"""Motor im Feld und sein Motorschutz im Schrank (gleiche Nummer)."""
model_config = _CONFIG
motor: str
protection: str = ""
bezeichnung: str = ""
protection_bezeichnung: str = ""
class EplanDocument(BaseModel):
"""Die Zuordnungs-JSON <csv>_eplan.json."""
@@ -234,235 +124,10 @@ class EplanDocument(BaseModel):
# ---------------------------------------------------------------------------
# Einlesen
# Layout
# ---------------------------------------------------------------------------
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 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 _patterns(text: str) -> list[str]:
return [p.strip() for p in text.split(",") if p.strip()]
def load_rules(path: Path) -> tuple[list[Rule], list[str]]:
"""Regeln und Anker-Verwendungen aus e_planning.ini lesen.
Abschnitt [Registrierung] -> anker (VERW-Muster fuer den Lagebezug);
jeder Abschnitt "<Art>.<Rolle>" ist eine Regel, in Dateireihenfolge.
"""
parser = configparser.ConfigParser(inline_comment_prefixes=(";", "#"),
interpolation=None)
parser.optionxform = str # Schluessel nicht kleinschreiben
with path.open(encoding="utf-8") as fh:
parser.read_file(fh)
anchors = _patterns(parser.get("Registrierung", "anker", fallback="IN SEP"))
rules = []
for section in parser.sections():
if "." not in section:
continue
s = parser[section]
max_mm = s.get("max_mm")
soll = s.get("soll_mm")
rules.append(Rule(
role=section,
layout=_patterns(s.get("layout", section.split(".", 1)[0])),
verw=_patterns(s.get("verw", "*")),
max_mm=float(max_mm) if max_mm else None,
number_from=s.get("nummer_von", "").strip(),
number_offset=int(s.get("nummer_versatz", "0")),
prefix=s.get("praefix", "").strip(),
target_mm=float(soll) if soll else None,
tolerance_mm=float(s.get("toleranz_mm", str(DEFAULT_TOLERANCE_MM))),
))
known = {r.role for r in rules}
for r in rules:
if r.number_from and r.number_from not in known:
raise ValueError(f"{path.name} [{r.role}]: nummer_von = {r.number_from} "
f"ist keine Regel")
return rules, anchors
def _match(text: str, patterns: list[str]) -> bool:
return any(fnmatch.fnmatchcase(text.upper(), p.upper()) for p in patterns)
# ---------------------------------------------------------------------------
# Zuordnung
# ---------------------------------------------------------------------------
@dataclass
class _State:
positions: dict[str, tuple[float, float]]
used: set[str] = field(default_factory=set)
result: list[Assignment] = field(default_factory=list)
# (layout_id, role) -> Assignment
by_slot: dict[tuple[str, str], Assignment] = field(default_factory=dict)
def _make(point: LayoutPoint, rule: Rule, dev: EDevice, dist: float, how: str,
runner_up: float | None) -> Assignment:
return Assignment(
layout_id=point.layout_id, layout_kind=point.kind, role=rule.role,
bmk=dev.bmk, verw=dev.verw, bezeichnung=dev.bezeichnung,
controller=dev.controller, station=dev.station,
distance=round(dist, 1), rule=how,
runner_up=None if runner_up is None else round(runner_up, 1),
)
def _candidates(rule: Rule, devices: list[EDevice], state: _State) -> list[EDevice]:
return [
d for d in devices
if d.bmk not in state.used and d.bmk in state.positions
and _match(d.verw, rule.verw)
and (not rule.prefix or d.prefix.upper() == rule.prefix.upper())
]
def _assign_spatial(rule: Rule, points: list[LayoutPoint], devices: list[EDevice],
state: _State) -> None:
"""Beste Bewertung zuerst (Rule.score); je Bauteil und E-Bauteil hoechstens einmal."""
cands = _candidates(rule, devices, state)
pairs = []
for p in points:
if (p.layout_id, rule.role) in state.by_slot:
continue
for d in cands:
dist = math.dist((p.x, p.y), state.positions[d.bmk])
score = rule.score(dist)
if score is not None:
pairs.append((score, dist, p.layout_id, d.bmk))
pairs.sort()
# je Bauteil die Kandidaten in Rangfolge - fuer "naechstbester Kandidat"
by_point: dict[str, list[tuple[str, float]]] = {}
for _, dist, pid, bmk in pairs:
by_point.setdefault(pid, []).append((bmk, dist))
point_of = {p.layout_id: p for p in points}
dev_of = {d.bmk: d for d in cands}
for _, dist, pid, bmk in pairs:
if (pid, rule.role) in state.by_slot or bmk in state.used:
continue
others = [x for b, x in by_point[pid] if b != bmk]
a = _make(point_of[pid], rule, dev_of[bmk], dist, "raeumlich",
others[0] if others else None)
state.used.add(bmk)
state.by_slot[(pid, rule.role)] = a
state.result.append(a)
def _assign_by_number(rule: Rule, points: list[LayoutPoint], devices: list[EDevice],
state: _State) -> None:
"""Nummer = Nummer des Partners (Rolle nummer_von) + Versatz."""
cands = {(d.number): d for d in _candidates(rule, devices, state)
if d.number is not None}
for p in points:
if (p.layout_id, rule.role) in state.by_slot:
continue
partner = state.by_slot.get((p.layout_id, rule.number_from))
if partner is None:
continue
m = _BMK.match(partner.bmk)
if not m:
continue
dev = cands.get(int(m.group(2)) + rule.number_offset)
if dev is None or dev.bmk in state.used:
continue
dist = math.dist((p.x, p.y), state.positions[dev.bmk])
# Auch die Nummernregel darf nicht quer ueber die Anlage greifen
if rule.max_mm is not None and dist > rule.max_mm:
continue
a = _make(p, rule, dev, dist, "nummer", None)
state.used.add(dev.bmk)
state.by_slot[(p.layout_id, rule.role)] = a
state.result.append(a)
def assign(points: list[LayoutPoint], devices: list[EDevice], rules: list[Rule],
transform: Transform) -> list[Assignment]:
"""Regeln der Reihe nach anwenden.
Eine Regel mit nummer_von versucht zuerst die Nummernregel und nimmt fuer
die verbleibenden Bauteile den Abstand (wenn max_mm/soll_mm gesetzt ist). So wird
das Ventil eines Separators ueber die Nummer gefunden, auch wenn das
Ventil des Nachbarn naeher liegt (Speicherlinien im 800-mm-Raster).
"""
state = _State(positions={
d.bmk: transform.to_csv(*d.pos) for d in devices
if d.pos is not None and not d.cabinet
})
for rule in rules:
targets = [p for p in points if _match(p.kind, rule.layout)]
if rule.number_from:
_assign_by_number(rule, targets, devices, state)
if rule.max_mm is not None or rule.target_mm is not None:
_assign_spatial(rule, targets, devices, state)
return sorted(state.result, key=lambda a: (a.layout_id, a.role))
def pair_drives(devices: list[EDevice]) -> list[Drive]:
"""Motor MAnnnn <-> Motorschutz FCnnnn (gleiche Nummer)."""
fc = {d.number: d for d in devices if d.prefix.upper() == "FC" and d.number is not None}
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)
drives.append(Drive(
motor=d.bmk,
protection=schutz.bmk if schutz else "",
bezeichnung=d.bezeichnung,
protection_bezeichnung=schutz.bezeichnung if schutz else "",
))
return drives
# ---------------------------------------------------------------------------
# Layout und Lagebezug
# ---------------------------------------------------------------------------
# Element.kind der punktartigen Layout-Bauteile
LAYOUT_KINDS = ("Separator", "Scanner", "Weiche", "Ausschleus", "Einschleus",
"Beladung", "Entladung")
def layout_points(elements) -> list[LayoutPoint]:
"""Punktartige Bauteile aus den CSV-Elementen (material_flow.Element)."""
points = []
@@ -476,31 +141,6 @@ def layout_points(elements) -> list[LayoutPoint]:
return points
def _anchor_points(points: list[LayoutPoint], devices: list[EDevice],
anchors: list[str]):
csv_pts = {"Separator": [(p.x, p.y) for p in points if p.kind == "Separator"]}
e_pts = {"Separator": [d.pos for d in devices
if d.pos is not None and not d.cabinet
and _match(d.verw, anchors)]}
return csv_pts, e_pts
def resolve_transform(points, devices, anchors, cfg_dir: Path, csv_name: str,
eplan_name: str, re_register: bool,
warnings: list[str]) -> Transform:
"""Abgelegten Lagebezug nachrechnen oder neu ermitteln und ablegen."""
csv_pts, e_pts = _anchor_points(points, devices, anchors)
if not re_register:
stored = load_transform(cfg_dir, csv_name, eplan_name)
if stored:
verify_transform(stored, csv_pts, e_pts)
return stored
transform = derive_transform(csv_pts, e_pts, csv_name, eplan_name)
saved = save_transform(cfg_dir, transform)
warnings.append(f"Lagebezug neu ermittelt und abgelegt: {saved}")
return transform
# ---------------------------------------------------------------------------
# Bericht
# ---------------------------------------------------------------------------
@@ -610,7 +250,8 @@ def build_document(csv_path: Path, eplan_path: Path, rules_path: Path, cfg_dir:
devices = load_positionsdraw(eplan_path, warnings)
rules, anchors = load_rules(rules_path)
transform = resolve_transform(points, devices, anchors, cfg_dir, csv_path.name,
eplan_path.name, re_register, warnings)
eplan_path.name, re_register, warnings,
file_fingerprint(eplan_path))
assignments = assign(points, devices, rules, transform)
used = {a.bmk for a in assignments}
unassigned = sorted(d.bmk for d in devices
lib/hw_import.py
+784
View File
@@ -0,0 +1,784 @@
# -*- 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.
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.
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",
"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)")
# Art des Foerderers im Klartext: "KREISEL-K0102", "KR-K0104" -> Kreisel;
# "CONVEYOR #106", "KLINKENFOERDERER" -> Strecke
_KIND_KREISEL = re.compile(r"(?<![A-Z])(KREISEL|KR)(?![A-Z])")
_KIND_STRECKE = re.compile(r"CONVEYOR|CONVEYER|F(OE|Ö)RDERER")
# ---------------------------------------------------------------------------
# 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")
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 _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) -> 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".
"""
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}"
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 = "") -> 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")
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))
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 ""),
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 "",
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 '-'}`")
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:
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)):
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)
try:
doc = build_hardware(tia_path, positions_path, project)
except (OSError, ValueError, KeyError, zipfile.BadZipFile, ET.ParseError) 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 '-'}")
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"
"\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("--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())
lib/material_flow.py
+10 -2
View File
@@ -463,10 +463,18 @@ def env_dir(var: str, fallback: str) -> Path:
def resolve_input(name: str) -> Path:
"""Dateinamen gegen SKEL_DATA aufloesen; Pfadangaben bleiben unveraendert."""
"""Dateinamen gegen SKEL_DATA aufloesen.
Ein blosser Name liegt in SKEL_DATA, ein absoluter Pfad bleibt. Ein
relativer Pfad mit Ordner ("Mubea/Mubea_positionsconv.json") gilt ab dem
aktuellen Verzeichnis, wenn er dort existiert, sonst ab SKEL_DATA.
"""
candidate = Path(name)
if candidate.is_absolute() or candidate.parent != Path("."):
if candidate.is_absolute():
return candidate
if candidate.parent != Path("."):
if candidate.exists():
return candidate
return env_dir("SKEL_DATA", "data") / name
lib/merge_hw_layout.py
+969
View File
@@ -0,0 +1,969 @@
# -*- coding: utf-8 -*-
# Absichtlich ohne Shebang: der Windows-py-Launcher wuerde daraus "python3"
# ableiten und scheitert, wenn keine PythonCore-Installation registriert ist.
"""
merge_hw_layout.py - Layout und Hardwareplanung zusammenfuehren -> <csv>_anlage.json.
Stufe 3 des Ablaufs in doc/Konzept_Layout_Hardware_SCL.md:
<csv>_tro.json (tro_flow --json) + <projekt>_hw.json (hw_import)
\\ /
merge_hw_layout -> <csv>_anlage.json + Bericht .md
Das CSV wird nicht gelesen: die TRO-JSON traegt die Slots der TROs und die
Lage der Layout-Bauteile (``parts``). Ablauf:
1. Lagebezug E-Zeichnung -> Layout (lib/dxf_registration.py) ueber die
Separator-Sensoren (VERW "IN SEP") gegen die Separatoren des Layouts;
abgelegt in SKEL_CFG/dxf_registration.json mit Pruefsumme der
Positionsdatei - ein neuer Stand gleichen Namens wird neu ermittelt.
2. Regeln aus SKEL_CFG/e_planning.ini (raeumlich mit Hoechst-/Sollabstand,
Nummernregel MB = BG + 1) ordnen E-Bauteile Layout-Bauteilen zu.
3. Jede Zuordnung geht an die Slot-Rolle des TRO, der das Bauteil traegt
(Schluessel TRO-ID + Slot + Rolle). Bauteile ohne Slot (Ausschleus-
elemente, Weichen) bleiben als ``ohne_slot`` sichtbar.
4. Foerderer: die Namen der E-Planung (K0102) und des Layouts (Kreisel3)
passen nicht zusammen. Alle Feldgeraete, deren Klartext einen Foerderer
nennt, stimmen fuer den naechsten Layout-Foerderer ab; die Mehrheit
entscheidet (cfg/parameter.ini [merge_hw_layout]). Daraus Antrieb, Motorschutz
und Geber je Layout-Foerderer und die Antriebs-Slots der Vario-TROs.
5. Handkorrekturen aus SKEL_CFG/hw_overrides.ini, zuletzt.
6. Pflichtrollen ohne BMK werden ``offen`` (spaeter TODO(E-Planung)), ein
BMK, das nicht im TIA-Import steht, ``widerspruch``.
Lieber eine Luecke als ein falscher Name: was keine Regel sicher trifft,
bleibt offen und steht im Bericht.
Die Zuordnungslogik (Regeln, raeumlich, Nummernregel) stand bis 2026-10-15
in e_planning.py; e_planning importiert sie jetzt von hier, damit
``scl_skeleton --eplan`` bis zur Umstellung auf die Anlagen-JSON weiterlaeuft.
Aufruf ueber bin/merge_hw_layout.bat bzw. bin/merge_hw_layout.sh.
"""
from __future__ import annotations
import argparse
import configparser
import fnmatch
import math
import sys
from collections import Counter
from dataclasses import dataclass, field
from pathlib import Path
from pydantic import BaseModel, ConfigDict, ValidationError
from dxf_registration import (
RegistrationError,
Transform,
derive_transform,
load_transform,
save_transform,
verify_transform,
)
from hw_import import HW_SCHEMA, EDevice, HardwareDocument, HwDevice
from model import (
AnlageDocument,
Binding,
ConveyorBinding,
LayoutPart,
Registration,
TroDocument,
)
from parameter import get_float, get_int
__all__ = [
"RULES_FILE",
"OVERRIDES_FILE",
"LAYOUT_KINDS",
"CONVEYOR_KINDS",
"LayoutPoint",
"Rule",
"Assignment",
"load_rules",
"assign",
"resolve_transform",
"points_from_parts",
"bind_slots",
"assign_conveyors",
"load_overrides",
"build_anlage",
]
RULES_FILE = "e_planning.ini"
OVERRIDES_FILE = "hw_overrides.ini"
# Zahlenwerte aus cfg/parameter.ini [merge_hw_layout] - Bedeutung und Einheit stehen dort.
_P = "merge_hw_layout"
DEFAULT_TOLERANCE_MM = get_float(_P, "toleranz_mm")
CONVEYOR_MAX_MM = get_float(_P, "conveyor_max_mm")
CONVEYOR_MIN_VOTES = get_int(_P, "conveyor_min_votes")
CONVEYOR_MIN_SHARE = get_float(_P, "conveyor_min_share")
REGISTRATION_MIN_SHARE = get_float(_P, "registration_min_share")
_CONFIG = ConfigDict(extra="forbid")
# Punktartige Layout-Bauteile (Element.kind bzw. Teil der TeileArt)
LAYOUT_KINDS = ("Separator", "Scanner", "Weiche", "Ausschleus", "Einschleus",
"Beladung", "Entladung")
# Foerderer, die fuer einen Foerderer der E-Planung in Frage kommen
CONVEYOR_KINDS = ("Kreisel", "Strecke", "Gefaellestrecke")
# Antriebs-Rollen eines Slots, die aus dem zugeordneten Foerderer folgen
_DRIVE_ROLES = {"motor": "motor", "motorProtection": "protection"}
# ---------------------------------------------------------------------------
# Regeln und Zuordnung (bis 2026-10-15 in e_planning.py)
# ---------------------------------------------------------------------------
@dataclass
class LayoutPoint:
"""Ein punktartiges Bauteil des Layouts in Layout-Koordinaten."""
layout_id: str
kind: str
x: float
y: float
name: str = ""
@dataclass
class Rule:
"""Eine Zuordnungsregel aus e_planning.ini (Abschnitt "<Art>.<Rolle>")."""
role: str # "Separator.sensor"
layout: list[str] # Muster fuer die Layout-Art
verw: list[str] # Muster fuer VERW
max_mm: float | None = None # raeumlich: Hoechstabstand
number_from: str = "" # Nummernregel: Rolle des Partners ...
number_offset: int = 0 # ... und Versatz der Nummer
prefix: str = "" # BMK-Praefix, leer = egal
# Sollabstand: das Symbol liegt immer gleich weit neben dem Bauteil.
# Dann zaehlt die Abweichung vom Soll, nicht der kleinste Abstand.
target_mm: float | None = None
tolerance_mm: float = DEFAULT_TOLERANCE_MM
@property
def slot_role(self) -> str:
"""Rolle im Slot: "Separator.sensor" -> "sensor"."""
return self.role.split(".", 1)[-1]
def score(self, dist: float) -> float | None:
"""Rangfolge eines Kandidaten (kleiner = besser), None = nicht zulaessig."""
if self.max_mm is not None and dist > self.max_mm:
return None
if self.target_mm is None:
return dist
dev = abs(dist - self.target_mm)
return dev if dev <= self.tolerance_mm else None
class Assignment(BaseModel):
"""Ein E-Bauteil in einer Rolle an einem Layout-Bauteil."""
model_config = _CONFIG
layout_id: str
layout_kind: str
role: str
bmk: str
verw: str = ""
bezeichnung: str = ""
controller: str = ""
station: str = ""
# Abstand nach dem Lagebezug in mm (auch bei der Nummernregel zur Kontrolle)
distance: float
# "raeumlich" | "nummer"
rule: str
# Abstand des naechstbesten Kandidaten derselben Rolle (None: keiner)
runner_up: float | None = None
def _patterns(text: str) -> list[str]:
return [p.strip() for p in text.split(",") if p.strip()]
def _match(text: str, patterns: list[str]) -> bool:
return any(fnmatch.fnmatchcase(text.upper(), p.upper()) for p in patterns)
def load_rules(path: Path) -> tuple[list[Rule], list[str]]:
"""Regeln und Anker-Verwendungen aus e_planning.ini lesen.
Abschnitt [Registrierung] -> anker (VERW-Muster fuer den Lagebezug);
jeder Abschnitt "<Art>.<Rolle>" ist eine Regel, in Dateireihenfolge.
"""
parser = configparser.ConfigParser(inline_comment_prefixes=(";", "#"),
interpolation=None)
parser.optionxform = str # Schluessel nicht kleinschreiben
with path.open(encoding="utf-8") as fh:
parser.read_file(fh)
anchors = _patterns(parser.get("Registrierung", "anker", fallback="IN SEP"))
rules = []
for section in parser.sections():
if "." not in section:
continue
s = parser[section]
max_mm = s.get("max_mm")
soll = s.get("soll_mm")
rules.append(Rule(
role=section,
layout=_patterns(s.get("layout", section.split(".", 1)[0])),
verw=_patterns(s.get("verw", "*")),
max_mm=float(max_mm) if max_mm else None,
number_from=s.get("nummer_von", "").strip(),
number_offset=int(s.get("nummer_versatz", "0")),
prefix=s.get("praefix", "").strip(),
target_mm=float(soll) if soll else None,
tolerance_mm=float(s.get("toleranz_mm", str(DEFAULT_TOLERANCE_MM))),
))
known = {r.role for r in rules}
for r in rules:
if r.number_from and r.number_from not in known:
raise ValueError(f"{path.name} [{r.role}]: nummer_von = {r.number_from} "
f"ist keine Regel")
return rules, anchors
@dataclass
class _State:
positions: dict[str, tuple[float, float]]
used: set[str] = field(default_factory=set)
result: list[Assignment] = field(default_factory=list)
# (layout_id, role) -> Assignment
by_slot: dict[tuple[str, str], Assignment] = field(default_factory=dict)
def _make(point: LayoutPoint, rule: Rule, dev: EDevice, dist: float, how: str,
runner_up: float | None) -> Assignment:
return Assignment(
layout_id=point.layout_id, layout_kind=point.kind, role=rule.role,
bmk=dev.bmk, verw=dev.verw, bezeichnung=dev.bezeichnung,
controller=dev.controller, station=dev.station,
distance=round(dist, 1), rule=how,
runner_up=None if runner_up is None else round(runner_up, 1),
)
def _candidates(rule: Rule, devices: list[EDevice], state: _State) -> list[EDevice]:
return [
d for d in devices
if d.bmk not in state.used and d.bmk in state.positions
and _match(d.verw, rule.verw)
and (not rule.prefix or d.prefix.upper() == rule.prefix.upper())
]
def _assign_spatial(rule: Rule, points: list[LayoutPoint], devices: list[EDevice],
state: _State) -> None:
"""Beste Bewertung zuerst (Rule.score); je Bauteil und E-Bauteil hoechstens einmal."""
cands = _candidates(rule, devices, state)
pairs = []
for p in points:
if (p.layout_id, rule.role) in state.by_slot:
continue
for d in cands:
dist = math.dist((p.x, p.y), state.positions[d.bmk])
score = rule.score(dist)
if score is not None:
pairs.append((score, dist, p.layout_id, d.bmk))
pairs.sort()
# je Bauteil die Kandidaten in Rangfolge - fuer "naechstbester Kandidat"
by_point: dict[str, list[tuple[str, float]]] = {}
for _, dist, pid, bmk in pairs:
by_point.setdefault(pid, []).append((bmk, dist))
point_of = {p.layout_id: p for p in points}
dev_of = {d.bmk: d for d in cands}
for _, dist, pid, bmk in pairs:
if (pid, rule.role) in state.by_slot or bmk in state.used:
continue
others = [x for b, x in by_point[pid] if b != bmk]
a = _make(point_of[pid], rule, dev_of[bmk], dist, "raeumlich",
others[0] if others else None)
state.used.add(bmk)
state.by_slot[(pid, rule.role)] = a
state.result.append(a)
def _assign_by_number(rule: Rule, points: list[LayoutPoint], devices: list[EDevice],
state: _State) -> None:
"""Nummer = Nummer des Partners (Rolle nummer_von) + Versatz."""
cands = {(d.number): d for d in _candidates(rule, devices, state)
if d.number is not None}
for p in points:
if (p.layout_id, rule.role) in state.by_slot:
continue
partner = state.by_slot.get((p.layout_id, rule.number_from))
if partner is None or partner.bmk == "":
continue
number = next((d.number for d in devices if d.bmk == partner.bmk), None)
if number is None:
continue
dev = cands.get(number + rule.number_offset)
if dev is None or dev.bmk in state.used:
continue
dist = math.dist((p.x, p.y), state.positions[dev.bmk])
# Auch die Nummernregel darf nicht quer ueber die Anlage greifen
if rule.max_mm is not None and dist > rule.max_mm:
continue
a = _make(p, rule, dev, dist, "nummer", None)
state.used.add(dev.bmk)
state.by_slot[(p.layout_id, rule.role)] = a
state.result.append(a)
def assign(points: list[LayoutPoint], devices: list[EDevice], rules: list[Rule],
transform: Transform) -> list[Assignment]:
"""Regeln der Reihe nach anwenden.
Eine Regel mit nummer_von versucht zuerst die Nummernregel und nimmt fuer
die verbleibenden Bauteile den Abstand (wenn max_mm/soll_mm gesetzt ist). So wird
das Ventil eines Separators ueber die Nummer gefunden, auch wenn das
Ventil des Nachbarn naeher liegt (Speicherlinien im 800-mm-Raster).
"""
state = _State(positions={
d.bmk: transform.to_csv(*d.pos) for d in devices
if d.pos is not None and not d.cabinet
})
for rule in rules:
targets = [p for p in points if _match(p.kind, rule.layout)]
if rule.number_from:
_assign_by_number(rule, targets, devices, state)
if rule.max_mm is not None or rule.target_mm is not None:
_assign_spatial(rule, targets, devices, state)
return sorted(state.result, key=lambda a: (a.layout_id, a.role))
# ---------------------------------------------------------------------------
# Lagebezug
# ---------------------------------------------------------------------------
def _anchor_points(points: list[LayoutPoint], devices: list[EDevice],
anchors: list[str]):
layout_pts = {"Separator": [(p.x, p.y) for p in points if p.kind == "Separator"]}
e_pts = {"Separator": [d.pos for d in devices
if d.pos is not None and not d.cabinet
and _match(d.verw, anchors)]}
return layout_pts, e_pts
def resolve_transform(points, devices, anchors, cfg_dir: Path, csv_name: str,
eplan_name: str, re_register: bool,
warnings: list[str], fingerprint: str = "") -> Transform:
"""Abgelegten Lagebezug nachrechnen oder neu ermitteln und ablegen.
Mit ``fingerprint`` (Pruefsumme der Positionsdatei) gilt ein abgelegter
Bezug nur fuer genau diesen Dateistand.
"""
layout_pts, e_pts = _anchor_points(points, devices, anchors)
if not re_register:
stored = load_transform(cfg_dir, csv_name, eplan_name, fingerprint)
if stored:
verify_transform(stored, layout_pts, e_pts)
return stored
transform = derive_transform(layout_pts, e_pts, csv_name, eplan_name)
transform.fingerprint = fingerprint
saved = save_transform(cfg_dir, transform)
warnings.append(f"Lagebezug neu ermittelt und abgelegt: {saved}")
return transform
# ---------------------------------------------------------------------------
# Slots
# ---------------------------------------------------------------------------
def points_from_parts(parts: list[LayoutPart]) -> list[LayoutPoint]:
"""Punktartige Bauteile aus TroDocument.parts."""
points = []
for p in parts:
if p.position is None:
continue
kind = next((k for k in LAYOUT_KINDS if k.lower() in p.kind.lower()), "")
if kind:
points.append(LayoutPoint(p.layout_id, kind, p.position[0], p.position[1],
p.name))
return points
def _binding_from(a: Assignment, device: HwDevice | None, tro: str = "",
slot: str = "", role: str = "") -> Binding:
status, note = "zugeordnet", ""
if device is None or not device.in_tia:
status, note = "widerspruch", "BMK nicht im TIA-Import"
if not tro:
status = "ohne_slot" if status == "zugeordnet" else status
note = note or f"{a.layout_kind} {a.layout_id} haengt an keinem TRO-Slot"
return Binding(
tro=tro, slot=slot, role=role or a.role, layout_id=a.layout_id, bmk=a.bmk,
constant=device.constant if device else "",
address=device.address if device else "",
source="positionsdraw", rule=a.rule, distance=a.distance,
runner_up=a.runner_up, status=status, note=note,
)
def bind_slots(doc: TroDocument, assignments: list[Assignment],
rules: list[Rule], hw: HardwareDocument) -> list[Binding]:
"""Zuordnungen an die Slot-Rollen der TROs haengen.
Ein Bauteil kann in zwei Slots desselben TRO stehen (Speicher:
``separator1`` und ``line1``) - dann bekommen beide die Zuordnung.
"""
devices = hw.by_bmk()
slots_of: dict[str, list[tuple[str, str, list[str]]]] = {}
for tro in doc.tros:
for s in tro.slots:
if s.layout_id:
roles = [r.rstrip("?") for r in s.roles]
slots_of.setdefault(s.layout_id, []).append((tro.tro_id, s.slot, roles))
rule_of = {r.role: r for r in rules}
bindings: list[Binding] = []
for a in assignments:
role = rule_of[a.role].slot_role if a.role in rule_of else a.role.split(".")[-1]
targets = [(t, s) for t, s, roles in slots_of.get(a.layout_id, []) if role in roles]
if not targets:
bindings.append(_binding_from(a, devices.get(a.bmk)))
continue
for tro_id, slot in targets:
bindings.append(_binding_from(a, devices.get(a.bmk), tro_id, slot, role))
return bindings
# ---------------------------------------------------------------------------
# Foerderer
# ---------------------------------------------------------------------------
def assign_conveyors(doc: TroDocument, hw: HardwareDocument, transform: Transform
) -> list[ConveyorBinding]:
"""Foerderer der E-Planung (K0102) per Mehrheit einem Layout-Foerderer zuordnen.
Jedes Feldgeraet, dessen Klartext einen Foerderer nennt, stimmt fuer den
Layout-Foerderer, dessen Huellbox ihm am naechsten liegt (hoechstens
CONVEYOR_MAX_MM). Zugeordnet wird bei mindestens CONVEYOR_MIN_VOTES
Stimmen und CONVEYOR_MIN_SHARE Anteil; je Layout-Foerderer hoechstens ein
Foerderer der E-Planung (der mit den meisten Stimmen).
"""
conveyors = [p for p in doc.parts if p.kind in CONVEYOR_KINDS and p.bbox and p.position]
drives = {dr.conveyor: dr for dr in hw.drives if dr.conveyor}
kind_of = _conveyor_kinds(hw)
votes: dict[str, Counter] = {}
totals: Counter = Counter()
for d in hw.field_devices():
if not d.conveyor:
continue
totals[d.conveyor] += 1
# Die Huellboxen ueberlappen (Speicherlinien reichen bis an den
# Kreisel); nennt der Klartext die Art, stimmen nur Foerderer dieser Art.
kinds = _LAYOUT_KINDS_FOR.get(kind_of.get(d.conveyor, ""), CONVEYOR_KINDS)
x, y = transform.to_csv(*d.pos)
best = min(((c.distance_to_box(x, y), c.layout_id) for c in conveyors
if c.kind in kinds), default=(None, ""))
if best[0] is not None and best[0] <= CONVEYOR_MAX_MM:
votes.setdefault(d.conveyor, Counter())[best[1]] += 1
names = sorted(set(totals) | set(drives))
name_of = {p.layout_id: p.name for p in doc.parts}
candidates = []
for name in names:
counter = votes.get(name, Counter())
layout_id, n = counter.most_common(1)[0] if counter else ("", 0)
candidates.append((n, name, layout_id))
taken: dict[str, str] = {}
result = []
for n, name, layout_id in sorted(candidates, key=lambda c: (-c[0], c[1])):
total = totals.get(name, 0)
dr = drives.get(name)
cb = ConveyorBinding(eplan_name=name, votes=n, total=total, rule="mehrheit",
motor=dr.motor if dr else "",
protection=dr.protection if dr else "",
encoder=dr.encoder if dr else "")
share = n / total if total else 0.0
if not layout_id:
cb.status = "offen"
cb.note = ("kein Feldgeraet nennt diesen Foerderer" if not total else
f"keine Stimme innerhalb {CONVEYOR_MAX_MM:.0f} mm eines Layout-Foerderers")
elif n < CONVEYOR_MIN_VOTES or share < CONVEYOR_MIN_SHARE:
cb.status = "offen"
cb.layout_id, cb.layout_name = layout_id, name_of.get(layout_id, "")
cb.note = (f"unsicher: {n} von {total} Stimmen fuer {cb.layout_name or layout_id} "
f"(noetig {CONVEYOR_MIN_VOTES} und {CONVEYOR_MIN_SHARE:.0%})")
elif layout_id in taken:
cb.status = "widerspruch"
cb.layout_id, cb.layout_name = layout_id, name_of.get(layout_id, "")
cb.note = f"{layout_id} ist schon {taken[layout_id]} zugeordnet (mehr Stimmen)"
else:
taken[layout_id] = name
cb.layout_id, cb.layout_name = layout_id, name_of.get(layout_id, "")
others = votes.get(name, Counter())
rest = [f"{name_of.get(k, k)} {v}" for k, v in others.most_common() if k != layout_id]
if rest:
cb.note = "Gegenstimmen: " + ", ".join(rest)
result.append(cb)
return sorted(result, key=lambda c: c.eplan_name)
# Art des Foerderers im Klartext der E-Planung -> Arten im Layout
_LAYOUT_KINDS_FOR = {"Kreisel": ("Kreisel",), "Strecke": ("Strecke", "Gefaellestrecke")}
def _conveyor_kinds(hw: HardwareDocument) -> dict[str, str]:
"""Art je Foerderer der E-Planung: Mehrheit aller Klartexte, die ihn nennen.
"PRUEFUNG BT. POS.1 (KR-K0102 EINFAHRT PUFFER)" und "ANTRIEB
KREISEL-K0102" sagen Kreisel; ein Geraet ohne Artwort erbt die Art.
"""
counts: dict[str, Counter] = {}
for d in hw.devices:
if d.conveyor and d.conveyor_kind:
counts.setdefault(d.conveyor, Counter())[d.conveyor_kind] += 1
for dr in hw.drives:
if dr.conveyor and dr.conveyor_kind:
counts.setdefault(dr.conveyor, Counter())[dr.conveyor_kind] += 1
return {name: c.most_common(1)[0][0] for name, c in counts.items()}
def bind_drives(doc: TroDocument, conveyors: list[ConveyorBinding],
hw: HardwareDocument) -> list[Binding]:
"""Antriebs-Slots (Vario): Motor und Motorschutz des zugeordneten Foerderers."""
devices = hw.by_bmk()
by_layout = {c.layout_id: c for c in conveyors if c.status == "zugeordnet"}
bindings = []
for tro in doc.tros:
for s in tro.slots:
if not s.layout_id:
continue
conv = by_layout.get(s.layout_id.split("-", 1)[0])
if conv is None:
continue
roles = [r.rstrip("?") for r in s.roles]
for role, attr in _DRIVE_ROLES.items():
bmk = getattr(conv, attr)
if role not in roles or not bmk:
continue
dev = devices.get(bmk)
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}"))
return bindings
# ---------------------------------------------------------------------------
# Handkorrekturen
# ---------------------------------------------------------------------------
@dataclass
class Overrides:
"""Handkorrekturen aus hw_overrides.ini fuer eine Anlage."""
# (tro, slot, role) -> BMK ("" = Zuordnung loeschen, Rolle offen)
bindings: dict[tuple[str, str, str], str] = field(default_factory=dict)
# Foerderer der E-Planung -> Layout-ID ("" = Zuordnung loeschen)
conveyors: dict[str, str] = field(default_factory=dict)
def load_overrides(path: Path, layout_source: str) -> Overrides:
"""Abschnitte ``[<csv>.bindings]`` und ``[<csv>.conveyors]`` lesen.
``TRO01.separator1.sensor = BG0532`` setzt ein BMK, ``= -`` loescht die
Zuordnung (Rolle bleibt offen); ``K0102 = 0010`` setzt einen Foerderer.
"""
result = Overrides()
if not path.is_file():
return result
parser = configparser.ConfigParser(inline_comment_prefixes=(";", "#"),
interpolation=None)
parser.optionxform = str
with path.open(encoding="utf-8") as fh:
parser.read_file(fh)
section = f"{layout_source}.bindings"
if parser.has_section(section):
for key, value in parser[section].items():
parts = key.split(".")
if len(parts) != 3:
raise ValueError(f"{path.name} [{section}] '{key}': erwartet "
f"<TRO-ID>.<Slot>.<Rolle>")
result.bindings[(parts[0], parts[1], parts[2].rstrip("?"))] = \
"" if value.strip() == "-" else value.strip()
section = f"{layout_source}.conveyors"
if parser.has_section(section):
for key, value in parser[section].items():
result.conveyors[key.strip()] = "" if value.strip() == "-" else value.strip()
return result
def _apply_overrides(doc: TroDocument, bindings: list[Binding],
conveyors: list[ConveyorBinding], overrides: Overrides,
hw: HardwareDocument, warnings: list[str]) -> list[Binding]:
devices = hw.by_bmk()
slots = {(t.tro_id, s.slot): s for t in doc.tros for s in t.slots}
for (tro, slot, role), bmk in overrides.bindings.items():
s = slots.get((tro, slot))
if s is None:
warnings.append(f"{OVERRIDES_FILE}: {tro}.{slot} gibt es in der TRO-JSON nicht "
f"- Korrektur nicht angewendet")
continue
bindings = [b for b in bindings
if not (b.tro == tro and b.slot == slot and b.role == role)]
if not bmk:
continue
# dasselbe BMK nicht zusaetzlich automatisch an anderer Stelle
bindings = [b for b in bindings if b.bmk != bmk or b.rule == "hand"]
dev = devices.get(bmk)
bindings.append(Binding(
tro=tro, slot=slot, role=role, layout_id=s.layout_id, bmk=bmk,
constant=dev.constant if dev else "", address=dev.address if dev else "",
source=OVERRIDES_FILE, rule="hand",
status="zugeordnet" if dev and dev.in_tia else "widerspruch",
note="" if dev and dev.in_tia else
("BMK nicht im TIA-Import" if dev else "BMK unbekannt in der Hardwareplanung")))
name_of = {p.layout_id: p.name for p in doc.parts}
for name, layout_id in overrides.conveyors.items():
cb = next((c for c in conveyors if c.eplan_name == name), None)
if cb is None:
cb = ConveyorBinding(eplan_name=name)
conveyors.append(cb)
if layout_id and doc.part(layout_id) is None:
warnings.append(f"{OVERRIDES_FILE}: Foerderer {name} -> {layout_id}: Layout-ID "
f"unbekannt - Korrektur nicht angewendet")
continue
cb.layout_id, cb.layout_name = layout_id, name_of.get(layout_id, "")
cb.rule = "hand"
cb.status = "zugeordnet" if layout_id else "offen"
cb.note = "von Hand"
return bindings
# ---------------------------------------------------------------------------
# Zusammenbau
# ---------------------------------------------------------------------------
def _open_and_conflicts(doc: TroDocument, bindings: list[Binding]) -> list[Binding]:
"""Pflichtrollen ohne BMK als ``offen``; doppelt vergebene BMK als Widerspruch."""
have = {(b.tro, b.slot, b.role) for b in bindings if b.tro}
for tro in doc.tros:
for s in tro.slots:
for role in s.roles:
if role.endswith("?") or (tro.tro_id, s.slot, role) in have:
continue
bindings.append(Binding(tro=tro.tro_id, slot=s.slot, role=role,
layout_id=s.layout_id, status="offen"))
# Ein BMK an zwei verschiedenen Bauteilen ist ein Widerspruch. Derselbe
# Separator in zwei Slots eines Speichers (separator1/line1) ist keiner.
places: dict[str, set[tuple[str, str]]] = {}
for b in bindings:
if b.bmk and b.status != "offen":
places.setdefault(b.bmk, set()).add((b.layout_id, b.role))
for b in bindings:
if b.bmk and len(places.get(b.bmk, ())) > 1 and b.status == "zugeordnet":
b.status = "widerspruch"
b.note = "BMK mehrfach vergeben"
order = {t.tro_id: i for i, t in enumerate(doc.tros)}
return sorted(bindings, key=lambda b: (b.tro == "", order.get(b.tro, 0), b.slot,
b.role, b.layout_id))
def build_anlage(doc: TroDocument, hw: HardwareDocument, rules: list[Rule],
anchors: list[str], cfg_dir: Path, layout_file: str, hw_file: str,
overrides: Overrides | None = None, re_register: bool = False,
transform: Transform | None = None) -> AnlageDocument:
"""Kompletter Lauf ohne Ausgabe: TRO-JSON + Hardware-JSON -> AnlageDocument.
``transform`` vorgeben (Tests) ueberspringt die Ermittlung des Lagebezugs.
"""
warnings: list[str] = []
if not doc.parts:
raise ValueError(f"{layout_file} enthaelt keine 'parts' (Lage der Bauteile) - "
f"mit tro_flow --json neu erzeugen")
points = points_from_parts(doc.parts)
devices: list[EDevice] = list(hw.devices)
assignments: list[Assignment] = []
conveyors: list[ConveyorBinding] = []
if hw.field_devices():
if transform is None:
transform = resolve_transform(points, devices, anchors, cfg_dir, doc.source,
hw.positions_file, re_register, warnings,
hw.positions_sha256)
share = transform.matched / transform.candidates if transform.candidates else 1.0
if share < REGISTRATION_MIN_SHARE:
warnings.append(
f"Lagebezug passt nur fuer {transform.matched} von {transform.candidates} "
f"Separatoren ({share:.0%}) - Layout und E-Zeichnung sind vermutlich "
f"verschiedene Planungsstaende; raeumliche Zuordnungen pruefen")
assignments = assign(points, devices, rules, transform)
conveyors = assign_conveyors(doc, hw, transform)
else:
warnings.append("Hardwareplanung ohne Lage (keine Positionsdatei) - keine "
"raeumliche Zuordnung, alle Rollen offen")
bindings = bind_slots(doc, assignments, rules, hw)
if overrides is not None:
bindings = _apply_overrides(doc, bindings, conveyors, overrides, hw, warnings)
bindings += bind_drives(doc, conveyors, hw)
bindings = _open_and_conflicts(doc, bindings)
used = {b.bmk for b in bindings if b.bmk}
unassigned = sorted(d.bmk for d in hw.field_devices() if d.bmk not in used)
registration = None
if transform is not None:
registration = Registration(
dx=round(transform.dx, 1), dy=round(transform.dy, 1),
rotation=round(transform.rotation, 4), source=transform.source,
matched=transform.matched, candidates=transform.candidates,
residual_max=round(transform.residual_max, 1),
description=transform.describe())
return AnlageDocument(
layout_file=layout_file, hw_file=hw_file, project=hw.project,
registration=registration, layout=doc, bindings=bindings,
conveyors=sorted(conveyors, key=lambda c: c.eplan_name),
unassigned=unassigned, warnings=warnings + hw.checks,
)
# ---------------------------------------------------------------------------
# Bericht
# ---------------------------------------------------------------------------
def render_report(anlage: AnlageDocument, hw: HardwareDocument) -> str:
out: list[str] = []
add = out.append
devices = hw.by_bmk()
add(f"# Anlage: {anlage.layout.source} + {anlage.project}")
add("")
add(f"- Layout: `{anlage.layout_file}`")
add(f"- Hardwareplanung: `{anlage.hw_file}` (TIA `{hw.tia_file or '-'}`, "
f"Positionen `{hw.positions_file or '-'}`)")
add(f"- Erzeugt: {anlage.created}")
if anlage.registration:
add(f"- Lagebezug: {anlage.registration.description}")
add("")
add("## Slot-Rollen")
add("")
add("| Slot-Art | Rolle | zugeordnet | offen | Widerspruch |")
add("|---|---|---|---|---|")
stats: dict[tuple[str, str], Counter] = {}
for b in anlage.bindings:
if not b.tro:
continue
kind = b.slot.rstrip("0123456789")
stats.setdefault((kind, b.role), Counter())[b.status] += 1
for (kind, role), c in sorted(stats.items()):
add(f"| {kind} | {role} | {c['zugeordnet']} | {c['offen']} | {c['widerspruch']} |")
add("")
add("## Foerderer (E-Planung -> Layout)")
add("")
if anlage.conveyors:
add("| E-Planung | Layout | Stimmen | Motor | Motorschutz | Geber | Status | Hinweis |")
add("|---|---|---|---|---|---|---|---|")
for c in anlage.conveyors:
layout = f"{c.layout_name} ({c.layout_id})" if c.layout_id else "-"
add(f"| {c.eplan_name} | {layout} | {c.votes}/{c.total} | {c.motor or '-'} "
f"| {c.protection or '-'} | {c.encoder or '-'} | {c.status} | {c.note} |")
else:
add("- keine")
add("")
add("## Zuordnungen je TRO")
add("")
add("Abstand nach dem Lagebezug; \"2.\" = naechstbester Kandidat (knapp = pruefen).")
add("")
add("| TRO | Slot | Rolle | Layout | BMK | Konstante | Adresse | VERW | Abstand mm "
"| 2. | Regel | Status |")
add("|---|---|---|---|---|---|---|---|---|---|---|---|")
for b in anlage.bindings:
if not b.tro or b.status == "offen":
continue
d = devices.get(b.bmk)
dist = "" if b.distance is None else f"{b.distance:.0f}"
second = "" if b.runner_up is None else f"{b.runner_up:.0f}"
status = b.status + (f" ({b.note})" if b.note and b.status != "zugeordnet" else "")
add(f"| {b.tro} | {b.slot} | {b.role} | {b.layout_id} | {b.bmk} | {b.constant} "
f"| {b.address} | {d.verw if d else ''} | {dist} | {second} | {b.rule} | {status} |")
add("")
add("## Offene Pflichtrollen")
add("")
open_by: dict[str, list[str]] = {}
for b in anlage.bindings:
if b.status == "offen":
open_by.setdefault(b.tro, []).append(f"{b.slot}.{b.role}")
if open_by:
for tro, roles in open_by.items():
add(f"- **{tro}**: {', '.join(roles)}")
else:
add("- keine")
add("")
loose = [b for b in anlage.bindings if not b.tro]
add("## Zugeordnet, aber ohne TRO-Slot")
add("")
if loose:
add("Diese Bauteile haengen (noch) an keinem Slot der TRO-JSON, z. B. "
"Ausschleuselemente: welcher TRO-Ausgang zu welchem Element gehoert, "
"weiss tro_flow noch nicht.")
add("")
add("| Regel | Layout | BMK | Abstand mm | Status |")
add("|---|---|---|---|---|")
for b in loose:
dist = "" if b.distance is None else f"{b.distance:.0f}"
add(f"| {b.role} | {b.layout_id} | {b.bmk} | {dist} | {b.status} |")
else:
add("- keine")
add("")
add("## Feldgeraete ohne Zuordnung")
add("")
if anlage.unassigned:
add("| BMK | VERW | Bezeichnung | Station |")
add("|---|---|---|---|")
for bmk in anlage.unassigned:
d = devices[bmk]
add(f"| {bmk} | {d.verw} | {d.bezeichnung} | {d.controller}-{d.station} |")
else:
add("- keine")
add("")
if anlage.warnings:
add("## Hinweise")
add("")
for w in anlage.warnings:
add(f"- {w}")
add("")
return "\n".join(out)
# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------
def _in_results(name: str, results: Path) -> Path:
path = Path(name)
if path.is_absolute() or path.parent != Path(".") or path.is_file():
return path
return results / name
def main(argv: list[str] | None = None) -> int:
from material_flow import env_dir
args = parse_args(argv)
results = env_dir("SKEL_RESULTS", "results")
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
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)):
if not path.is_file():
print(f"FEHLER: {label} nicht gefunden: {path}", file=sys.stderr)
return 1
try:
doc = TroDocument.from_json(tro_path.read_text(encoding="utf-8"))
except ValidationError as exc:
print(f"FEHLER: {tro_path} ist keine gueltige TRO-JSON ({exc.error_count()} "
f"Abweichungen) - mit tro_flow --json neu erzeugen.", file=sys.stderr)
return 1
try:
hw = HardwareDocument.from_json(hw_path.read_text(encoding="utf-8"))
except ValidationError as exc:
print(f"FEHLER: {hw_path} passt nicht zum Schema '{HW_SCHEMA}' "
f"({exc.error_count()} Abweichungen) - mit hw_import neu erzeugen.",
file=sys.stderr)
return 1
try:
rules, anchors = load_rules(rules_path)
overrides = load_overrides(overrides_path, doc.source)
anlage = build_anlage(doc, hw, rules, anchors, cfg_dir, tro_path.name, hw_path.name,
overrides, args.re_register)
except RegistrationError as exc:
print(f"FEHLER: Lagebezug E-Zeichnung -> Layout nicht nachweisbar: {exc}",
file=sys.stderr)
print(f" Passen Layout und E-Zeichnung zusammen? Anker-VERW in "
f"{rules_path.name} [Registrierung] pruefen.", file=sys.stderr)
return 1
except (OSError, ValueError) as exc:
print(f"FEHLER: {exc}", file=sys.stderr)
return 1
stem = Path(doc.source).stem or tro_path.stem.removesuffix("_tro")
results.mkdir(parents=True, exist_ok=True)
json_file = results / f"{stem}_anlage.json"
md_file = results / f"{stem}_anlage.md"
json_file.write_text(anlage.to_json(), encoding="utf-8")
md_file.write_text(render_report(anlage, hw), encoding="utf-8")
status = Counter(b.status for b in anlage.bindings)
print("")
print("================================================================")
print("LAYOUT + HARDWAREPLANUNG -> ANLAGE")
print("================================================================")
print(f"TRO-JSON = {tro_path}")
print(f"Hardware-JSON = {hw_path}")
print(f"Lagebezug = {anlage.registration.description if anlage.registration else '-'}")
print(f"Slot-Rollen = {status['zugeordnet']} zugeordnet, {status['offen']} offen, "
f"{status['widerspruch']} Widerspruch")
print(f"Ohne Slot = {status['ohne_slot']} Zuordnungen an Bauteilen ohne TRO-Slot")
conv_ok = [c for c in anlage.conveyors if c.status == "zugeordnet"]
print(f"Foerderer = {len(conv_ok)} von {len(anlage.conveyors)} zugeordnet"
+ (": " + ", ".join(f"{c.eplan_name}={c.layout_name or c.layout_id}"
for c in conv_ok) if conv_ok else ""))
print(f"Ohne Zuordnung = {len(anlage.unassigned)} Feldgeraete")
for w in anlage.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="merge_hw_layout",
description="Fuehrt die TRO-JSON (tro_flow --json) und die Hardware-JSON "
"(hw_import) zusammen: ordnet Sensoren, Ventile und Antriebe den "
"Slots der TROs zu und schreibt <csv>_anlage.json plus einen "
"Bericht <csv>_anlage.md.",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="Beispiele:\n"
" merge_hw_layout.bat --tro mubea_tro.json --hw Mubea_hw.json\n"
" merge_hw_layout.bat --tro mubea_tro.json --hw Mubea_hw.json --re-register\n"
"\n"
"Exit-Codes:\n"
" 0 ok (auch wenn nicht alles zugeordnet wurde - siehe Bericht)\n"
" 1 Eingabe-, Aufruf- oder Lagebezugsfehler\n",
)
parser.add_argument("--tro", required=True, metavar="NAME",
help="TRO-JSON <csv>_tro.json in %%SKEL_RESULTS%% (oder ein Pfad)")
parser.add_argument("--hw", required=True, metavar="NAME",
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}")
parser.add_argument("--overrides", metavar="DATEI",
help=f"Handkorrekturen. Standard: %%SKEL_CFG%%\\{OVERRIDES_FILE} "
f"(fehlt sie, keine)")
parser.add_argument("--re-register", action="store_true",
help="Lagebezug neu ermitteln, auch wenn in %%SKEL_CFG%% "
"einer abgelegt ist")
return parser.parse_args(argv)
if __name__ == "__main__":
sys.exit(main())
lib/model.py
+224 -4
View File
@@ -26,9 +26,17 @@ Klassen:
Hardwareplanung dort fuellen muss
JamArea ein Staubereich: Eingang genau eines TRO (je Separator
laut Katalog), mit allen Vorgaengern, die dort einfahren
LayoutPart ein Bauteil des Layouts mit Lage und Huellbox - das Ziel
der raeumlichen Zuordnung in merge_hw_layout (die TRO-JSON traegt
es, damit merge_hw_layout das CSV nicht lesen muss)
Analysis Arbeitsergebnis von tro_flow.analyse_tros
TroDocument die TRO-JSON (<csv>_tro.json): von tro_flow --json
geschrieben, von scl_skeleton gelesen und geprueft
Binding eine Zuordnung Hardware -> TRO-Slot-Rolle (merge_hw_layout)
ConveyorBinding ein Foerderer der Hardwareplanung (K0102) an einem
Layout-Foerderer, mit Antrieb, Motorschutz, Geber
AnlageDocument die Anlagen-JSON (<csv>_anlage.json): Layout unveraendert
plus bindings[] - von merge_hw_layout geschrieben
In der JSON stehen die Schluessel wie bisher in der TRO-JSON (`id`, `type`,
`fbType`, `candidates`, ...); im Python-Code heissen die Felder weiter
@@ -59,22 +67,44 @@ __all__ = [
"TroBase",
"Slot",
"JamArea",
"LayoutPart",
"DerivedTro",
"Connection",
"TroCandidate",
"Analysis",
"TroDocument",
"Registration",
"Binding",
"ConveyorBinding",
"AnlageDocument",
"ANLAGE_SCHEMA",
"BINDING_STATUS",
"derive_slots",
"derive_jam_areas",
"storage_lines",
]
# Kennung der TRO-JSON. v1 war die Ausgabe von tro_extract (aus der Zeichnung),
# v2 die von tro_flow --json (aus dem CSV, ueber dieses Modell), v3 wie v2 mit
# `candidates` statt `openPoints`, v4 wie v3 mit `slots` je TRO und `jamAreas`.
# Eine v3-Datei liest sich weiter (die neuen Felder fehlen dann und werden beim
# Rendern abgeleitet, siehe TroDocument.jam_areas_or_derived).
# Rendern abgeleitet, siehe TroDocument.jam_areas_or_derived). Seit 2026-10-15
# traegt v4 optional `parts` (Lage der Layout-Bauteile fuer merge_hw_layout); eine
# v4-Datei ohne `parts` liest sich weiter, nur merge_hw_layout braucht sie.
SCHEMA = "sps_skel-tro-v4"
# Kennung der Anlagen-JSON (<csv>_anlage.json, merge_hw_layout)
ANLAGE_SCHEMA = "sps_skel-anlage-v1"
# Zustand einer Zuordnung (doc/Konzept_Layout_Hardware_SCL.md, Umbau 3):
# zugeordnet ein BMK steht fest
# offen Pflichtrolle eines Slots ohne BMK -> TODO(E-Planung)
# widerspruch BMK steht fest, passt aber nicht (fehlt im TIA-Import,
# doppelt vergeben, Handkorrektur auf unbekanntes BMK)
# ohne_slot E-Bauteil einem Layout-Bauteil zugeordnet, das (noch) an
# keinem TRO-Slot haengt (z. B. Ausschleuselemente)
BINDING_STATUS = ("zugeordnet", "offen", "widerspruch", "ohne_slot")
# Gemeinsame Einstellungen: Pruefung bei Erzeugung und Zuweisung, unbekannte
# Schluessel sind ein Fehler (ein umbenannter JSON-Schluessel faellt auf statt
# still auf einen Standardwert zurueckzufallen), Aliase und Feldnamen gelten
@@ -246,6 +276,34 @@ class JamArea(_Positioned):
position: tuple[float, float, float] | None = None
class LayoutPart(_Positioned):
"""
Ein Bauteil des Layouts (eine CSV-Zeile mit Lage), z. B. Separator,
Ausschleuselement, Kreisel.
Ziel der raeumlichen Zuordnung in merge_hw_layout: die Slots nennen nur die
``layout_id``, die Lage steht hier. ``bbox`` ist die achsparallele
Ausdehnung (mm) um ``position`` aus der CSV-Spalte "Boundingbox" -
gebraucht, um Geraete einem langen Foerderer (Kreisel) zuzuordnen.
"""
model_config = _CONFIG
layout_id: str = Field(alias="layoutId")
kind: str
name: str = ""
position: tuple[float, float, float] | None = None
bbox: tuple[float, float, float] | None = None
def distance_to_box(self, x: float, y: float) -> float | None:
"""Abstand eines Punkts zur Huellbox (0 innerhalb); None ohne Lage/Box."""
if not self.position or not self.bbox:
return None
dx = max(abs(x - self.position[0]) - self.bbox[0] / 2, 0.0)
dy = max(abs(y - self.position[1]) - self.bbox[1] / 2, 0.0)
return (dx * dx + dy * dy) ** 0.5
class DerivedTro(TroBase):
"""
Ein von tro_flow aus dem CSV abgeleiteter TRO (oder eine Systemgrenze).
@@ -346,14 +404,18 @@ class TroDocument(BaseModel):
connections: list[Connection] = Field(default_factory=list)
candidates: list[TroCandidate] = Field(default_factory=list)
jam_areas: list[JamArea] = Field(default_factory=list, alias="jamAreas")
# Layout-Bauteile mit Lage (fuer merge_hw_layout); leer in aelteren Dateien
parts: list[LayoutPart] = Field(default_factory=list)
findings: list[str] = Field(default_factory=list)
@classmethod
def from_analysis(cls, analysis: Analysis, source: str, note: str = "") -> TroDocument:
def from_analysis(cls, analysis: Analysis, source: str, note: str = "",
parts: list[LayoutPart] | None = None) -> TroDocument:
"""Dokument aus dem Ergebnis von tro_flow.analyse_tros bilden.
Slots und Staubereiche werden hier abgeleitet, damit die JSON sie
traegt und nicht erst das Rendern sie kennt.
traegt und nicht erst das Rendern sie kennt. ``parts`` (Lage der
Layout-Bauteile) gibt tro_flow aus dem CSV mit.
"""
tros = [t.model_copy(update={"slots": derive_slots(t)}) for t in analysis.tros]
return cls(
@@ -362,6 +424,7 @@ class TroDocument(BaseModel):
tros=tros,
boundaries=analysis.boundaries,
jam_areas=derive_jam_areas(tros),
parts=list(parts or []),
connections=[
Connection(src=src, dst=dst, kind=analysis.edge_kinds.get((src, dst), ""))
for src, dst in analysis.edges
@@ -369,6 +432,11 @@ class TroDocument(BaseModel):
findings=analysis.findings,
)
def part(self, layout_id: str) -> LayoutPart | None:
"""Layout-Bauteil zur TeileId; ein Bahn-Suffix ("0010-R") zaehlt nicht."""
base = layout_id.split("-", 1)[0]
return next((p for p in self.parts if p.layout_id == base), None)
@classmethod
def from_json(cls, text: str) -> TroDocument:
"""JSON-Text einlesen und pruefen (pydantic.ValidationError bei Fehlern)."""
@@ -389,6 +457,138 @@ class TroDocument(BaseModel):
return None
# ---------------------------------------------------------------------------
# Anlagen-JSON: Layout + Zuordnungen der Hardwareplanung (merge_hw_layout)
# ---------------------------------------------------------------------------
class Registration(BaseModel):
"""Lagebezug E-Zeichnung -> Layout, mit dem die Zuordnung lief (Nachweis).
Kopie der Kennwerte aus dxf_registration.Transform - model.py importiert
dxf_registration bewusst nicht (Schichten, siehe Kopf).
"""
model_config = _CONFIG
dx: float
dy: float
rotation: float = 0.0
source: str = ""
matched: int = 0
candidates: int = 0
residual_max: float = Field(default=0.0, alias="residualMax")
description: str = ""
class Binding(BaseModel):
"""
Eine Zuordnung: Rolle ``role`` im Slot ``slot`` des TRO ``tro`` <- BMK.
Stabiler Schluessel ist TRO-ID + Slot + Rolle (TRO-IDs aus der ID-Datei);
``layout_id`` dient nur zur Nachverfolgung im aktuellen Layoutstand. Bei
``status == "ohne_slot"`` sind ``tro``/``slot`` leer und ``role`` ist die
Regel (``Ausschleus.stau``). ``constant``/``address`` kommen aus dem
TIA-Import und sind das, was im SCL steht (``cInBG0532``, ``%E053.2``).
"""
model_config = _CONFIG
tro: str = ""
slot: str = ""
role: str
layout_id: str = Field(default="", alias="layoutId")
bmk: str = ""
constant: str = ""
address: str = ""
# positionsdraw | hw_overrides.ini
source: str = ""
# raeumlich | nummer | hand
rule: str = ""
distance: float | None = None
runner_up: float | None = Field(default=None, alias="runnerUp")
status: str = "zugeordnet"
note: str = ""
@field_validator("status")
@classmethod
def _known_status(cls, value: str) -> str:
if value not in BINDING_STATUS:
raise ValueError(f"status '{value}' unbekannt, erlaubt: {', '.join(BINDING_STATUS)}")
return value
class ConveyorBinding(BaseModel):
"""
Ein Foerderer der Hardwareplanung (``K0102``, ``K0106``) an einem
Layout-Foerderer (``layout_id`` "0010" = Kreisel3).
Die Namenswelten passen nicht zusammen (Layout: Kreisel1..3, E-Planung:
K0102/K0104/K0105), deshalb raeumlich: alle Feldgeraete, deren
Bezeichnung den Foerderer nennt, stimmen fuer den naechsten
Layout-Foerderer ab (``votes`` von ``total``).
"""
model_config = _CONFIG
eplan_name: str = Field(alias="eplanName")
layout_id: str = Field(default="", alias="layoutId")
layout_name: str = Field(default="", alias="layoutName")
votes: int = 0
total: int = 0
motor: str = ""
protection: str = ""
encoder: str = ""
# mehrheit | hand
rule: str = ""
status: str = "zugeordnet"
note: str = ""
class AnlageDocument(BaseModel):
"""
Die Anlagen-JSON ``<csv>_anlage.json``: das Layout (TRO-JSON) unveraendert
unter ``layout``, daneben die Zuordnungen der Hardwareplanung.
Geschrieben von merge_hw_layout, kuenftig die eine Eingabe von scl_skeleton
(doc/Konzept_Layout_Hardware_SCL.md, Umbau 4).
"""
model_config = _CONFIG
schema_id: str = Field(default=ANLAGE_SCHEMA, alias="schema")
created: str = Field(default_factory=lambda: datetime.now().strftime("%Y-%m-%d %H:%M"))
layout_file: str = Field(default="", alias="layoutFile")
hw_file: str = Field(default="", alias="hwFile")
project: str = ""
registration: Registration | None = None
layout: TroDocument
bindings: list[Binding] = Field(default_factory=list)
conveyors: list[ConveyorBinding] = Field(default_factory=list)
# Feldgeraete mit Lage, die keine Zuordnung bekommen haben (BMK)
unassigned: list[str] = Field(default_factory=list)
warnings: list[str] = Field(default_factory=list)
def binding(self, tro_id: str, slot: str, role: str) -> Binding | None:
"""Zuordnung einer Slot-Rolle (Rolle ohne "?"), None wenn keine."""
role = role.rstrip("?")
return next((b for b in self.bindings
if b.tro == tro_id and b.slot == slot and b.role == role), None)
def conveyor_for(self, layout_id: str) -> ConveyorBinding | None:
"""Foerderer der Hardwareplanung an einem Layout-Foerderer ("0010-R" geht)."""
base = layout_id.split("-", 1)[0]
return next((c for c in self.conveyors
if c.layout_id == base and c.status == "zugeordnet"), None)
def to_json(self) -> str:
return self.model_dump_json(by_alias=True, indent=2) + "\n"
@classmethod
def from_json(cls, text: str) -> AnlageDocument:
return cls.model_validate_json(text)
# ---------------------------------------------------------------------------
# Ableitungen aus Katalogtyp und Topologie
# ---------------------------------------------------------------------------
@@ -399,23 +599,43 @@ _SWITCH_ROLES = ("valve", "exitSensor?")
_SCANNER_ROLES = ("reader",)
_DRIVE_ROLES = ("motor", "motorProtection", "senCarrInPos", "senFinger",
"senLastPos", "senJam")
_STORAGE_LINE_ROLES = ("sensor", "stopper", "jamSensor?")
def _at(values: list[str], index: int) -> str:
return values[index] if index < len(values) else ""
def storage_lines(tro: TroBase) -> list[str]:
"""Separatoren der Lagerlinien eines Speicher-TRO.
tro_flow legt sie als [steuernde ..., Linien ...] ab; die Linien sind die
letzten len(hosts) (eine je Speicherbahn). Fehlen Hosts, zaehlt jeder
Separator als Linie - dieselbe Regel wie scl_context._store.
"""
hosts = list(getattr(tro, "hosts", []) or [])
seps = list(tro.separators)
count = len(hosts) if 0 < len(hosts) <= len(seps) else len(seps)
return seps[len(seps) - count:]
def derive_slots(tro: DerivedTro) -> list[Slot]:
"""Slots eines TRO: Anzahl aus dem Katalog, Layout-ID aus dem TRO.
Weichen tragen noch keine Layout-ID - tro_flow speichert das Weichen-
bzw. Ausschleuselement heute nicht am TRO.
bzw. Ausschleuselement heute nicht am TRO. Ein Speicher bekommt je
Lagerlinie einen Slot ``line<n>``; der erste Separator-Slot und
``line1`` koennen dasselbe Bauteil sein (wie im PinStore-Template).
"""
items = tro.items
slots: list[Slot] = []
for i in range(items.get(TroItem.SEPARATOR, 0)):
slots.append(Slot(slot=f"separator{i + 1}", layout_id=_at(tro.separators, i),
roles=list(_SEPARATOR_ROLES)))
if items.get(TroItem.STORAGE_LINE):
for n, sep in enumerate(storage_lines(tro), start=1):
slots.append(Slot(slot=f"line{n}", layout_id=sep,
roles=list(_STORAGE_LINE_ROLES)))
for i in range(items.get(TroItem.SWITCH, 0)):
slots.append(Slot(slot=f"switch{i + 1}", roles=list(_SWITCH_ROLES)))
for i in range(items.get(TroItem.SCANNER, 0)):
lib/tro_flow.py
+15 -1
View File
@@ -179,7 +179,7 @@ from material_flow import (
resolve_input,
)
from kreisel_junctions import Junction, find_junctions
from model import Analysis, DerivedTro, TroDocument
from model import Analysis, DerivedTro, LayoutPart, TroDocument
from parameter import get_float, get_int
from tro_ids import (
DEFAULT_TOLERANCE,
@@ -2210,6 +2210,19 @@ def _tro_label(tro: DerivedTro) -> str:
return _dot_label(lines)
def layout_parts(elements: list[Element]) -> list[LayoutPart]:
"""Alle Bauteile mit Lage fuer die TRO-JSON (``parts``).
Ziel der raeumlichen Zuordnung in merge_hw_layout: Separatoren, Aus-/
Einschleuselemente, Weichen, Scanner, Foerderer (mit Huellbox).
"""
return [
LayoutPart(layout_id=e.teile_id, kind=e.kind, name=e.name,
position=e.position, bbox=e.bbox)
for e in elements if e.position is not None
]
def missing_positions(analysis: Analysis) -> list[str]:
"""TROs/Systemgrenzen ohne Koordinate - mit ihnen ist kein lagerichtiges
Layout moeglich (auch eine Systemgrenze wird als Knoten gezeichnet)."""
@@ -3312,6 +3325,7 @@ def main(argv: list[str] | None = None) -> int:
analysis, source=csv_file.name,
note="Aus dem CSV-Export abgeleitet (lib/tro_flow.py) - die "
"Zeichnung ist nur Ansicht, keine Datenquelle.",
parts=layout_parts(elements),
)
json_target.write_text(document.to_json(), encoding="utf-8")
json_file = json_target