Schoenenberger_Systeme_GmbH/sps_skel
8
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

tro_annotate --anlage: Staubereiche in der Zeichnung, --read-jams zurueck

Browse Source 
Mit --anlage <csv>_anlage.json zeichnet tro_annotate je Staubereich einen
Block JAM_AREA (Layer TRO_JAM): Rechteck mit dem Namen Jam<NNNN>_<k> wie in
der SCL, jam_versatz vor dem Separator Richtung Vorgaenger. Attribute
CONVEYOR, NUMBER, TRO + ENTRY (stabiler Schluessel), AREA, AUTO. Doppelklick
oeffnet den Dialog JAM_EDIT (dxfmakros).

--read-jams uebernimmt im Dialog geaenderte Nummern aus der beschrifteten
Zeichnung nach cfg/<csv>_jam_names.ini (nur Abweichungen von AUTO);
scl_skeleton wendet die Datei mit der Anlagen-JSON an (scl_context:
jam_numbers + Korrekturen, load/save_jam_overrides).

Dematic: 25 Staubereiche gezeichnet; Rundreise TRO20 -> Jam0054_9 bis in die
SCL geprueft. Gemeinsame Zeichnung mit --offset 0,0 beschriften (automatischer
Lagebezug waehlte dy +360 mm).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Samer Ayadi
2026-10-23 12:47:11 +02:00
parent 404d500271
commit e5d1fce41f
7 changed files with 540 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/tro_annotate.py
+256 -6
View File
@@ -85,6 +85,7 @@ from __future__ import annotations
import argparse
import collections
import configparser
import csv as csv_module
import fnmatch
import json
@@ -120,6 +121,17 @@ from tro_flow import (
resolve_ids_file,
)
from tro_ids import TroIds
from model import ANLAGE_SCHEMA, AnlageDocument, JamArea
from scl_context import (
JAM_NAMES_SUFFIX,
HwSource,
flow_order,
jam_key,
jam_label,
jam_numbers,
load_jam_overrides,
save_jam_overrides,
)
# ---------------------------------------------------------------------------
# Konstanten
@@ -131,12 +143,18 @@ BLOCK_PREFIX = "TRO_SYM_"
# Verbindungsblock. Der Name beginnt mit CONNECTION_, damit der Dispatcher
# SSG_BLOCKEDIT in dxfmakros ihn ueber "CONNECTION_*" erkennt.
CONNECTION_BLOCK = "CONNECTION_ARROW"
# Staubereich (--anlage). Der Name beginnt mit JAM_, damit SSG_BLOCKEDIT ihn
# ueber "JAM_*" an den Dialog JAM_EDIT (dxfmakros/Lisp/Jam_Edit.lsp) gibt.
JAM_BLOCK = "JAM_AREA"
# Anzeige eines Staubereichs ohne Foerderer (Nummer nicht ableitbar)
JAM_OFFEN = "Jam????_?"
# Zeichnungsgroessen und Restfehler-Grenze des Lagebezugs aus cfg/parameter.ini,
# Abschnitt [tro_annotate] (lib/parameter.py)
DEFAULT_MARKER = get_float("tro_annotate", "default_marker")
DEFAULT_TEXT = get_float("tro_annotate", "default_text")
DEFAULT_MAX_RESIDUAL = get_float("tro_annotate", "default_max_residual")
JAM_VERSATZ = get_float("tro_annotate", "jam_versatz")
def hex_to_rgb(value: str) -> tuple[int, int, int]:
@@ -335,6 +353,7 @@ def build_layers(doc, prefix: str) -> dict[str, str]:
("weiche", "#cc8800"),
("eingang", STATION_FARBE[BOUNDARY_ENTRY]),
("ausgang", STATION_FARBE[BOUNDARY_EXIT]),
("jam", "#7030a0"),
("legend", "#808080")):
name = f"{prefix}{key.upper()}"
names[key] = name
@@ -702,6 +721,150 @@ def draw_connections(
return drawn
def build_jam_symbol(doc, text: float) -> str:
"""
Blockdefinition fuer einen Staubereich: Rechteck, der Name mittig darin.
Attribute:
NAME sichtbar - Jam0054_3 (Foerderer + laufende Nummer)
CONVEYOR unsichtbar - Foerderer NNNN (Kreisel-/Vario-Motor), im Dialog
NUMBER unsichtbar - laufende Nummer k auf dem Foerderer, im Dialog
TRO unsichtbar - TRO, in den der Staubereich fuehrt (TRO17)
ENTRY unsichtbar - Eingang des TRO (1, bei 2Sep-Typen auch 2)
AREA unsichtbar - Kennung der TRO-JSON (J0039)
AUTO unsichtbar - abgeleitete Nummer NNNN_k; --read-jams schreibt
nur, was davon abweicht
TRO + ENTRY ist der stabile Schluessel (scl_context.jam_key).
"""
if JAM_BLOCK in doc.blocks:
return JAM_BLOCK
blk = doc.blocks.new(name=JAM_BLOCK)
w, h = text * 4.2, text * 1.2
blk.add_lwpolyline([(-w, -h), (w, -h), (w, h), (-w, h)], close=True,
dxfattribs={"color": 256})
name = blk.add_attdef("NAME", insert=(0, 0), height=text,
dxfattribs={"style": "Standard", "color": 256,
"text": JAM_OFFEN})
name.set_placement((0, 0), align=TextEntityAlignment.MIDDLE_CENTER)
offset = -h - text
for tag_name in ("CONVEYOR", "NUMBER", "TRO", "ENTRY", "AREA", "AUTO"):
blk.add_attdef(tag_name, insert=(-w, offset), height=text * 0.7,
dxfattribs={"style": "Standard", "color": 256, "text": "",
"invisible": 1})
offset -= text
return JAM_BLOCK
def _jam_position(area: JamArea, anlage: AnlageDocument) -> tuple[float, float] | None:
"""Lage eines Staubereichs (CSV-Koordinaten): vor seinem Separator.
Ausgangspunkt ist der Separator (sonst die TRO-Position); von dort
JAM_VERSATZ in Richtung des ersten Vorgaenger-TRO - gegen den Fluss, wo
die Traeger warten. Liegt der Vorgaenger naeher, nur die halbe Strecke.
"""
layout = anlage.layout
part = layout.part(area.separator) if area.separator else None
if part is not None and part.position is not None:
x, y = part.position[0], part.position[1]
elif area.position is not None:
x, y = area.position[0], area.position[1]
else:
return None
for vor in area.predecessors:
tro = layout.by_id(vor)
if tro is None or not tro.position:
continue
vx, vy = tro.position[0] - x, tro.position[1] - y
dist = math.hypot(vx, vy)
if dist > 1e-6:
schritt = min(JAM_VERSATZ, dist / 2)
return x + vx / dist * schritt, y + vy / dist * schritt
return x, y
def place_jams(
msp, doc, anlage: AnlageDocument, overrides: dict[str, tuple[str, int]],
transform: Transform, layers: dict[str, str], text: float, warnings: list[str],
) -> int:
"""
Je Staubereich der Anlagen-JSON einen JAM_AREA-Block einfuegen.
Die Nummern sind dieselben wie in der SCL (scl_context.jam_numbers, dann
die Korrekturen aus <csv>_jam_names.ini). Ein Staubereich ohne Foerderer
zeigt Jam????_? - im Dialog nachtragen.
"""
layout = anlage.layout
geordnet, _grund = flow_order(layout.tros)
automatisch = jam_numbers(layout, geordnet, HwSource(anlage=anlage))
block = build_jam_symbol(doc, text)
gezeichnet = 0
for area in layout.jam_areas_or_derived():
lage = _jam_position(area, anlage)
if lage is None:
warnings.append(f"Staubereich {area.jam_id} ({area.tro}): keine Lage - "
f"nicht gezeichnet")
continue
auto = automatisch.get(area.jam_id)
nr, k = overrides.get(jam_key(area), auto or ("", 0))
ref = msp.add_blockref(block, transform.to_dxf(*lage),
dxfattribs={"layer": layers["jam"]})
ref.add_auto_attribs({
"NAME": jam_label(nr, k) if nr else JAM_OFFEN,
"CONVEYOR": nr,
"NUMBER": str(k) if nr else "",
"TRO": area.tro,
"ENTRY": str(area.entry),
"AREA": area.jam_id,
"AUTO": f"{auto[0]}_{auto[1]}" if auto else "",
})
tag(ref, area.tro)
gezeichnet += 1
return gezeichnet
def read_jams(dxf_file: Path, ini_file: Path) -> tuple[dict[str, tuple[str, int]], list[str]]:
"""Im Dialog JAM_EDIT geaenderte Staubereichs-Nummern uebernehmen.
Liest die JAM_AREA-Bloecke der beschrifteten Zeichnung. Weicht
CONVEYOR_NUMBER von AUTO ab, kommt der Staubereich in die Korrekturdatei
``<csv>_jam_names.ini``; stimmt er (wieder) mit AUTO ueberein, faellt sein
Eintrag heraus. Eintraege zu Staubereichen, die nicht in der Zeichnung
stehen, bleiben. Rueckgabe: (Korrekturen danach, Meldungen).
"""
meldungen: list[str] = []
overrides = load_jam_overrides(ini_file)
doc = ezdxf.readfile(str(dxf_file))
namen: dict[str, str] = {}
for ref in doc.modelspace().query(f'INSERT[name=="{JAM_BLOCK}"]'):
a = {att.dxf.tag.upper(): (att.dxf.text or "").strip() for att in ref.attribs}
key = f"{a.get('TRO', '')}.{a.get('ENTRY', '')}"
nr, k = a.get("CONVEYOR", ""), a.get("NUMBER", "")
if not a.get("TRO") or not a.get("ENTRY"):
meldungen.append(f"Staubereich ohne TRO/ENTRY ({a.get('NAME', '?')}) - uebersprungen")
continue
if not nr and not k:
continue
if not (nr.isdigit() and len(nr) == 4 and k.isdigit() and int(k) > 0):
meldungen.append(f"{key} ({a.get('AREA', '')}): Foerderer '{nr}' / Nummer '{k}' "
f"ungueltig (erwartet NNNN und k >= 1) - uebersprungen")
continue
wert = f"{nr}_{int(k)}"
if wert in namen.values():
andere = next(s for s, w in namen.items() if w == wert)
meldungen.append(f"{key}: Jam{wert} ist schon {andere} - doppelt vergeben")
namen[key] = wert
alt = overrides.get(key)
if wert == a.get("AUTO", ""):
if alt is not None:
del overrides[key]
meldungen.append(f"{key}: wieder abgeleitet (Jam{wert}) - Korrektur entfernt")
elif alt != (nr, int(k)):
overrides[key] = (nr, int(k))
meldungen.append(f"{key}: Jam{wert} (abgeleitet: Jam{a.get('AUTO') or '-'})")
save_jam_overrides(ini_file, overrides)
return overrides, meldungen
def draw_legend(
msp, analysis: Analysis, placed: dict[str, tuple[float, float]],
layers: dict[str, str], marker: float, text: float, source: str,
@@ -790,6 +953,9 @@ def report(
print(f"TROs = {len(analysis.tros)}, beschriftet: {counts['marker']}")
print(f"Verbindungen = {counts['flow']}"
+ (" (Block CONNECTION_ARROW mit FROM_UID/TO_UID)" if counts["flow"] else ""))
if counts.get("jams"):
print(f"Staubereiche = {counts['jams']} (Block {JAM_BLOCK}, Doppelklick -> "
f"JAM_EDIT; uebernehmen mit --read-jams)")
if counts.get("first_id"):
print(f"Interne IDs ab = {format_internal_id(counts['first_id'])} "
f"(ID-Raum von SSG_LIB, siehe IDSCHECK)")
@@ -820,6 +986,9 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
epilog="Beispiele:\n"
" tro_annotate.bat --file mubea.csv --dxf 500573_60_1.dxf --check\n"
" tro_annotate.bat --file mubea.csv --dxf 500573_60_1.dxf --flow --fb\n"
" tro_annotate.bat --file dematic_merge.csv --dxf dematic_merge..dxf --flow "
"--anlage dematic_merge_anlage.json\n"
" tro_annotate.bat --file dematic_merge.csv --read-jams\n"
"\n"
"Exit-Codes:\n"
" 0 ok\n"
@@ -855,6 +1024,17 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
parser.add_argument("--re-register", action="store_true",
help="Lagebezug neu ermitteln, auch wenn in %%SKEL_CFG%% "
"einer abgelegt ist.")
parser.add_argument("--anlage", metavar="NAME",
help="Anlagen-JSON (<csv>_anlage.json aus merge_hw_layout, in "
"%%SKEL_RESULTS%% oder ein Pfad): zusaetzlich die "
"Staubereiche als JAM_AREA-Bloecke einzeichnen "
"(Jam<NNNN>_<k>, Doppelklick -> Dialog JAM_EDIT).")
parser.add_argument("--read-jams", nargs="?", const="", default=None, metavar="DXF",
help="Im Dialog JAM_EDIT geaenderte Staubereichs-Nummern aus der "
"beschrifteten Zeichnung uebernehmen (Standard: "
"<csv>_annotated.dxf in %%SKEL_RESULTS%%) und nach "
"%%SKEL_CFG%%\\<csv>_jam_names.ini schreiben, dann beenden. "
"scl_skeleton liest die Datei.")
parser.add_argument("--check", action="store_true",
help="Nur pruefen und berichten, keine Datei schreiben.")
parser.add_argument("--emit-lisp", action="store_true",
@@ -863,6 +1043,38 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
return parser.parse_args(argv)
def _main_read_jams(args, csv_file: Path) -> int:
"""--read-jams: Nummern aus der beschrifteten Zeichnung -> <csv>_jam_names.ini."""
results = env_dir("SKEL_RESULTS", "results")
name = args.read_jams or args.out or f"{csv_file.stem}_annotated.dxf"
dxf_file = Path(name)
if not dxf_file.is_absolute() and dxf_file.parent == Path("."):
dxf_file = results / name
if not dxf_file.is_file():
print(f"FEHLER: beschriftete Zeichnung nicht gefunden: {dxf_file}", file=sys.stderr)
print(" In BricsCAD als DXF speichern (Dateiname wie beim Beschriften).",
file=sys.stderr)
return 1
ini_file = env_dir("SKEL_CFG", "cfg") / f"{csv_file.stem}{JAM_NAMES_SUFFIX}"
try:
overrides, meldungen = read_jams(dxf_file, ini_file)
except (OSError, ValueError, configparser.Error, ezdxf.DXFError) as exc:
print(f"FEHLER: {exc}", file=sys.stderr)
return 1
print("")
print("================================================================")
print("STAUBEREICHE UEBERNEHMEN")
print("================================================================")
print(f"Zeichnung = {dxf_file}")
print(f"Korrekturen = {len(overrides)} -> {ini_file if overrides else '- (keine, Datei entfernt)'}")
for m in meldungen:
print(f" {m}")
print("Danach: scl_skeleton mit der Anlagen-JSON neu laufen lassen.")
print("================================================================")
print("")
return 0
def main(argv: list[str] | None = None) -> int:
args = parse_args(argv)
warnings: list[str] = []
@@ -875,16 +1087,19 @@ def main(argv: list[str] | None = None) -> int:
"SSG_LIB-Module (dxfmakros/Lisp), damit TRO_EDIT sie findet.")
return 0
if not args.dxf:
print("FEHLER: --dxf fehlt (nur --emit-lisp geht ohne Zeichnung).",
file=sys.stderr)
return 1
csv_file = resolve_input(args.file)
if csv_file.suffix.lower() != ".csv":
print(f"FEHLER: --file erwartet eine CSV-Datei, keine '{csv_file.suffix}'-Datei: "
f"{csv_file}", file=sys.stderr)
return 1
if args.read_jams is not None:
return _main_read_jams(args, csv_file)
if not args.dxf:
print("FEHLER: --dxf fehlt (nur --emit-lisp und --read-jams gehen ohne "
"Zeichnung).", file=sys.stderr)
return 1
if not csv_file.is_file():
print(f"FEHLER: CSV nicht gefunden: {csv_file}", file=sys.stderr)
return 1
@@ -904,6 +1119,29 @@ def main(argv: list[str] | None = None) -> int:
cfg_dir = env_dir("SKEL_CFG", "cfg")
results.mkdir(parents=True, exist_ok=True)
# Anlagen-JSON fuer die Staubereiche (optional)
anlage = None
jam_overrides: dict[str, tuple[str, int]] = {}
if args.anlage:
anlage_file = Path(args.anlage)
if not anlage_file.is_absolute() and anlage_file.parent == Path("."):
anlage_file = results / args.anlage
try:
text = anlage_file.read_text(encoding="utf-8")
if json.loads(text).get("schema") != ANLAGE_SCHEMA:
raise ValueError(f"kein Schema '{ANLAGE_SCHEMA}' - erwartet wird die "
f"Anlagen-JSON aus merge_hw_layout")
anlage = AnlageDocument.from_json(text)
jam_overrides = load_jam_overrides(
cfg_dir / f"{csv_file.stem}{JAM_NAMES_SUFFIX}")
except (OSError, ValueError, configparser.Error) as exc:
print(f"FEHLER: Anlagen-JSON {anlage_file} nicht lesbar: {exc}", file=sys.stderr)
return 1
if Path(anlage.layout.source).name != csv_file.name:
print(f"FEHLER: {anlage_file.name} gehoert zu '{anlage.layout.source}', "
f"nicht zu '{csv_file.name}'.", file=sys.stderr)
return 1
# TROs aus dem CSV
read_warnings: list[str] = []
try:
@@ -955,7 +1193,16 @@ def main(argv: list[str] | None = None) -> int:
return 4
counts = {"marker": 0, "flow": 0, "removed": 0, "layers": 0, "blocks": 0,
"first_id": 0}
"first_id": 0, "jams": 0}
if anlage is not None:
aus_csv = {t.tro_id for t in analysis.tros}
aus_json = {t.tro_id for t in anlage.layout.tros}
if aus_csv != aus_json:
warnings.append(
f"TROs der Zeichnung und der Anlagen-JSON weichen ab (nur CSV: "
f"{', '.join(sorted(aus_csv - aus_json)) or '-'}; nur JSON: "
f"{', '.join(sorted(aus_json - aus_csv)) or '-'}) - tro_flow --json und "
f"merge_hw_layout neu laufen lassen")
if args.check:
counts["marker"] = sum(1 for t in analysis.tros if t.position)
@@ -986,6 +1233,9 @@ def main(argv: list[str] | None = None) -> int:
msp, doc, analysis, {**placed, **stationen}, internal, layers,
args.marker_size, args.text_height, next_id=next_id + len(placed),
)
if anlage is not None:
counts["jams"] = place_jams(msp, doc, anlage, jam_overrides, transform, layers,
args.text_height, warnings)
if args.legend:
draw_legend(msp, analysis, placed, layers, args.marker_size,
args.text_height, csv_file.name)