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

Merge e-planung-bruecke: stabile TRO-IDs, TRO-JSON v4, E-Planung-Bruecke

Browse Source 
Uebernimmt Michael Stangls Zweig (tro_ids, Slots/jamAreas v4, e_planning
--eplan, Konzept- und Datenquellen-Doku) in master, auf dem seitdem die
SCL-Templates fuer alle 7 Typen mit Referenz-Aufrufstelle liegen.

Konfliktloesung:
- lib/scl_context.py: beide Seiten zusammengefuehrt - Michaels _separator
  (E-Planung-Namen) und Staubereiche (_entries/_exit_area, ein JamEntr je
  eigenem Staubereich) plus die Template-Felder (_scanner, _store,
  carriers/data, part_sensor). _jam() fuehrt jetzt auch jam_id. Die beim
  automatischen Zusammenfuehren verlorene Property SclJam.title
  wiederhergestellt.
- Templates 1Sep, 1Sep2Swi, 1Sep_SSCC, Vario, Vario_workStation, PinStore_Auto
  auf v4 nachgezogen: Kommentar "Staubereich J.... von/vor ..." wie im
  Pilot; PinStore-TODO zaehlt Staubereiche statt Nachbarn (mubea: 2 Zulaeufe
  laufen in einen Eingang).
- Golden-Files *_templates neu erzeugt; geaendert sind ausschliesslich
  Staubereichs-Zeilen (Kennung im Kommentar, ein Eingang statt einem je
  Vorgaenger). Generische Golden-Files = Michaels Stand.
- Neuer Test: E-Planung-Namen kommen auch in der Template-Ausgabe an.
- Doku: README Schema v4, SCL_Templates.md Staubereiche und TODO-Zahlen.

Geprueft: 71 Tests gruen; tro_flow --json + scl_skeleton --templates auf
allen 8 CSVs ohne Fehler.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Samer Ayadi
2026-10-14 09:24:20 +02:00
co-authored by Claude Opus 5.5
parent e68ded7e55 0115ef68b7
commit 0c69faf3db
32 changed files with 3631 additions and 324 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/model.py
+145 -5
View File
@@ -21,6 +21,11 @@ Klassen:
Connection eine gerichtete Verbindung zwischen zwei TROs/Stationen
TroCandidate eine im Review gemeldete Stelle, an der ein TRO erwartet
wird, die aber (noch) keiner ist
Slot eine Stelle eines TRO, an die Hardware gehoert (Separator,
Weiche, Scanner, Antrieb) mit den Rollen, die die
Hardwareplanung dort fuellen muss
JamArea ein Staubereich: Eingang genau eines TRO (je Separator
laut Katalog), mit allen Vorgaengern, die dort einfahren
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
@@ -47,22 +52,28 @@ from pydantic import (
model_validator,
)
from tro_catalog import TroDefinition, get_tro
from tro_catalog import TroDefinition, TroItem, get_tro
__all__ = [
"SCHEMA",
"TroBase",
"Slot",
"JamArea",
"DerivedTro",
"Connection",
"TroCandidate",
"Analysis",
"TroDocument",
"derive_slots",
"derive_jam_areas",
]
# 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`.
SCHEMA = "sps_skel-tro-v3"
# `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).
SCHEMA = "sps_skel-tro-v4"
# Gemeinsame Einstellungen: Pruefung bei Erzeugung und Zuweisung, unbekannte
# Schluessel sind ein Fehler (ein umbenannter JSON-Schluessel faellt auf statt
@@ -191,6 +202,50 @@ class TroBase(_Positioned):
return data
class Slot(BaseModel):
"""
Eine Stelle eines TRO, an die Hardware gehoert (Konzept Umbau 1).
``slot`` ist der stabile Schluessel innerhalb des TRO (``separator1``,
``switch1``, ``scanner1``, ``drive``); zusammen mit der TRO-ID adressiert er
die Zuordnung der Hardwareplanung. ``layout_id`` ist die TeileId des
Bauteils im aktuellen Export - nur zur Nachverfolgung, nicht stabil ("",
wenn das Bauteil im Layout nicht bekannt ist). Rollen mit ``?`` sind
optional: fehlt das Geraet, wird der zugehoerige xPlugged*-Schalter FALSE.
"""
model_config = _CONFIG
slot: str
layout_id: str = Field(default="", alias="layoutId")
roles: list[str] = Field(default_factory=list)
class JamArea(_Positioned):
"""
Ein Staubereich: die Warteschlange vor einem Separator, Eingang genau eines
TRO (``tro``, Eingang Nummer ``entry``).
Kennung ``J<Separator>`` (Mubea: Sep1024 -> J1024). Mehrere Vorgaenger
koennen in denselben Staubereich einfahren (Zusammenfuehrung); deren
Ausgang zeigt dann auf diesen Staubereich. Der SCL-Name
``"DB_JamArea<NNNN>".stJam<NNNN>_<k>`` (Standard UH01-UH05) braucht die
Foerderernummer aus der Hardwareplanung und steht deshalb nicht hier.
"""
model_config = _CONFIG
jam_id: str = Field(alias="id")
tro: str
entry: int = 1
separator: str = ""
# tragender Foerderer (erster Host des TRO im Layout)
host: str = ""
predecessors: list[str] = Field(default_factory=list)
note: str = ""
position: tuple[float, float, float] | None = None
class DerivedTro(TroBase):
"""
Ein von tro_flow aus dem CSV abgeleiteter TRO (oder eine Systemgrenze).
@@ -207,6 +262,8 @@ class DerivedTro(TroBase):
# Mittelpunkt der Bauteile, aus denen dieser TRO gebildet wurde
position_from: str = Field(default="", alias="positionFrom")
position_points: int = Field(default=0, alias="positionPoints")
# Stellen fuer die Hardwareplanung (derive_slots); leer bei Systemgrenzen
slots: list[Slot] = Field(default_factory=list)
@property
def connected(self) -> bool:
@@ -288,16 +345,23 @@ class TroDocument(BaseModel):
boundaries: list[DerivedTro] = Field(default_factory=list)
connections: list[Connection] = Field(default_factory=list)
candidates: list[TroCandidate] = Field(default_factory=list)
jam_areas: list[JamArea] = Field(default_factory=list, alias="jamAreas")
findings: list[str] = Field(default_factory=list)
@classmethod
def from_analysis(cls, analysis: Analysis, source: str, note: str = "") -> TroDocument:
"""Dokument aus dem Ergebnis von tro_flow.analyse_tros bilden."""
"""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.
"""
tros = [t.model_copy(update={"slots": derive_slots(t)}) for t in analysis.tros]
return cls(
source=source,
note=note,
tros=analysis.tros,
tros=tros,
boundaries=analysis.boundaries,
jam_areas=derive_jam_areas(tros),
connections=[
Connection(src=src, dst=dst, kind=analysis.edge_kinds.get((src, dst), ""))
for src, dst in analysis.edges
@@ -310,6 +374,10 @@ class TroDocument(BaseModel):
"""JSON-Text einlesen und pruefen (pydantic.ValidationError bei Fehlern)."""
return cls.model_validate_json(text)
def jam_areas_or_derived(self) -> list[JamArea]:
"""jamAreas aus der Datei; bei einer v3-Datei (ohne jamAreas) abgeleitet."""
return self.jam_areas or derive_jam_areas(self.tros)
def to_json(self) -> str:
"""Als JSON-Text mit den Schluesseln der TRO-JSON (id, type, fbType ...)."""
return self.model_dump_json(by_alias=True, indent=2) + "\n"
@@ -319,3 +387,75 @@ class TroDocument(BaseModel):
if tro.tro_id == tro_id:
return tro
return None
# ---------------------------------------------------------------------------
# Ableitungen aus Katalogtyp und Topologie
# ---------------------------------------------------------------------------
# Rollen je Slot-Art (doc/Konzept_Layout_Hardware_SCL.md, Umbau 1); "?" = optional
_SEPARATOR_ROLES = ("sensor", "stopper", "jamSensor?", "partSensor?")
_SWITCH_ROLES = ("valve", "exitSensor?")
_SCANNER_ROLES = ("reader",)
_DRIVE_ROLES = ("motor", "motorProtection", "senCarrInPos", "senFinger",
"senLastPos", "senJam")
def _at(values: list[str], index: int) -> str:
return values[index] if index < len(values) else ""
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.
"""
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)))
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)):
slots.append(Slot(slot=f"scanner{i + 1}", layout_id=_at(tro.scanners, i),
roles=list(_SCANNER_ROLES)))
if items.get(TroItem.VARIO_DRIVE):
slots.append(Slot(slot="drive", layout_id=_at(tro.hosts, 0),
roles=list(_DRIVE_ROLES)))
return slots
def derive_jam_areas(tros: list[DerivedTro]) -> list[JamArea]:
"""Staubereiche: je Separator laut Katalog ein Eingang des TRO.
Kennung J<Separator>; fehlt der Separator im Layout, J<TRO-ID>_<n>. Alle
Vorgaenger fahren in den ersten Eingang ein - bei zwei Eingaengen
(2Sep-Typen) ist die Aufteilung aus dem Layout nicht ableitbar und als
Hinweis vermerkt.
"""
areas: list[JamArea] = []
for tro in tros:
n_entries = max(1, tro.items.get(TroItem.SEPARATOR, 1))
if n_entries > 1 and tro.predecessors:
note = "Aufteilung der Vorgaenger auf die Eingaenge offen"
elif tro.items.get(TroItem.STORAGE_LINE):
# Die Separatoren eines Speichers sitzen auf den Lagerlinien; der
# Eingang liegt davor und hat im Layout keinen eigenen Separator.
note = "Eingang des Speichers - Kennung vorlaeufig nach erstem Linien-Separator"
else:
note = ""
for n in range(1, n_entries + 1):
sep = _at(tro.separators, n - 1)
areas.append(JamArea(
jam_id=f"J{sep}" if sep else f"J{tro.tro_id}_{n}",
tro=tro.tro_id,
entry=n,
separator=sep,
host=_at(tro.hosts, 0),
predecessors=list(tro.predecessors) if n == 1 else [],
note=note,
position=tro.position if n == 1 else None,
))
return areas
lib/scl_context.py
+105 -27
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 model import JamArea, TroBase, TroDocument
from tro_catalog import TroItem
if TYPE_CHECKING:
from e_planning import EplanDocument
__all__ = [
"TODO",
"SclName",
@@ -69,6 +76,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."""
@@ -130,12 +147,19 @@ class SclScanner(BaseModel):
class SclJam(BaseModel):
"""Ein Staubereich am Ein- oder Ausgang (JamEntr<n> / JamExit<n>)."""
"""Ein Staubereich am Ein- oder Ausgang (JamEntr<n> / JamExit<n>).
Ein Eingang ist ein Staubereich des TRO selbst (TroDocument.jamAreas), ein
Ausgang der Eingangs-Staubereich des Nachfolgers.
"""
model_config = _CONFIG
n: int
# Nachbar als Text fuer den Kommentar, SYSn als Station markiert
# Kennung aus der TRO-JSON (J<Separator>); "" am Ausgang zu einer Station
jam_id: str = ""
# Eingang: die Vorgaenger, Ausgang: der Nachfolger - als Text fuer den
# Kommentar, SYSn als Station markiert
neighbour: str
db: SclName
# Stausensor des Ausgangs (stInSenJam<n>); None an Eingaengen
@@ -145,6 +169,11 @@ class SclJam(BaseModel):
carriers: SclName
data: SclName
@property
def title(self) -> str:
"""Fuer den Kommentar: 'Staubereich J0003' (ohne Kennung nur 'Staubereich')."""
return f"Staubereich {self.jam_id}" if self.jam_id else "Staubereich"
class SclSscc(BaseModel):
"""SSCC-Scanner eines 1Sep_SSCC: zweite Kamera, Endmessung, HMI-Anzeige."""
@@ -209,6 +238,11 @@ class SclStore(BaseModel):
# Eine Linie je Speicherbahn im Layout (nicht die 19 aus dem Katalog)
lines: list[SclStoreLine]
@property
def title(self) -> str:
"""Fuer den Kommentar: 'Staubereich J0003' (ohne Kennung nur 'Staubereich')."""
return f"Staubereich {self.jam_id}" if self.jam_id else "Staubereich"
class SclTro(BaseModel):
"""Alles, was eine REGION fuer einen TRO braucht."""
@@ -282,6 +316,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 = ""
# ---------------------------------------------------------------------------
@@ -388,9 +424,10 @@ def _scanner(n: int, suffix: str, interface: str = "cScanner____") -> SclScanner
)
def _jam(n: int, neighbour: str, jam_db: str, sensor: SclName | None = None) -> SclJam:
def _jam(n: int, neighbour: str, jam_db: str, sensor: SclName | None = None,
jam_id: str = "") -> SclJam:
return SclJam(
n=n, neighbour=neighbour, db=_offen(jam_db), sensor=sensor,
n=n, jam_id=jam_id, neighbour=neighbour, db=_offen(jam_db), sensor=sensor,
carriers=_offen(f"{jam_db}.arCarrier"), data=_offen(f"{jam_db}.stData"),
)
@@ -434,23 +471,55 @@ def _store(tro: TroBase) -> SclStore:
)
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____"),
# Teilesensor kennt die E-Planung-Bruecke noch nicht - offen
part_sensor=_offen("cInBG____"),
)
def _entries(tro: TroBase, eigene: list[JamArea],
alle: dict[str, TroBase], jam_db: str) -> list[SclJam]:
"""Ein JamEntr je eigenem Staubereich - nicht je Vorgaenger: mehrere
Vorgaenger fahren in denselben Staubereich ein (Referenzen UH01-UH05)."""
eingaenge = []
for area in eigene:
von = ", ".join(_nachbar(v, alle) for v in area.predecessors) or "-"
if area.note:
von = f"{von} ({area.note})"
eingaenge.append(_jam(area.entry, von, jam_db, jam_id=area.jam_id))
return eingaenge
def _exit_area(tro_id: str, nach: str,
bereiche: dict[str, list[JamArea]]) -> str:
"""Kennung des Staubereichs, in den tro_id beim Nachfolger einfaehrt."""
ziel = bereiche.get(nach, [])
passend = [a for a in ziel if tro_id in a.predecessors] or ziel
return passend[0].jam_id if passend else ""
def _tro_context(tro: TroBase, alle: dict[str, TroBase],
eplan: EplanDocument | None = None,
bereiche: dict[str, list[JamArea]] | None = None) -> SclTro:
items = tro.items
n_sep = items.get(TroItem.SEPARATOR, 1)
n_swi = items.get(TroItem.SWITCH, 0)
n_scan = items.get(TroItem.SCANNER, 0)
tro_id = tro.tro_id
jam_db = '"DB_JamArea____".stJam____'
bereiche = bereiche or {}
separators = [
SclSeparator(
n=i,
layout=tro.separators[i - 1] if i <= len(tro.separators) else "?",
no=_offen("cSep_____"),
sensor=_offen("cInBG____"),
stopper=_offen("MB____"),
part_sensor=_offen("cInBG____"),
)
_separator(i, tro.separators[i - 1] if i <= len(tro.separators) else "?", eplan)
for i in range(1, n_sep + 1)
]
switches = [
@@ -475,15 +544,14 @@ def _tro_context(tro: TroBase, alle: dict[str, TroBase]) -> SclTro:
SclSwitchExit(via=abgang, targets=reachable_via(tro_id, abgang, alle))
for abgang in tro.successors
] if n_swi else []
# Staubereiche: je ankommender/abgehender Verbindung einer. Welche
# DB-Struktur dahintersteht, ist eine Planungsentscheidung - hier steht
# nur, WIE VIELE es sein muessen und zu welchem Nachbarn sie gehoeren.
jam_entries = [
_jam(n, _nachbar(vor, alle), jam_db)
for n, vor in enumerate(tro.predecessors, start=1)
]
# Staubereiche aus der TRO-JSON (jamAreas): Eingaenge sind die eigenen
# (ein JamEntr je eigenem Staubereich, nicht je Vorgaenger), Ausgaenge die
# Eingaenge der Nachfolger. Der DB-Name braucht die Foerderernummer aus der
# Hardwareplanung und bleibt offen.
jam_entries = _entries(tro, bereiche.get(tro_id, []), alle, jam_db)
jam_exits = [
_jam(n, _nachbar(nach, alle), jam_db, sensor=_offen("cInBG____"))
_jam(n, _nachbar(nach, alle), jam_db, sensor=_offen("cInBG____"),
jam_id=_exit_area(tro_id, nach, bereiche))
for n, nach in enumerate(tro.successors, start=2)
]
@@ -541,13 +609,23 @@ 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}
bereiche: dict[str, list[JamArea]] = {}
for area in document.jam_areas_or_derived():
bereiche.setdefault(area.tro, []).append(area)
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, bereiche) for t in geordnet],
eplan_source=(f"{eplan.eplan_file} (Zuordnung vom {eplan.created})"
if eplan is not None else ""),
)
lib/scl_skeleton.py
+58 -6
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
@@ -220,11 +221,11 @@ def emit_call(add, tro: SclTro) -> None:
for jam in tro.jam_entries:
zeilen.append((f"arInOutJamEntr{jam.n} := {jam.db.value}.arCarrier",
offen(jam.db, f"Staubereich von {jam.neighbour}")))
offen(jam.db, f"{jam.title} von {jam.neighbour}")))
zeilen.append((f"stInOutJamEntr{jam.n} := {jam.db.value}.stData", ""))
for jam in tro.jam_exits:
zeilen.append((f"arInOutJamExit{jam.n} := {jam.db.value}.arCarrier",
offen(jam.db, f"Staubereich nach {jam.neighbour}")))
offen(jam.db, f"{jam.title} vor {jam.neighbour}")))
zeilen.append((f"stInOutJamExit{jam.n} := {jam.db.value}.stData", ""))
for s in tro.separators:
@@ -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",
lib/scl_templates/FB_ILS_MTRO_1Sep.scl.j2
+2 -2
View File
@@ -50,11 +50,11 @@ REGION {{ t.region_title }}
{% set _ = z.append(("stInOutMachineState := " ~ t.machine_state.value, m.note(t.machine_state, "Schaltschrank"))) %}
{% set _ = z.append(("stInOutCognexInterface := #stCognexInterfaceDummy", "")) %}
{% for j in t.jam_entries %}
{% set _ = z.append(("arInOutJamEntr" ~ j.n ~ " := " ~ j.carriers.value, m.note(j.carriers, "Staubereich von " ~ j.neighbour))) %}
{% set _ = z.append(("arInOutJamEntr" ~ j.n ~ " := " ~ j.carriers.value, m.note(j.carriers, j.title ~ " von " ~ j.neighbour))) %}
{% set _ = z.append(("stInOutJamEntr" ~ j.n ~ " := " ~ j.data.value, m.note(j.data))) %}
{% endfor %}
{% for j in t.jam_exits %}
{% set _ = z.append(("arInOutJamExit" ~ j.n ~ " := " ~ j.carriers.value, m.note(j.carriers, "Staubereich nach " ~ j.neighbour))) %}
{% set _ = z.append(("arInOutJamExit" ~ j.n ~ " := " ~ j.carriers.value, m.note(j.carriers, j.title ~ " vor " ~ j.neighbour))) %}
{% set _ = z.append(("stInOutJamExit" ~ j.n ~ " := " ~ j.data.value, m.note(j.data))) %}
{% endfor %}
{% for s in t.separators %}
lib/scl_templates/FB_ILS_MTRO_1Sep1Swi.scl.j2
+2 -2
View File
@@ -42,11 +42,11 @@ REGION {{ t.region_title }}
{% set _ = z.append(("stInOutMachineState := " ~ t.machine_state.value, m.note(t.machine_state, "Schaltschrank"))) %}
{% set _ = z.append(("stInOutCognexInterface := #stCognexInterfaceDummy", "")) %}
{% for j in t.jam_entries %}
{% set _ = z.append(("arInOutJamEntr" ~ j.n ~ " := " ~ j.db.value ~ ".arCarrier", m.note(j.db, "Staubereich von " ~ j.neighbour))) %}
{% set _ = z.append(("arInOutJamEntr" ~ j.n ~ " := " ~ j.db.value ~ ".arCarrier", m.note(j.db, j.title ~ " von " ~ j.neighbour))) %}
{% set _ = z.append(("stInOutJamEntr" ~ j.n ~ " := " ~ j.db.value ~ ".stData", m.note(j.db))) %}
{% endfor %}
{% for j in t.jam_exits %}
{% set _ = z.append(("arInOutJamExit" ~ j.n ~ " := " ~ j.db.value ~ ".arCarrier", m.note(j.db, "Staubereich nach " ~ j.neighbour))) %}
{% set _ = z.append(("arInOutJamExit" ~ j.n ~ " := " ~ j.db.value ~ ".arCarrier", m.note(j.db, j.title ~ " vor " ~ j.neighbour))) %}
{% set _ = z.append(("stInOutJamExit" ~ j.n ~ " := " ~ j.db.value ~ ".stData", m.note(j.db))) %}
{% endfor %}
{% set _ = z.append(('stInOutHMI := "DB_Interface_HMI".stTRO.' ~ t.id, "")) %}
lib/scl_templates/FB_ILS_MTRO_1Sep2Swi.scl.j2
+2 -2
View File
@@ -83,11 +83,11 @@ REGION {{ t.region_title }}
{% set _ = z.append(("stInOutCognexInterface := #stCognexInterfaceDummy", "")) %}
{% endif %}
{% for j in t.jam_entries %}
{% set _ = z.append(("arInOutJamEntr" ~ j.n ~ " := " ~ j.carriers.value, m.note(j.carriers, "Staubereich von " ~ j.neighbour))) %}
{% set _ = z.append(("arInOutJamEntr" ~ j.n ~ " := " ~ j.carriers.value, m.note(j.carriers, j.title ~ " von " ~ j.neighbour))) %}
{% set _ = z.append(("stInOutJamEntr" ~ j.n ~ " := " ~ j.data.value, m.note(j.data))) %}
{% endfor %}
{% for j in t.jam_exits %}
{% set _ = z.append(("arInOutJamExit" ~ j.n ~ " := " ~ j.carriers.value, m.note(j.carriers, "Staubereich nach " ~ j.neighbour))) %}
{% set _ = z.append(("arInOutJamExit" ~ j.n ~ " := " ~ j.carriers.value, m.note(j.carriers, j.title ~ " vor " ~ j.neighbour))) %}
{% set _ = z.append(("stInOutJamExit" ~ j.n ~ " := " ~ j.data.value, m.note(j.data))) %}
{% endfor %}
{% for s in t.separators %}
lib/scl_templates/FB_ILS_MTRO_1Sep_SSCC.scl.j2
+2 -2
View File
@@ -77,11 +77,11 @@ REGION {{ t.region_title }}
{% endif %}
{% set _ = z.append(("stInOutCognexInterfaceSSCC := " ~ ss.scanner.interface.value, m.note(ss.scanner.interface))) %}
{% for j in t.jam_entries %}
{% set _ = z.append(("arInOutJamEntr" ~ j.n ~ " := " ~ j.carriers.value, m.note(j.carriers, "Staubereich von " ~ j.neighbour))) %}
{% set _ = z.append(("arInOutJamEntr" ~ j.n ~ " := " ~ j.carriers.value, m.note(j.carriers, j.title ~ " von " ~ j.neighbour))) %}
{% set _ = z.append(("stInOutJamEntr" ~ j.n ~ " := " ~ j.data.value, m.note(j.data))) %}
{% endfor %}
{% for j in t.jam_exits %}
{% set _ = z.append(("arInOutJamExit" ~ j.n ~ " := " ~ j.carriers.value, m.note(j.carriers, "Staubereich nach " ~ j.neighbour))) %}
{% set _ = z.append(("arInOutJamExit" ~ j.n ~ " := " ~ j.carriers.value, m.note(j.carriers, j.title ~ " vor " ~ j.neighbour))) %}
{% set _ = z.append(("stInOutJamExit" ~ j.n ~ " := " ~ j.data.value, m.note(j.data))) %}
{% endfor %}
{% set _ = z.append(('stInOutHMI := "DB_Interface_HMI".stTRO.' ~ t.id, "")) %}
lib/scl_templates/FB_ILS_MTRO_PinStore_Auto.scl.j2
+9 -7
View File
@@ -15,8 +15,10 @@
Sep Store n nJamLength offen (Referenz: 4 bzw. "cStorage111ExitJam<n>"),
Zeiten ueber die Konstanten cTimeSep...StorageLine
arInSep[7..19].stSimu Simulationsmodus der Referenz - nicht geschrieben
JamEntr/JamExit je einer: erster Vorgaenger/Nachfolger. Hat das
Lager mehr Zu-/Ablaeufe, steht ein TODO im Kopf.
JamEntr/JamExit je einer: der eigene Eingangs-Staubereich (mehrere
Zulaeufe laufen dort zusammen, TRO-JSON v4) und der
des ersten Nachfolgers. Hat das Lager mehr
Staubereiche, steht ein TODO im Kopf.
JamBypass Platzhalter wie Referenz
Carr_Glob_Correction nInStation offen (Referenz 1), Traeger vom Eingang
Alle noch mit den SPS-Kollegen zu bestaetigen.
@@ -28,9 +30,9 @@
{% set kamera = "" if sc.tag.todo else " (" ~ (sc.tag.value|upper) ~ ")" %}
REGION {{ t.region_title }}
{{ m.kopf(t) }}
{% if t.predecessors|length > 1 or t.successors|length > 1 %}
// {{ TODO }}: {{ t.predecessors|length }} Zulaeufe / {{ t.successors|length }} Ablaeufe im Layout -
// der FB hat je nur einen Staubereich (JamEntr/JamExit, dazu JamBypass)
{% if t.jam_entries|length > 1 or t.jam_exits|length > 1 %}
// {{ TODO }}: {{ t.jam_entries|length }} Eingangs- / {{ t.jam_exits|length }} Ausgangs-Staubereiche im Layout -
// der FB hat je nur einen (JamEntr/JamExit, dazu JamBypass)
{% endif %}
// Linien (Separatoren aus dem Layout): {{ st.lines|map(attribute="layout")|join(", ") or "-" }}
@@ -103,9 +105,9 @@ REGION {{ t.region_title }}
{% set _ = z.append(("xInConveyorRunOut := " ~ st.run_out.value, m.note(st.run_out, "Foerderer am Ablauf"))) %}
{% set _ = z.append(("stInOutControlUnitCabinet := " ~ t.machine_state.value, m.note(t.machine_state, "Schaltschrank"))) %}
{% set _ = z.append(("arInOutStorageLines := " ~ st.db.value ~ ".arLine", m.note(st.db))) %}
{% set _ = z.append(("arInOutJamEntr := " ~ t.jam_in.carriers.value, m.note(t.jam_in.carriers, "Staubereich von " ~ t.jam_in.neighbour))) %}
{% set _ = z.append(("arInOutJamEntr := " ~ t.jam_in.carriers.value, m.note(t.jam_in.carriers, t.jam_in.title ~ " von " ~ t.jam_in.neighbour))) %}
{% set _ = z.append(("stInOutJamEntr := " ~ t.jam_in.data.value, m.note(t.jam_in.data))) %}
{% set _ = z.append(("arInOutJamExit := " ~ t.jam_out.carriers.value, m.note(t.jam_out.carriers, "Staubereich nach " ~ t.jam_out.neighbour))) %}
{% set _ = z.append(("arInOutJamExit := " ~ t.jam_out.carriers.value, m.note(t.jam_out.carriers, t.jam_out.title ~ " vor " ~ t.jam_out.neighbour))) %}
{% set _ = z.append(("stInOutJamExit := " ~ t.jam_out.data.value, m.note(t.jam_out.data))) %}
{% set _ = z.append(("arInOutJamBypass := #arCarrierPlaceHolder", "")) %}
{% set _ = z.append(("stInOutJamBypass := #stJamDataPlaceHolder", "")) %}
lib/scl_templates/FB_ILS_MTRO_Vario.scl.j2
+2 -2
View File
@@ -69,9 +69,9 @@ REGION {{ t.region_title }}
{% set _ = z.append(("xInErrorQuit := " ~ t.error_quit.value, m.note(t.error_quit, "Quittierung"))) %}
{% set _ = z.append(("xInAutomatic := " ~ t.automatic.value, m.note(t.automatic, "Automatik des Schaltschranks"))) %}
{% set _ = z.append(("xInRelease := " ~ t.release, "")) %}
{% set _ = z.append(("arInOutJamEntr := " ~ t.jam_in.carriers.value, m.note(t.jam_in.carriers, "Staubereich von " ~ t.jam_in.neighbour))) %}
{% set _ = z.append(("arInOutJamEntr := " ~ t.jam_in.carriers.value, m.note(t.jam_in.carriers, t.jam_in.title ~ " von " ~ t.jam_in.neighbour))) %}
{% set _ = z.append(("stInOutJamEntr := " ~ t.jam_in.data.value, m.note(t.jam_in.data))) %}
{% set _ = z.append(("arInOutJamFinger := " ~ t.jam_out.carriers.value, m.note(t.jam_out.carriers, "Staubereich nach " ~ t.jam_out.neighbour))) %}
{% set _ = z.append(("arInOutJamFinger := " ~ t.jam_out.carriers.value, m.note(t.jam_out.carriers, t.jam_out.title ~ " vor " ~ t.jam_out.neighbour))) %}
{% set _ = z.append(("stInOutJamFinger := " ~ t.jam_out.data.value, m.note(t.jam_out.data))) %}
{% set _ = z.append(("stInOutMachineState := " ~ t.machine_state.value, m.note(t.machine_state, "Schaltschrank"))) %}
{% set _ = z.append(("stInOutCognexInterface := #stCognexInterfaceDummy", "")) %}
lib/scl_templates/FB_ILS_MTRO_Vario_workStation.scl.j2
+2 -2
View File
@@ -71,9 +71,9 @@ REGION {{ t.region_title }}
{% set _ = z.append(("xInMotorProtection := " ~ t.motor_protection.value, m.note(t.motor_protection))) %}
{% set _ = z.append(('stInStationSensor := "DB_Inputs".Sensors["' ~ ws.station_sensor.value ~ '"]', m.note(ws.station_sensor, "Arbeitsplatz"))) %}
{% set _ = z.append(("xInReleaseStation := " ~ ws.station_release.value, m.note(ws.station_release, "Freigabe Arbeitsplatz"))) %}
{% set _ = z.append(("arInOutJamEntr := " ~ t.jam_in.carriers.value, m.note(t.jam_in.carriers, "Staubereich von " ~ t.jam_in.neighbour))) %}
{% set _ = z.append(("arInOutJamEntr := " ~ t.jam_in.carriers.value, m.note(t.jam_in.carriers, t.jam_in.title ~ " von " ~ t.jam_in.neighbour))) %}
{% set _ = z.append(("stInOutJamEntr := " ~ t.jam_in.data.value, m.note(t.jam_in.data))) %}
{% set _ = z.append(("arInOutJamFinger := " ~ t.jam_out.carriers.value, m.note(t.jam_out.carriers, "Staubereich nach " ~ t.jam_out.neighbour))) %}
{% set _ = z.append(("arInOutJamFinger := " ~ t.jam_out.carriers.value, m.note(t.jam_out.carriers, t.jam_out.title ~ " vor " ~ t.jam_out.neighbour))) %}
{% set _ = z.append(("stInOutJamFinger := " ~ t.jam_out.data.value, m.note(t.jam_out.data))) %}
{% set _ = z.append(("stInOutMachineState := " ~ t.machine_state.value, m.note(t.machine_state, "Schaltschrank"))) %}
{% set _ = z.append(("stInOutCognexInterface := #stCognexInterfaceDummy", "")) %}
lib/tro_annotate.py
+16 -2
View File
@@ -116,7 +116,9 @@ from tro_flow import (
analyse_tros,
find_sibling_raw_json,
load_sepliste_counts,
resolve_ids_file,
)
from tro_ids import TroIds
# ---------------------------------------------------------------------------
# Konstanten
@@ -844,6 +846,9 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
metavar="MM", help="Markerradius in mm. Standard: %(default)s")
parser.add_argument("--text-height", type=float, default=DEFAULT_TEXT,
metavar="MM", help="Texthoehe in mm. Standard: %(default)s")
parser.add_argument("--ids", metavar="DATEI",
help="ID-Datei der TROs wie bei tro_flow (nur gelesen). "
"Standard: <csv>_tro_ids.ini in %%SKEL_CFG%%.")
parser.add_argument("--offset", metavar="DX,DY",
help="Lagebezug von Hand setzen statt ermitteln.")
parser.add_argument("--re-register", action="store_true",
@@ -907,12 +912,21 @@ def main(argv: list[str] | None = None) -> int:
return 1
graph = build_graph(elements, csv_file.name, read_warnings)
# Derselbe Aufruf wie in tro_flow.main (ohne cfg/tro_overrides.ini, mit
# der Separator-Gegenprobe aus <name>_raw.json), damit die Zeichnung genau
# der Separator-Gegenprobe aus <name>_raw.json und der ID-Datei), damit die Zeichnung genau
# die TROs der TRO-JSON zeigt - sie ist nur Ansicht, keine Datenquelle.
raw_json_file = find_sibling_raw_json(csv_file)
sepliste_counts = (load_sepliste_counts(raw_json_file, read_warnings)
if raw_json_file is not None else {})
analysis = analyse_tros(graph, elements, sepliste_counts=sepliste_counts)
# Dieselbe ID-Datei wie tro_flow, aber nur lesen: neue TROs bekommen
# dieselbe vorlaeufige ID, die tro_flow anhaengen wuerde.
ids_file = resolve_ids_file(args.ids, csv_file)
try:
tro_ids = TroIds.load(ids_file) if ids_file.is_file() else None
except (OSError, ValueError) as exc:
print(f"FEHLER: ID-Datei nicht lesbar: {exc}", file=sys.stderr)
return 1
analysis = analyse_tros(graph, elements, sepliste_counts=sepliste_counts,
tro_ids=tro_ids)
if not analysis.tros:
print(f"FEHLER: keine TROs aus {csv_file} ableitbar.", file=sys.stderr)
return 1
lib/tro_flow.py
+186 -8
View File
@@ -90,6 +90,14 @@ Ausgabe (alles in SKEL_RESULTS)
fuer scl_skeleton.py
<name>_tro_errors.md nur wenn TRO-Fehler gefunden wurden
TRO-IDs (ID-Datei in SKEL_CFG)
------------------------------
Die IDs kommen aus %SKEL_CFG%/<name>_tro_ids.ini (lib/tro_ids.py): jeder TRO
wird ueber die Lage seiner Separatoren wiedergefunden und behaelt seine ID,
auch wenn die SPS-Programmierer sie umbenannt haben. Fehlt die Datei, werden
die TROs entlang des Materialflusses durchnummeriert und die Datei angelegt;
neue TROs werden angehaengt. --renumber nummeriert alles neu.
Aufruf ueber bin/tro_flow.bat bzw. bin/tro_flow.sh.
"""
@@ -126,6 +134,16 @@ from material_flow import (
resolve_input,
)
from model import Analysis, DerivedTro, TroDocument
from tro_ids import (
DEFAULT_TOLERANCE,
TroIds,
assign_ids,
ids_path,
insert_entries,
other_sections,
render_ids,
render_new_entries,
)
from tro_catalog import (
CATALOG_AS_OF,
ITEM_FB,
@@ -655,6 +673,7 @@ def derive_tros(
findings: list[str],
*,
sepliste_counts: dict[str, int] | None = None,
tro_ids: TroIds | None = None,
) -> list[DerivedTro]:
"""TRO-Liste aus den Separatoren und ihren angeschlossenen Objekten bilden.
@@ -666,6 +685,10 @@ def derive_tros(
der CSV abgeleitete Separatorenzahl eines Strecke/Gefaellestrecke-Hosts
nicht mit dieser Zahl ueberein, wird ein Finding angehaengt; die TRO-Daten
selbst bleiben unangetastet (CSV bleibt die massgebliche Quelle).
tro_ids (optional): ID-Datei der Anlage. Ohne sie werden die TROs entlang
des Materialflusses neu durchnummeriert; mit ihr behalten wiedergefundene
TROs ihre ID (Ergebnis in tro_ids.result).
"""
def resolve_host(sep: Element, findings: list[str]) -> str | None:
return _default_resolve_host(sep, findings, graph)
@@ -1156,11 +1179,16 @@ def derive_tros(
tro.reason = f"{tro.reason}; {len(exits)} abgehende Wege -> Weiche"
tro.confidence = "niedrig"
# 5) Stabile, laufende Nummern in Reihenfolge der Host-Knoten
tros.sort(key=lambda t: (t.hosts[0], t.separators[0] if t.separators else ""))
width = max(2, len(str(len(tros))))
for index, tro in enumerate(tros, start=1):
tro.tro_id = f"TRO{index:0{width}d}"
# 5) IDs: Reihenfolge entlang des Materialflusses, wiedergefundene TROs
# behalten ihre ID aus der ID-Datei (siehe lib/tro_ids.py)
tros.sort(key=_flow_key(graph, elements_by_id))
anchors = separator_anchors(elements)
if tro_ids is None:
assign_ids(tros, anchors, [], renumber=True)
else:
tro_ids.result = assign_ids(
tros, anchors, tro_ids.entries,
tolerance=tro_ids.tolerance, renumber=tro_ids.renumber)
# 6) Scanner am Separator: 1Sep_SSCC laesst sich mechanisch nicht erkennen
for tro in tros:
@@ -1174,6 +1202,56 @@ def derive_tros(
return tros
def separator_anchors(elements: list[Element]) -> dict[str, tuple[float, float]]:
"""Separator-TeileId -> (x, y) in mm, Bezugslage fuer die ID-Datei."""
return {e.teile_id: (e.position[0], e.position[1])
for e in elements if e.kind == "Separator" and e.position}
def _flow_rank(graph: Graph) -> dict[str, int]:
"""Knoten -> Rang entlang des Materialflusses (Tiefensuche).
Start an den Materialeingaengen (Beladung/Eingang), danach an jedem noch
nicht erreichten Knoten - eine geschlossene Anlage ohne Station wird so
trotzdem vollstaendig erfasst. Gleichrangige Nachfolger nach Knotennamen.
"""
succ: dict[str, list[str]] = {}
for edge in graph.edges:
succ.setdefault(edge.src, []).append(edge.dst)
starts = sorted(n for n, node in graph.nodes.items() if node.element.is_entry)
rank: dict[str, int] = {}
for start in starts + sorted(graph.nodes):
stack = [start]
while stack:
node = stack.pop()
if node in rank:
continue
rank[node] = len(rank)
stack.extend(n for n in sorted(succ.get(node, []), reverse=True)
if n not in rank)
return rank
def _flow_key(graph: Graph, elements_by_id: dict[str, Element]):
"""Sortierschluessel fuer TROs entlang des Materialflusses.
Erst der Rang des Host-Knotens, dann die Lage des Separators in der
TrackIds-Kette des Hosts (Kreisel: Umlaufrichtung, Strecke: Baureihenfolge).
"""
rank = _flow_rank(graph)
def key(tro: DerivedTro) -> tuple:
host_rank = min((rank.get(h, len(rank)) for h in tro.hosts),
default=len(rank))
sep = tro.separators[0] if tro.separators else ""
host = elements_by_id.get(tro.hosts[0]) if tro.hosts else None
track = host.track_ids if host else []
in_track = track.index(sep) if sep in track else len(track)
return (host_rank, in_track, sep)
return key
# Typnamen fuer Systemgrenz-Knoten (siehe derive_boundaries) - bewusst NICHT
# in TRO_CATALOG: eine Be-/Entladestation (BTMT oder generisches Eingang/
# Ausgang-Symbol) ist kein Steuerobjekt mit eigenem FB-Baustein, nur der
@@ -1781,11 +1859,15 @@ def analyse_tros(
elements: list[Element],
*,
sepliste_counts: dict[str, int] | None = None,
tro_ids: TroIds | None = None,
) -> Analysis:
"""TROs ermitteln, verknuepfen, verorten und gegen die Typdefinition pruefen."""
"""TROs ermitteln, verknuepfen, verorten und gegen die Typdefinition pruefen.
tro_ids: ID-Datei der Anlage (siehe derive_tros); None = neu nummerieren.
"""
analysis = Analysis()
analysis.tros = derive_tros(graph, elements, analysis.findings,
sepliste_counts=sepliste_counts)
sepliste_counts=sepliste_counts, tro_ids=tro_ids)
analysis.boundaries = derive_boundaries(elements, graph)
# Fuer Verortung und Verknuepfung zaehlen TROs und Systemgrenzen zusammen:
# link_tros() registriert eine Kante nur zu einem Knoten, der Host eines
@@ -2638,6 +2720,8 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
" tro_flow.bat --file mubea.csv --tosvg --doc\n"
" tro_flow.bat --file mubea.csv --tosvg --use-cords\n"
" tro_flow.bat --file mubea.csv --json\n"
" tro_flow.bat --file mubea_export.csv --ids mubea_tro_ids.ini --json\n"
" tro_flow.bat --file mubea.csv --renumber\n"
"\n"
"Exit-Codes:\n"
" 0 ok\n"
@@ -2694,9 +2778,93 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
f"Leerstrecken ueber {COORD_MAX_LUECKE} mm zwischen den Knoten "
"(Reihenfolge und Nachbarschaft bleiben erhalten).",
)
parser.add_argument(
"--ids",
metavar="DATEI",
help="ID-Datei der Anlage (Name in %%SKEL_CFG%% oder vollstaendiger "
"Pfad). Standard: <name>_tro_ids.ini in %%SKEL_CFG%%. Fehlt sie, "
"wird sie angelegt.",
)
parser.add_argument(
"--renumber",
action="store_true",
help="Alle TRO-IDs neu entlang des Materialflusses vergeben und die "
"Abschnitt [tro] der ID-Datei neu schreiben (fruehe Projektphase - "
"umbenannte IDs gehen verloren, andere Abschnitte bleiben).",
)
parser.add_argument(
"--id-tolerance",
dest="id_tolerance",
type=float,
default=DEFAULT_TOLERANCE,
metavar="MM",
help="Hoechster Abstand (mm) zwischen Eintrag der ID-Datei und "
"Separator, bis zu dem ein TRO als wiedergefunden gilt. "
"Standard: %(default)s",
)
return parser.parse_args(argv)
def resolve_ids_file(arg: str | None, csv_file: Path) -> Path:
"""Pfad der ID-Datei: --ids (Name in SKEL_CFG oder Pfad), sonst Standard."""
cfg_dir = env_dir("SKEL_CFG", "cfg")
if not arg:
return ids_path(csv_file.stem, cfg_dir)
path = Path(arg)
if path.is_absolute() or path.parent != Path("."):
return path
return cfg_dir / path
def write_ids_file(tro_ids: TroIds, analysis: Analysis, elements: list[Element],
source: str) -> str:
"""ID-Datei anlegen, neu schreiben oder neue TROs anhaengen.
Rueckgabe: kurze Beschreibung fuer den Bericht.
"""
result = tro_ids.result
if result is None:
return "unveraendert"
anchors = separator_anchors(elements)
if result.renumbered:
tro_ids.path.parent.mkdir(parents=True, exist_ok=True)
rest = other_sections(tro_ids.path)
text = render_ids(analysis.tros, anchors, source)
tro_ids.path.write_text(text + (f"\n{rest}\n" if rest else ""),
encoding="utf-8")
return "neu geschrieben" if tro_ids.exists else "angelegt"
if result.new:
insert_entries(tro_ids.path,
render_new_entries(analysis.tros, anchors, result.new, source))
return f"{len(result.new)} neue(r) TRO(s) angehaengt"
return "unveraendert"
def report_ids(tro_ids: TroIds, status: str) -> None:
"""Abschnitt TRO-IDs im Konsolenbericht."""
result = tro_ids.result
if result is None:
return
print(f"TRO-IDs = {tro_ids.path} ({status})")
if result.renumbered:
grund = "--renumber" if tro_ids.renumber else "keine ID-Datei"
print(f" neu entlang des Materialflusses nummeriert "
f"({grund})")
return
print(f" wiedergefunden: {len(result.kept)}, "
f"neu: {len(result.new)}, weggefallen: {len(result.missing)} "
f"(Toleranz {tro_ids.tolerance:g} mm)")
if result.new:
print(f" + neu (vorlaeufige ID, in der ID-Datei umbenennbar): "
f"{', '.join(result.new)}")
for entry in result.missing:
merged = result.merged_into.get(entry.tro_id)
grund = (f"Separator gehoert jetzt zu {merged}" if merged else
f"kein Separator innerhalb {tro_ids.tolerance:g} mm")
print(f" - weggefallen: {entry.tro_id} bei {entry.x:.1f}, {entry.y:.1f} "
f"({grund}; Eintrag bleibt stehen, ID wird nicht wiederverwendet)")
def main(argv: list[str] | None = None) -> int:
args = parse_args(argv)
@@ -2739,11 +2907,20 @@ def main(argv: list[str] | None = None) -> int:
connect_target = results / f"{stem}_tro_connect.ini"
json_target = results / f"{stem}_tro.json"
analysis = analyse_tros(graph, elements, sepliste_counts=sepliste_counts)
try:
tro_ids = TroIds.load(resolve_ids_file(args.ids, csv_file),
tolerance=args.id_tolerance, renumber=args.renumber)
except (OSError, ValueError) as exc:
print(f"FEHLER: ID-Datei nicht lesbar: {exc}", file=sys.stderr)
return 1
analysis = analyse_tros(graph, elements, sepliste_counts=sepliste_counts,
tro_ids=tro_ids)
if not analysis.tros:
print(f"FEHLER: {csv_file} enthaelt keine Separatoren - keine TROs "
f"ableitbar.", file=sys.stderr)
return 1
ids_status = write_ids_file(tro_ids, analysis, elements, csv_file.name)
dot_file = results / f"{stem}_tro_flow.dot"
svg_file = results / f"{stem}_tro_flow.svg"
@@ -2806,6 +2983,7 @@ def main(argv: list[str] | None = None) -> int:
report(analysis, csv_file, dot_file, created_svg, error_file, doc_file,
warnings, engine=engine)
report_ids(tro_ids, ids_status)
if connect_file:
print(f"connect.ini = {connect_file}")
if json_file:
lib/tro_ids.py
+309
View File
@@ -0,0 +1,309 @@
# -*- coding: utf-8 -*-
"""
tro_ids.py - stabile TRO-IDs ueber eine ID-Datei je Anlage.
Die TRO-IDs vergeben die SPS-Programmierer; eine ID muss nur eindeutig sein.
Sobald SCL-Code steht, darf sich eine ID nicht mehr aendern, nur weil das
Layout neu exportiert wurde. Die Layout-TeileId taugt dafuer nicht (sie
wechselt zwischen zwei Exporten derselben Anlage), stabil ist nur die Lage.
Siehe doc/Konzept_Layout_Hardware_SCL.md, "Festlegungen" -> "TRO-IDs".
Die ID-Datei (`%SKEL_CFG%/<csv>_tro_ids.ini`, versioniert, von Hand
umbenennbar) haelt je TRO die ID und die Lage seines Anker-Separators:
[tro]
TRO01 = 1234.5, -567.8 ; 1Sep1Swi | Separator 0003 auf Kreisel2 Bahn L
K0102_1 = 2210.0, 410.0 ; von Hand umbenannt
Ablauf (assign_ids):
- ohne ID-Datei bzw. mit renumber: alle TROs in der uebergebenen Reihenfolge
(entlang des Materialflusses) neu durchnummerieren, TRO01, TRO02, ...
- mit ID-Datei: jeder Eintrag findet den TRO, dessen Separator seiner Lage am
naechsten liegt (innerhalb der Toleranz, je TRO hoechstens ein Eintrag).
Nicht gefundene TROs bekommen die naechste freie Nummer, nicht gefundene
Eintraege bleiben in der Datei stehen und ihre ID wird nicht wiederverwendet.
Das Modul liest und schreibt nur die Datei; Lage und Reihenfolge der TROs
liefert tro_flow.py.
"""
from __future__ import annotations
import configparser
import math
import re
from dataclasses import dataclass, field
from datetime import datetime
from pathlib import Path
__all__ = [
"SECTION",
"DEFAULT_TOLERANCE",
"IdEntry",
"IdAssignment",
"TroIds",
"ids_path",
"load_ids",
"assign_ids",
"render_ids",
"render_new_entries",
"insert_entries",
"other_sections",
]
SECTION = "tro"
ID_FILE_SUFFIX = "_tro_ids.ini"
PREFIX = "TRO"
# Abstand (mm) zwischen Eintrag und Separator, bis zu dem ein TRO als
# wiedergefunden gilt. Separatoren derselben Bahn liegen deutlich weiter
# auseinander; ein verschobenes Bauteil faellt so als "neu" auf.
DEFAULT_TOLERANCE = 100.0
# Eine ID wird zum SCL-Bezeichner (Instanz #<ID>, "DB_TRO_TEST".<ID>).
_ID_PATTERN = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$")
# SYSn sind die Systemgrenzen (tro_flow.derive_boundaries), keine TROs.
_RESERVED = re.compile(r"^SYS\d+$")
_NUMBERED = re.compile(rf"^{PREFIX}(\d+)$")
@dataclass
class IdEntry:
"""Ein Eintrag der ID-Datei: ID und Lage des Anker-Separators (mm)."""
tro_id: str
x: float
y: float
@dataclass
class IdAssignment:
"""Ergebnis von assign_ids, fuer den Bericht."""
renumbered: bool = False
# (ID, Abstand in mm) der wiedergefundenen TROs
kept: list[tuple[str, float]] = field(default_factory=list)
# IDs, die in diesem Lauf neu vergeben wurden (ohne renumbered)
new: list[str] = field(default_factory=list)
# Eintraege der Datei ohne passenden TRO
missing: list[IdEntry] = field(default_factory=list)
# weggefallene ID -> TRO, dessen Separator dort liegt, der aber schon eine
# naehere ID traegt (z. B. zwei TROs zu einem 2Sep2Swi zusammengelegt)
merged_into: dict[str, str] = field(default_factory=dict)
@dataclass
class TroIds:
"""ID-Datei einer Anlage plus Einstellungen fuer einen Lauf.
entries ist leer, wenn die Datei (noch) nicht existiert. result fuellt
tro_flow.derive_tros ueber assign_ids.
"""
path: Path
entries: list[IdEntry] = field(default_factory=list)
exists: bool = False
tolerance: float = DEFAULT_TOLERANCE
renumber: bool = False
result: IdAssignment | None = None
@classmethod
def load(cls, path: Path, *, tolerance: float = DEFAULT_TOLERANCE,
renumber: bool = False) -> "TroIds":
exists = path.is_file()
entries = load_ids(path) if exists else []
return cls(path=path, entries=entries, exists=exists,
tolerance=tolerance, renumber=renumber)
def ids_path(csv_stem: str, cfg_dir: Path) -> Path:
"""Standardpfad der ID-Datei zu einer CSV: <cfg>/<stem>_tro_ids.ini."""
return cfg_dir / f"{csv_stem}{ID_FILE_SUFFIX}"
def _parser() -> configparser.ConfigParser:
parser = configparser.ConfigParser(
inline_comment_prefixes=(";", "#"),
interpolation=None,
strict=True,
)
parser.optionxform = str # IDs nicht in Kleinbuchstaben wandeln
return parser
def load_ids(path: Path) -> list[IdEntry]:
"""ID-Datei lesen. ValueError bei doppelter ID, falscher Lage oder ID."""
parser = _parser()
try:
parser.read(path, encoding="utf-8")
except configparser.Error as exc:
raise ValueError(f"{path}: {exc}") from exc
if not parser.has_section(SECTION):
raise ValueError(f"{path}: Abschnitt [{SECTION}] fehlt")
entries: list[IdEntry] = []
for tro_id, value in parser.items(SECTION):
if not _ID_PATTERN.match(tro_id):
raise ValueError(f"{path}: '{tro_id}' ist kein gueltiger SCL-Bezeichner")
if _RESERVED.match(tro_id):
raise ValueError(f"{path}: '{tro_id}' ist fuer Systemgrenzen reserviert")
parts = [p.strip() for p in value.split(",")]
try:
x, y = (float(p) for p in parts)
except ValueError:
raise ValueError(f"{path}: {tro_id} = '{value}' - erwartet 'x, y' in mm") \
from None
entries.append(IdEntry(tro_id, x, y))
return entries
def _distance(tro, entry: IdEntry, anchors: dict[str, tuple[float, float]]) -> float:
"""Kleinster Abstand eines Separators des TRO zum Eintrag (inf ohne Lage)."""
best = math.inf
for sep in tro.separators:
pos = anchors.get(sep)
if pos is not None:
best = min(best, math.hypot(pos[0] - entry.x, pos[1] - entry.y))
return best
def _width(count: int) -> int:
return max(2, len(str(count)))
def assign_ids(
tros: list,
anchors: dict[str, tuple[float, float]],
entries: list[IdEntry],
*,
tolerance: float = DEFAULT_TOLERANCE,
renumber: bool = False,
) -> IdAssignment:
"""tro.tro_id fuer alle TROs setzen (Reihenfolge = Reihenfolge der Liste).
anchors: Separator-TeileId -> (x, y) in mm.
"""
result = IdAssignment(renumbered=renumber or not entries)
if result.renumbered:
width = _width(len(tros))
for index, tro in enumerate(tros, start=1):
tro.tro_id = f"{PREFIX}{index:0{width}d}"
return result
# Paare nach Abstand; jeder TRO und jeder Eintrag hoechstens einmal
pairs = sorted(
(d, i, j)
for i, tro in enumerate(tros)
for j, entry in enumerate(entries)
if (d := _distance(tro, entry, anchors)) <= tolerance
)
tro_taken: dict[int, tuple[int, float]] = {}
entry_taken: set[int] = set()
for d, i, j in pairs:
if i in tro_taken or j in entry_taken:
continue
tro_taken[i] = (j, d)
entry_taken.add(j)
used = {e.tro_id for e in entries}
numbers = [int(m.group(1)) for e in entries if (m := _NUMBERED.match(e.tro_id))]
widths = [len(m.group(1)) for e in entries if (m := _NUMBERED.match(e.tro_id))]
width = max(widths + [_width(len(tros))])
next_number = max(numbers, default=0) + 1
for i, tro in enumerate(tros):
if i in tro_taken:
j, d = tro_taken[i]
tro.tro_id = entries[j].tro_id
result.kept.append((tro.tro_id, d))
continue
while f"{PREFIX}{next_number:0{width}d}" in used:
next_number += 1
tro.tro_id = f"{PREFIX}{next_number:0{width}d}"
used.add(tro.tro_id)
next_number += 1
result.new.append(tro.tro_id)
result.missing = [e for j, e in enumerate(entries) if j not in entry_taken]
for d, i, j in pairs:
if j not in entry_taken:
result.merged_into.setdefault(entries[j].tro_id, tros[i].tro_id)
return result
def _anchor(tro, anchors: dict[str, tuple[float, float]]) -> tuple[float, float] | None:
for sep in tro.separators:
if sep in anchors:
return anchors[sep]
return None
def _entry_lines(tros: list, anchors: dict[str, tuple[float, float]]) -> list[str]:
lines = []
for tro in tros:
pos = _anchor(tro, anchors)
if pos is None:
continue
seps = ", ".join(tro.separators)
lines.append(f"{tro.tro_id} = {pos[0]:.1f}, {pos[1]:.1f}"
f" ; {tro.type_name} | {tro.label} | Sep {seps}")
return lines
def render_ids(tros: list, anchors: dict[str, tuple[float, float]], source: str) -> str:
"""Komplette ID-Datei (erster Lauf oder renumber)."""
head = [
"; ============================================================",
f"; TRO-IDs zu {source} - erzeugt von lib/tro_flow.py "
f"am {datetime.now():%Y-%m-%d}",
";",
"; Je TRO: ID = x, y (mm) seines Anker-Separators im CSV-Koordinatensystem.",
"; Die IDs duerfen umbenannt werden (eindeutig, gueltiger SCL-Bezeichner);",
"; tro_flow findet jeden TRO ueber die Lage wieder. Neue TROs werden",
"; unten angehaengt, weggefallene bleiben stehen und werden gemeldet.",
"; Alles neu nummerieren: tro_flow --renumber (ueberschreibt diese Datei).",
"; Siehe doc/Konzept_Layout_Hardware_SCL.md, Abschnitt TRO-IDs.",
"; ============================================================",
"",
f"[{SECTION}]",
]
return "\n".join(head + _entry_lines(tros, anchors)) + "\n"
def render_new_entries(tros: list, anchors: dict[str, tuple[float, float]],
new_ids: list[str], source: str) -> str:
"""Anhang fuer neu gefundene TROs an eine bestehende ID-Datei."""
neu = [t for t in tros if t.tro_id in set(new_ids)]
head = [f"; neu aus {source} am {datetime.now():%Y-%m-%d}"]
return "\n".join(head + _entry_lines(neu, anchors)) + "\n"
def other_sections(path: Path) -> str:
"""Alle Abschnitte ausser [tro] als Text (bleiben bei renumber erhalten)."""
if not path.is_file():
return ""
header = re.compile(r"^\s*\[(.+)\]\s*$")
keep: list[str] = []
current = None
for line in path.read_text(encoding="utf-8").splitlines():
if m := header.match(line):
current = m.group(1).strip()
if current is not None and current != SECTION:
keep.append(line)
return "\n".join(keep).strip("\n")
def insert_entries(path: Path, block: str) -> None:
"""block am Ende des Abschnitts [tro] einfuegen, Rest der Datei unveraendert."""
lines = path.read_text(encoding="utf-8").splitlines()
header = re.compile(r"^\s*\[(.+)\]\s*$")
start = next(i for i, line in enumerate(lines)
if (m := header.match(line)) and m.group(1).strip() == SECTION)
end = next((i for i in range(start + 1, len(lines)) if header.match(lines[i])),
len(lines))
# vor Leerzeilen am Abschnittsende einfuegen
while end > start + 1 and not lines[end - 1].strip():
end -= 1
lines[end:end] = block.rstrip("\n").splitlines()
path.write_text("\n".join(lines) + "\n", encoding="utf-8")