Stabile TRO-IDs: ID-Datei je Anlage, Wiederfinden ueber die Lage
- lib/tro_ids.py: cfg/<csv>_tro_ids.ini lesen/schreiben, TROs ueber die Lage ihrer Separatoren wiederfinden, neue IDs vergeben, Weggefallene melden - tro_flow: Nummerierung entlang des Materialflusses, Schalter --ids, --renumber, --id-tolerance; Bericht der IDs - tro_annotate: liest dieselbe ID-Datei (nur lesend) - Tests tests/test_tro_ids.py, Doku Python_Scripts.md, Konzept (Stufe 1 erledigt) Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Michael Stangl
co-authored by
Claude Opus 5.5
Claude Opus 5.5
parent
ef57a2218e
commit
dd2d98f2ee
+309
@@ -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")
|
||||
Reference in New Issue
Block a user