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

E-Planung -> Layout: e_planning.py, scl_skeleton --eplan, Kapitel Daten

Browse Source 
- doc/Klassenhierarchie.md: Kapitel "Daten" - alle Daten der SCL-Referenzen
  nach Quelle (Layout, E-Planung, Namenskonvention, Parameter)
- doc/E-Planung_Bruecke.md: Bruecke ueber *_positionsdraw.json des
  Kabellaengen-Tools, Befunde an Mubea
- lib/e_planning.py + cfg/e_planning.ini + bin/e_planning.bat/.sh:
  Lagebezug E-Zeichnung -> CSV, Zuordnung BMK -> Layout-Bauteil nach Regeln
  (Abstand, Sollabstand, Nummernregel MB = BG + 1), <csv>_eplan.json + Bericht
- scl_skeleton --eplan: Separator-Sensor und Stopperventil eingesetzt,
  ohne Schalter unveraenderte Ausgabe (Golden-Files gleich)
- tests/test_e_planning.py

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Michael Stangl
2026-10-11 09:30:56 +02:00
co-authored by Claude Opus 5.5
parent 5764df1ac8
commit c736288476
11 changed files with 1683 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/e_planning.py
+715
View File
@@ -0,0 +1,715 @@
# -*- coding: utf-8 -*-
# Absichtlich ohne Shebang: der Windows-py-Launcher wuerde daraus "python3"
# ableiten und scheitert, wenn keine PythonCore-Installation registriert ist.
"""
e_planning.py - Bruecke Layout <-> E-Planung ueber die Symbolpositionen.
Das Layout (CSV) sagt, welche Separatoren, Weichen, Aus-/Einschleuselemente
es gibt und wo; die E-Planung, wie ihre Sensoren und Ventile heissen
(BG0642, MB0643 ...). Die EA-Liste allein kennt keine Topologie. Die
E-Planung wird aber als Layer ueber der Hardwareplanung gezeichnet: jedes
Sensor-/Aktorsymbol liegt neben dem Bauteil, zu dem es gehoert. Das
Kabellaengen-Tool (kabellaengen/lib/getpositions.py) schreibt diese Symbole
mit Position und Attributen nach <zeichnung>_positionsdraw.json.
Hier wird daraus die Zuordnung Layout-Bauteil -> BMK:
1. E-Bauteile einlesen (`sensors` und `schaltschrank_elemente`).
2. Lagebezug E-Zeichnung -> CSV mit lib/dxf_registration.py; Anker sind die
Separator-Sensoren der E-Planung gegen die Separatoren des CSV. Der
gepruefte Bezug wird in SKEL_CFG/dxf_registration.json abgelegt.
3. Regeln aus SKEL_CFG/e_planning.ini anwenden, je Rolle (z. B.
"Separator.sensor"): welche Layout-Art, welche E-Verwendung (VERW),
welcher Hoechstabstand - oder "Nummer = Nummer des Partners + Versatz"
(Ventil MB0643 zum Sensor BG0642). Die Zuordnung ist eindeutig: jedes
E-Bauteil und jede Rolle eines Layout-Bauteils wird hoechstens einmal
vergeben, die kuerzesten Abstaende zuerst.
4. Ergebnis als <csv>_eplan.json (EplanDocument) und Bericht
<csv>_eplan.md nach SKEL_RESULTS.
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.
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,
)
__all__ = [
"EPLAN_SCHEMA",
"EDevice",
"LayoutPoint",
"Rule",
"Assignment",
"Drive",
"EplanDocument",
"load_positionsdraw",
"load_rules",
"assign",
"pair_drives",
]
EPLAN_SCHEMA = "sps_skel-eplan-v1"
RULES_FILE = "e_planning.ini"
_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+)$")
# ---------------------------------------------------------------------------
# Modell
# ---------------------------------------------------------------------------
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 = 150.0
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."""
model_config = ConfigDict(extra="forbid", populate_by_name=True)
schema_id: str = Field(default=EPLAN_SCHEMA, alias="schema")
created: str = ""
csv_file: str = ""
eplan_file: str = ""
transform: Transform | None = None
assignments: list[Assignment] = Field(default_factory=list)
drives: list[Drive] = Field(default_factory=list)
# E-Bauteile im Feld ohne Zuordnung (BMK)
unassigned: list[str] = Field(default_factory=list)
warnings: list[str] = Field(default_factory=list)
def for_layout(self, layout_id: str) -> dict[str, Assignment]:
"""Rollen eines Layout-Bauteils: {"sensor": Assignment, ...}."""
return {
a.role.split(".", 1)[-1]: a
for a in self.assignments if a.layout_id == layout_id
}
def to_json(self) -> str:
return self.model_dump_json(by_alias=True, indent=2) + "\n"
@classmethod
def from_json(cls, text: str) -> "EplanDocument":
return cls.model_validate_json(text)
# ---------------------------------------------------------------------------
# Einlesen
# ---------------------------------------------------------------------------
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", "150")),
))
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 = []
for e in elements:
if e.position is None:
continue
kind = next((k for k in LAYOUT_KINDS
if k.lower() in f"{e.kind} {e.teile_art}".lower()), "")
if kind:
points.append(LayoutPoint(e.teile_id, kind, e.x, e.y, e.name))
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
# ---------------------------------------------------------------------------
def render_report(doc: EplanDocument, rules: list[Rule], points: list[LayoutPoint],
devices: list[EDevice]) -> str:
out = []
add = out.append
add(f"# E-Planung -> Layout: {doc.csv_file}")
add("")
add(f"- E-Planung: `{doc.eplan_file}`")
add(f"- Erzeugt: {doc.created}")
if doc.transform:
add(f"- Lagebezug: {doc.transform.describe()}")
add("")
add("## Zuordnung je Rolle")
add("")
add("| Rolle | Layout-Bauteile | zugeordnet | davon Nummernregel |")
add("|---|---|---|---|")
for r in rules:
n_pts = sum(1 for p in points if _match(p.kind, r.layout))
hits = [a for a in doc.assignments if a.role == r.role]
n_num = sum(1 for a in hits if a.rule == "nummer")
add(f"| {r.role} | {n_pts} | {len(hits)} | {n_num} |")
add("")
add("## Zuordnungen")
add("")
add("Abstand nach dem Lagebezug; \"2.\" = Abstand des naechstbesten Kandidaten "
"(knapp beieinander = pruefen).")
add("")
add("| Layout | Art | Rolle | BMK | VERW | Bezeichnung | Abstand mm | 2. | Regel | Station |")
add("|---|---|---|---|---|---|---|---|---|---|")
for a in doc.assignments:
second = "" if a.runner_up is None else f"{a.runner_up:.0f}"
add(f"| {a.layout_id} | {a.layout_kind} | {a.role.split('.', 1)[-1]} | {a.bmk} "
f"| {a.verw} | {a.bezeichnung} | {a.distance:.0f} | {second} | {a.rule} "
f"| {a.controller}-{a.station} |")
add("")
roles_of: dict[str, set[str]] = {}
for a in doc.assignments:
roles_of.setdefault(a.layout_id, set()).add(a.role)
add("## Layout-Bauteile ohne Zuordnung")
add("")
any_missing = False
for r in rules:
missing = [p for p in points
if _match(p.kind, r.layout) and r.role not in roles_of.get(p.layout_id, set())]
if missing:
any_missing = True
add(f"- **{r.role}**: {', '.join(p.layout_id for p in missing)}")
if not any_missing:
add("- keine")
add("")
by_bmk = {d.bmk: d for d in devices}
add("## E-Bauteile im Feld ohne Zuordnung")
add("")
if doc.unassigned:
add("| BMK | VERW | Bezeichnung | Station |")
add("|---|---|---|---|")
for bmk in doc.unassigned:
d = by_bmk[bmk]
add(f"| {bmk} | {d.verw} | {d.bezeichnung} | {d.controller}-{d.station} |")
else:
add("- keine")
add("")
add("## Antriebe (Motor <-> Motorschutz)")
add("")
if doc.drives:
add("| Motor | Bezeichnung | Motorschutz | Bezeichnung |")
add("|---|---|---|---|")
for dr in doc.drives:
add(f"| {dr.motor} | {dr.bezeichnung} | {dr.protection or '-'} "
f"| {dr.protection_bezeichnung} |")
else:
add("- keine Motoren in der E-Planung")
add("")
if doc.warnings:
add("## Hinweise")
add("")
for w in doc.warnings:
add(f"- {w}")
add("")
return "\n".join(out)
# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------
def build_document(csv_path: Path, eplan_path: Path, rules_path: Path, cfg_dir: Path,
re_register: bool = False
) -> tuple[EplanDocument, list[Rule], list[LayoutPoint], list[EDevice]]:
"""Kompletter Lauf ohne Ausgabe: CSV + positionsdraw -> EplanDocument."""
from material_flow import read_elements # erst hier: Tests ohne CSV brauchen es nicht
warnings: list[str] = []
elements = read_elements(csv_path, warnings)
points = layout_points(elements)
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)
assignments = assign(points, devices, rules, transform)
used = {a.bmk for a in assignments}
unassigned = sorted(d.bmk for d in devices
if not d.cabinet and d.pos is not None and d.bmk not in used)
doc = EplanDocument(
created=datetime.now().strftime("%Y-%m-%d %H:%M"),
csv_file=csv_path.name,
eplan_file=eplan_path.name,
transform=transform,
assignments=assignments,
drives=pair_drives(devices),
unassigned=unassigned,
warnings=warnings,
)
return doc, rules, points, devices
def main(argv: list[str] | None = None) -> int:
from material_flow import env_dir, resolve_input
args = parse_args(argv)
csv_path = resolve_input(args.file)
eplan_path = resolve_input(args.eplan)
cfg_dir = env_dir("SKEL_CFG", "cfg")
rules_path = Path(args.rules) if args.rules else cfg_dir / RULES_FILE
for label, path in (("CSV", csv_path), ("E-Planung", eplan_path),
("Regeln", rules_path)):
if not path.is_file():
print(f"FEHLER: {label} nicht gefunden: {path}", file=sys.stderr)
return 1
try:
doc, rules, points, devices = build_document(
csv_path, eplan_path, rules_path, cfg_dir, args.re_register)
except RegistrationError as exc:
print(f"FEHLER: Lagebezug E-Planung -> CSV nicht nachweisbar: {exc}",
file=sys.stderr)
print(" Passen CSV 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
results = env_dir("SKEL_RESULTS", "results")
results.mkdir(parents=True, exist_ok=True)
json_file = results / f"{csv_path.stem}_eplan.json"
md_file = results / f"{csv_path.stem}_eplan.md"
json_file.write_text(doc.to_json(), encoding="utf-8")
md_file.write_text(render_report(doc, rules, points, devices), encoding="utf-8")
print("")
print("================================================================")
print("E-PLANUNG -> LAYOUT")
print("================================================================")
print(f"CSV = {csv_path}")
print(f"E-Planung = {eplan_path}")
print(f"Lagebezug = {doc.transform.describe() if doc.transform else '-'}")
for r in rules:
n_pts = sum(1 for p in points if _match(p.kind, r.layout))
hits = sum(1 for a in doc.assignments if a.role == r.role)
print(f" {r.role:<26} {hits:4} von {n_pts}")
print(f"Ohne Zuordnung = {len(doc.unassigned)} E-Bauteile im Feld")
print(f"Antriebe = {len(doc.drives)} "
f"({sum(1 for d in doc.drives if d.protection)} mit Motorschutz)")
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="e_planning",
description="Ordnet die Sensoren/Ventile der E-Planung "
"(*_positionsdraw.json des Kabellaengen-Tools) den Bauteilen "
"des Layouts (CSV) zu. Ergebnis: <csv>_eplan.json fuer "
"scl_skeleton --eplan und ein Bericht <csv>_eplan.md.",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="Beispiele:\n"
" e_planning.bat --file mubea.csv --eplan Mubea_positionsdraw.json\n"
" e_planning.bat --file mubea.csv --eplan C:\\pfad\\Mubea_positionsdraw.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("--file", default="export.csv", metavar="NAME",
help="CSV-Export in %%SKEL_DATA%% (oder ein Pfad). "
"Standard: %(default)s")
parser.add_argument("--eplan", required=True, metavar="NAME",
help="*_positionsdraw.json der E-Planung in %%SKEL_DATA%% "
"(oder ein Pfad)")
parser.add_argument("--rules", metavar="DATEI",
help=f"Regeldatei. Standard: %%SKEL_CFG%%\\{RULES_FILE}")
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/scl_context.py
+47 -13
View File
@@ -16,8 +16,11 @@ Schichten:
Namen (Separator-, Weichen-, MainTro-Nummer, Adressen ...) sind ein
:class:`SclName`: Wert plus Merker, ob er noch aus der E-Planung kommen muss.
Heute sind alle Platzhalter mit ``todo=True``; die Namensregeln aus Schritt 4
(cfg/scl_naming.ini) ersetzen dann nur den Wert und loeschen den Merker.
Ohne E-Planung sind alle Platzhalter ``todo=True``. Mit einer Zuordnung aus
lib/e_planning.py (``eplan``) werden Separator-Sensor und Stopperventil der
zugeordneten Separatoren eingesetzt und ``todo`` geloescht; alles andere
bleibt offen. Die Namensregeln aus Schritt 4 (cfg/scl_naming.ini) arbeiten
spaeter genauso: Wert ersetzen, Merker loeschen.
"""
from __future__ import annotations
@@ -25,12 +28,16 @@ from __future__ import annotations
import re
from collections import deque
from datetime import datetime
from typing import TYPE_CHECKING
from pydantic import BaseModel, ConfigDict, Field
from model import TroBase, TroDocument
from tro_catalog import TroItem
if TYPE_CHECKING:
from e_planning import EplanDocument
__all__ = [
"TODO",
"SclName",
@@ -65,6 +72,16 @@ def _offen(platzhalter: str) -> SclName:
return SclName(value=platzhalter, todo=True)
def _bekannt(wert: str) -> SclName:
return SclName(value=wert, todo=False)
# Rolle in e_planning.ini -> wie der Name in der SCL geschrieben wird
# (Sensor als Konstante der DB_Inputs-Sensorliste, Ventil als Variable)
_SENSOR = "cIn{bmk}"
_VENTIL = "{bmk}"
class SclSeparator(BaseModel):
"""Ein stInSeparator<n>-Block."""
@@ -183,6 +200,8 @@ class SclFile(BaseModel):
created: str
start_reason: str
tros: list[SclTro] = Field(default_factory=list)
# Zuordnung zur E-Planung, aus der Namen eingesetzt wurden ("" = keine)
eplan_source: str = ""
# ---------------------------------------------------------------------------
@@ -282,7 +301,21 @@ def _nummer(tro_id: str) -> str:
# Kontext bauen
# ---------------------------------------------------------------------------
def _tro_context(tro: TroBase, alle: dict[str, TroBase]) -> SclTro:
def _separator(i: int, layout: str, eplan: EplanDocument | None) -> SclSeparator:
rollen = eplan.for_layout(layout) if eplan is not None and layout != "?" else {}
sensor = rollen.get("sensor")
stopper = rollen.get("stopper")
return SclSeparator(
n=i,
layout=layout,
no=_offen("cSep_____"),
sensor=_bekannt(_SENSOR.format(bmk=sensor.bmk)) if sensor else _offen("cInBG____"),
stopper=_bekannt(_VENTIL.format(bmk=stopper.bmk)) if stopper else _offen("MB____"),
)
def _tro_context(tro: TroBase, alle: dict[str, TroBase],
eplan: EplanDocument | None = None) -> SclTro:
items = tro.items
n_sep = items.get(TroItem.SEPARATOR, 1)
n_swi = items.get(TroItem.SWITCH, 0)
@@ -291,13 +324,7 @@ def _tro_context(tro: TroBase, alle: dict[str, TroBase]) -> SclTro:
jam_db = '"DB_JamArea____".stJam____'
separators = [
SclSeparator(
n=i,
layout=tro.separators[i - 1] if i <= len(tro.separators) else "?",
no=_offen("cSep_____"),
sensor=_offen("cInBG____"),
stopper=_offen("MB____"),
)
_separator(i, tro.separators[i - 1] if i <= len(tro.separators) else "?", eplan)
for i in range(1, n_sep + 1)
]
switches = [
@@ -371,13 +398,20 @@ def _tro_context(tro: TroBase, alle: dict[str, TroBase]) -> SclTro:
def build_context(document: TroDocument, source: str, start_id: str = "",
created: str | None = None) -> SclFile:
"""TroDocument -> SclFile (TROs in Fliessrichtung)."""
created: str | None = None,
eplan: EplanDocument | None = None) -> SclFile:
"""TroDocument -> SclFile (TROs in Fliessrichtung).
eplan: Zuordnung Layout -> E-Planung (e_planning.py); ohne sie bleiben
alle Namen offen und die Ausgabe ist die bisherige.
"""
alle = {t.tro_id: t for t in document.tros}
geordnet, startgrund = flow_order(document.tros, start_id)
return SclFile(
source=source,
created=created or datetime.now().strftime("%Y-%m-%d %H:%M"),
start_reason=startgrund,
tros=[_tro_context(t, alle) for t in geordnet],
tros=[_tro_context(t, alle, eplan) for t in geordnet],
eplan_source=(f"{eplan.eplan_file} (Zuordnung vom {eplan.created})"
if eplan is not None else ""),
)
lib/scl_skeleton.py
+56 -4
View File
@@ -51,6 +51,7 @@ from jinja2 import (
)
from pydantic import ValidationError
from e_planning import EPLAN_SCHEMA, EplanDocument
from material_flow import env_dir
from model import SCHEMA, TroDocument
from scl_context import TODO, SclFile, SclName, SclTro, build_context
@@ -329,6 +330,12 @@ def render_context(context: SclFile, templates: SclTemplates | None = None) -> s
add("//")
add("// Reihenfolge der REGIONs = Materialfluss.")
add(f"// Anfang bei {context.tros[0].id}: {context.start_reason}.")
if context.eplan_source:
add("//")
add(f"// E-Planung: {context.eplan_source}")
add("// Separator-Sensoren und Stopperventile aus der Zuordnung sind")
add("// eingesetzt (ohne Marker); was die Zuordnung nicht sicher fand,")
add(f"// bleibt '{TODO}'. Details: <csv>_eplan.md (lib/e_planning.py).")
if templates is not None:
mit, ohne = template_usage(context, templates)
add("//")
@@ -372,9 +379,7 @@ def main(argv: list[str] | None = None) -> int:
args = parse_args(argv)
results = env_dir("SKEL_RESULTS", "results")
json_file = Path(args.json)
if not json_file.is_absolute() and json_file.parent == Path("."):
json_file = results / args.json
json_file = _in_results(args.json, results)
if not json_file.is_file():
print(f"FEHLER: JSON nicht gefunden: {json_file}", file=sys.stderr)
print(" Zuerst bin/tro_flow.bat --file <name>.csv --json laufen lassen.",
@@ -412,7 +417,28 @@ def main(argv: list[str] | None = None) -> int:
return 1
templates = SclTemplates(ordner)
context = build_context(document, json_file.name, args.start)
eplan = None
if args.eplan:
eplan_file = _in_results(args.eplan, results)
try:
eplan = EplanDocument.from_json(eplan_file.read_text(encoding="utf-8"))
except OSError as exc:
print(f"FEHLER: E-Planungs-Zuordnung nicht lesbar: {exc}", file=sys.stderr)
print(" Zuerst bin/e_planning.bat --file <name>.csv --eplan "
"<zeichnung>_positionsdraw.json laufen lassen.", file=sys.stderr)
return 1
except ValidationError as exc:
print(f"FEHLER: {eplan_file} passt nicht zum Schema '{EPLAN_SCHEMA}' "
f"({exc.error_count()} Abweichungen).", file=sys.stderr)
return 1
# Eine Zuordnung zu einem anderen Layout setzt falsche Namen ein -
# Separatornummern sind nur innerhalb eines CSV-Exports eindeutig.
if Path(eplan.csv_file).name != Path(document.source).name:
print(f"FEHLER: {eplan_file.name} gehoert zu '{eplan.csv_file}', die "
f"TRO-JSON zu '{document.source}'.", file=sys.stderr)
return 1
context = build_context(document, json_file.name, args.start, eplan=eplan)
mit, ohne = template_usage(context, templates)
if args.strict and ohne:
print(f"FEHLER: kein Template fuer {', '.join(ohne)} (--strict) - "
@@ -440,6 +466,9 @@ def main(argv: list[str] | None = None) -> int:
if templates is not None:
print(f"Templates = {len(mit)} von {len(mit) + len(ohne)} Typen "
f"(generisch: {', '.join(ohne) or '-'})")
if eplan is not None:
print(f"E-Planung = {eplan_file}")
print(f"Eingesetzt = {_eingesetzt(context)} Namen aus der E-Planung")
print(f"Offene Stellen = {offen} (Marker '{TODO}')")
print(f"SCL = {out_file}")
print("================================================================")
@@ -447,6 +476,21 @@ def main(argv: list[str] | None = None) -> int:
return 0
def _in_results(name: str, results: Path) -> Path:
"""Blosser Dateiname -> in SKEL_RESULTS; Pfadangaben bleiben unveraendert."""
path = Path(name)
if not path.is_absolute() and path.parent == Path("."):
return results / name
return path
def _eingesetzt(context: SclFile) -> int:
"""Wie viele Separator-Namen aus der E-Planung kamen."""
return sum(not name.todo
for tro in context.tros for s in tro.separators
for name in (s.sensor, s.stopper))
def _schema_of(path: Path) -> str:
"""Schema-Kennung einer JSON-Datei, nur fuer die Fehlermeldung ("" wenn unlesbar)."""
try:
@@ -465,6 +509,7 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
" scl_skeleton.bat\n"
" scl_skeleton.bat --json export_tro.json --out FB_Main.scl\n"
" scl_skeleton.bat --json mubea_tro.json --templates\n"
" scl_skeleton.bat --json mubea_tro.json --eplan mubea_eplan.json\n"
"\n"
"Exit-Codes:\n"
" 0 ok (auch wenn TROs ohne Template generisch erzeugt wurden)\n"
@@ -500,6 +545,13 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
"Ohne DIR: lib/scl_templates. FB-Typen ohne Template bleiben "
"generisch. Ohne diesen Schalter: bisherige Ausgabe.",
)
parser.add_argument(
"--eplan",
metavar="NAME",
help="Zuordnung zur E-Planung aus e_planning.py (<csv>_eplan.json in "
"%%SKEL_RESULTS%% oder ein Pfad): Separator-Sensoren und "
"Stopperventile werden eingesetzt statt als TODO markiert.",
)
parser.add_argument(
"--strict",
action="store_true",