latest changes
This commit is contained in:
+12
-4
@@ -69,11 +69,19 @@ __all__ = [
|
||||
"registration_path",
|
||||
]
|
||||
|
||||
# Rollen fuer den Lagebezug: CSV-Objektart -> Blocknamen im DXF.
|
||||
# Erweiterbar, wenn eine Zeichnung andere Bloecke verwendet.
|
||||
# Rollen fuer den Lagebezug: CSV-Objektart -> Blocknamen im DXF, als Muster
|
||||
# (fnmatch, Gross-/Kleinschreibung egal). Es gibt zwei Namenswelten, und beide
|
||||
# kommen im Projekt vor:
|
||||
# - Original-Kundenzeichnung (z.B. 500573_60_1.dxf): S-SP / S-LP / SCAN
|
||||
# - von den dxfmakros-Makros erzeugte Zeichnung (z.B. Mubea.dxf): die flachen
|
||||
# Blocknamen mit Dimensionssuffix, also Separator_SP_2D/_3D, Scanner_2D/_3D
|
||||
# Bewusst NICHT aufgenommen: Staustrecke_Separator_SP_*. Das ist ein Streckenmodul
|
||||
# mit eingebautem Separator - sein Einfuegepunkt ist der Modulursprung, nicht der
|
||||
# Separator, was einen zweiten, um genau diesen Versatz verschobenen Cluster in
|
||||
# die Abstimmung braechte.
|
||||
ANCHOR_BLOCKS: dict[str, tuple[str, ...]] = {
|
||||
"Separator": ("S-SP", "S-LP"),
|
||||
"Scanner": ("SCAN",),
|
||||
"Separator": ("S-SP", "S-LP", "Separator_SP*"),
|
||||
"Scanner": ("SCAN", "Scanner_*"),
|
||||
}
|
||||
|
||||
# Rastergroesse der Abstimmung in mm
|
||||
|
||||
+472
-10
@@ -77,6 +77,35 @@ CIRCLE_KINDS = ("Kreisel",)
|
||||
# Objektarten, die per "Zuordnung" an einem Knoten haengen (kein eigener Knoten)
|
||||
DEVICE_KINDS = ("Separator", "Scanner")
|
||||
|
||||
# Ein-/Ausgabestationen (BTMT Be-/Entladung) = Materialquelle/-senke, d.h. die
|
||||
# Ein-/Ausgangspunkte des Materialflusses. Sie werden in sps_skel erkannt;
|
||||
# dxfmakros exportiert sie derzeit NICHT (Blockmuster fehlt), die Erkennung
|
||||
# greift also, sobald eine passende Zeile im CSV auftaucht - erkannt an
|
||||
# TeileArt/Bezeichnung: "Beladung"/"BTMT-Beladung" -> Eingang, "Entladung"/
|
||||
# "SC_Entladung"/"Abwurf" -> Ausgang.
|
||||
ENTRY_KINDS = ("Beladung",) # BTMT-Beladung -> Materialeingang (Quelle)
|
||||
EXIT_KINDS = ("Entladung",) # SC_Entladung -> Materialausgang (Senke)
|
||||
STATION_KINDS = ENTRY_KINDS + EXIT_KINDS
|
||||
|
||||
# Schluesselwoerter zur Erkennung der Stationen (Kleinschreibung, Teilstring)
|
||||
STATION_PATTERNS = (
|
||||
("Beladung", ("beladung", "beladestation", "btmt-beladung", "btmt beladung")),
|
||||
("Entladung", ("entladung", "entladestation", "sc_entladung", "abwurf")),
|
||||
)
|
||||
|
||||
# Transfer-/Weichenelemente aus dem erweiterten Export. Sie bilden KEINE eigenen
|
||||
# Knoten, liefern aber die *gerichtete* Verbindung zwischen den Transportobjekten:
|
||||
# Ausschleuselement = Kreisel -> Strecke (am Streckenanfang, Material verlaesst Kreisel)
|
||||
# Einschleuselement = Strecke -> Kreisel (am Streckenende, Material laeuft in Kreisel)
|
||||
# Weiche = Uebergang zwischen zwei Kreiseln
|
||||
# Merkmale: {"Kreisel","Strecke","Streckenende"} bzw. {"Kreisel A","Kreisel B"}.
|
||||
TRANSFER_KINDS = ("Einschleus", "Ausschleus", "Weiche")
|
||||
TRANSFER_PATTERNS = (
|
||||
("Ausschleus", ("ausschleus",)),
|
||||
("Einschleus", ("einschleus",)),
|
||||
("Weiche", ("weiche",)),
|
||||
)
|
||||
|
||||
LANE_LEFT = "L"
|
||||
LANE_RIGHT = "R"
|
||||
LANE_NAMES = {LANE_LEFT: "links", LANE_RIGHT: "rechts"}
|
||||
@@ -104,6 +133,8 @@ NODE_STYLES = {
|
||||
"Kreisel": dict(fillcolor="#2f5597", fontcolor="#ffffff", color="#1f3864"),
|
||||
"Gefaellestrecke": dict(fillcolor="#fbe5d6", fontcolor="#833c00", color="#c55a11"),
|
||||
"Strecke": dict(fillcolor="#e2f0d9", fontcolor="#375623", color="#548235"),
|
||||
"Beladung": dict(fillcolor="#d4f0d4", fontcolor="#1e5b1e", color="#338833"),
|
||||
"Entladung": dict(fillcolor="#f4d4d4", fontcolor="#7a1f1f", color="#cc3333"),
|
||||
}
|
||||
DEFAULT_NODE_STYLE = dict(fillcolor="#f2f2f2", fontcolor="#404040", color="#808080")
|
||||
|
||||
@@ -152,6 +183,25 @@ class Element:
|
||||
def is_device(self) -> bool:
|
||||
return self.kind in DEVICE_KINDS
|
||||
|
||||
@property
|
||||
def is_entry(self) -> bool:
|
||||
"""BTMT-Beladung: Materialeingang (Quelle)."""
|
||||
return self.kind in ENTRY_KINDS
|
||||
|
||||
@property
|
||||
def is_exit(self) -> bool:
|
||||
"""SC_Entladung: Materialausgang (Senke)."""
|
||||
return self.kind in EXIT_KINDS
|
||||
|
||||
@property
|
||||
def is_station(self) -> bool:
|
||||
return self.kind in STATION_KINDS
|
||||
|
||||
@property
|
||||
def is_transfer(self) -> bool:
|
||||
"""Ein-/Ausschleuselement oder Weiche (liefert Richtung, kein Knoten)."""
|
||||
return self.kind in TRANSFER_KINDS
|
||||
|
||||
@property
|
||||
def height(self) -> float | None:
|
||||
"""Montage-/Einbauhoehe in Meter, falls ableitbar."""
|
||||
@@ -323,6 +373,25 @@ def parse_neighbours(raw: str) -> list[tuple[str, str | None]]:
|
||||
return result
|
||||
|
||||
|
||||
def _match_kind(patterns, *texts: str) -> str | None:
|
||||
"""Erste Musterklasse, deren Schluesselwort in den Texten vorkommt."""
|
||||
haystack = " ".join(t.lower() for t in texts if t)
|
||||
for kind, keys in patterns:
|
||||
if any(key in haystack for key in keys):
|
||||
return kind
|
||||
return None
|
||||
|
||||
|
||||
def classify_station(*texts: str) -> str | None:
|
||||
"""BTMT Be-/Entladestation an TeileArt/Bezeichnung erkennen (sonst None)."""
|
||||
return _match_kind(STATION_PATTERNS, *texts)
|
||||
|
||||
|
||||
def classify_transfer(*texts: str) -> str | None:
|
||||
"""Ein-/Ausschleuselement bzw. Weiche erkennen (sonst None)."""
|
||||
return _match_kind(TRANSFER_PATTERNS, *texts)
|
||||
|
||||
|
||||
def read_elements(path: Path, warnings: list[str]) -> list[Element]:
|
||||
"""CSV-Export einlesen."""
|
||||
with path.open("r", encoding=CSV_ENCODING, newline="") as handle:
|
||||
@@ -343,6 +412,14 @@ def read_elements(path: Path, warnings: list[str]) -> list[Element]:
|
||||
|
||||
teile_art = (row.get("TeileArt") or "").strip()
|
||||
kind = teile_art[len(TEILEART_PREFIX):] if teile_art.startswith(TEILEART_PREFIX) else teile_art
|
||||
# BTMT Be-/Entladestation als Ein-/Ausgang erkennen (sps_skel-seitig)
|
||||
station_kind = classify_station(teile_art, row.get("Bezeichnung") or "")
|
||||
if station_kind:
|
||||
kind = station_kind
|
||||
# Ein-/Ausschleuselement bzw. Weiche erkennen (liefern Richtung)
|
||||
transfer_kind = classify_transfer(teile_art, row.get("Bezeichnung") or "")
|
||||
if transfer_kind:
|
||||
kind = transfer_kind
|
||||
|
||||
raw_merkmale = (row.get("Merkmale") or "").strip()
|
||||
merkmale: dict = {}
|
||||
@@ -395,6 +472,12 @@ def _build_nodes(elements: list[Element], warnings: list[str]) -> dict[str, Node
|
||||
for lane in (LANE_LEFT, LANE_RIGHT):
|
||||
node_id = f"{element.teile_id}-{lane}"
|
||||
nodes[node_id] = Node(node_id, element.kind, element, lane=lane)
|
||||
elif element.is_station:
|
||||
nodes[element.teile_id] = Node(element.teile_id, element.kind, element)
|
||||
elif element.is_transfer:
|
||||
# Ein-/Ausschleuselement / Weiche liefern nur die (gerichtete)
|
||||
# Verbindung, siehe _apply_transfers - kein eigener Knoten.
|
||||
continue
|
||||
elif not element.is_device:
|
||||
warnings.append(
|
||||
f"{element.describe()}: unbekannte TeileArt '{element.teile_art}' - kein Knoten erzeugt"
|
||||
@@ -611,6 +694,160 @@ def _orient(
|
||||
return edges
|
||||
|
||||
|
||||
def _nearest_node(station: Element, nodes: dict[str, Node]) -> str | None:
|
||||
"""Naechster Nicht-Stations-Knoten zur Station (2D-Abstand)."""
|
||||
if station.x is None or station.y is None:
|
||||
return None
|
||||
best, best_d = None, None
|
||||
for node_id, node in nodes.items():
|
||||
if node.element.is_station or node.element.x is None or node.element.y is None:
|
||||
continue
|
||||
d = (node.element.x - station.x) ** 2 + (node.element.y - station.y) ** 2
|
||||
if best_d is None or d < best_d:
|
||||
best, best_d = node_id, d
|
||||
return best
|
||||
|
||||
|
||||
def _attach_stations(
|
||||
elements: list[Element], nodes: dict[str, Node], warnings: list[str]
|
||||
) -> list[Edge]:
|
||||
"""
|
||||
Ein-/Ausgabestationen (BTMT Be-/Entladung) mit dem Fluss verbinden.
|
||||
|
||||
Eingang (Beladung) ist Quelle: Kante Station -> Ziel.
|
||||
Ausgang (Entladung) ist Senke: Kante Ziel -> Station.
|
||||
Ziel kommt aus den 'Nachbarn' des Exports; fehlen sie (dxfmakros exportiert
|
||||
fuer BTMT derzeit keine), wird das naechstgelegene Objekt per Abstand genommen.
|
||||
"""
|
||||
edges: list[Edge] = []
|
||||
for element in elements:
|
||||
if not element.is_station:
|
||||
continue
|
||||
sid = element.teile_id
|
||||
if sid not in nodes:
|
||||
continue
|
||||
partners: list[str] = []
|
||||
for base, lane in element.neighbours:
|
||||
if lane and f"{base}-{lane}" in nodes:
|
||||
partners.append(f"{base}-{lane}")
|
||||
elif base in nodes:
|
||||
partners.append(base)
|
||||
else:
|
||||
partners += [f"{base}-{L}" for L in (LANE_LEFT, LANE_RIGHT)
|
||||
if f"{base}-{L}" in nodes]
|
||||
if not partners:
|
||||
nearest = _nearest_node(element, nodes)
|
||||
if nearest:
|
||||
partners = [nearest]
|
||||
warnings.append(
|
||||
f"{element.describe()}: keine Nachbarn im Export - "
|
||||
f"naechstes Objekt '{nearest}' per Abstand verbunden"
|
||||
)
|
||||
else:
|
||||
warnings.append(f"{element.describe()}: kein Ziel gefunden - unverbunden")
|
||||
continue
|
||||
for partner in partners:
|
||||
if element.is_entry:
|
||||
edges.append(Edge(sid, partner, EDGE_FLOW))
|
||||
else:
|
||||
edges.append(Edge(partner, sid, EDGE_FLOW))
|
||||
return edges
|
||||
|
||||
|
||||
def _kreisel_lane_node(kreisel_base: str, strecke: Element, nodes: dict[str, Node]) -> str | None:
|
||||
"""Die konkrete Kreisel-Bahn (…-L/…-R), an der die Strecke haengt."""
|
||||
for base, lane in strecke.neighbours:
|
||||
if base == kreisel_base:
|
||||
node_id = f"{base}-{lane}" if lane else base
|
||||
return node_id if node_id in nodes else None
|
||||
for lane in (LANE_LEFT, LANE_RIGHT): # Fallback: irgendeine Bahn
|
||||
if f"{kreisel_base}-{lane}" in nodes:
|
||||
return f"{kreisel_base}-{lane}"
|
||||
return kreisel_base if kreisel_base in nodes else None
|
||||
|
||||
|
||||
def _apply_transfers(
|
||||
elements: list[Element], by_id: dict[str, Element],
|
||||
nodes: dict[str, Node], edges: list[Edge], warnings: list[str]
|
||||
) -> list[Edge]:
|
||||
"""
|
||||
Kreisel<->Strecke-Kanten anhand der Transferelemente gerichtet setzen.
|
||||
|
||||
Ausschleuselement: Kreisel -> Strecke (Streckenanfang).
|
||||
Einschleuselement: Strecke -> Kreisel (Streckenende).
|
||||
Weiche: Kreisel<->Kreisel als Weiche markieren.
|
||||
Die so definierten Kanten sind massgeblich und ersetzen die aus Hoehe/
|
||||
Nachbarn abgeleitete Richtung fuer genau diese Knotenpaare.
|
||||
"""
|
||||
# Transfer je Strecke sammeln: aus = Einlauf (Kreisel->Strecke, Anfang),
|
||||
# ein = Auslauf (Strecke->Kreisel, Ende)
|
||||
per_strecke: dict[str, dict[str, str]] = {}
|
||||
weiche_pairs: list[set[str]] = []
|
||||
for el in elements:
|
||||
if el.kind in ("Ausschleus", "Einschleus"):
|
||||
k = str(_merkmal(el.merkmale, "Kreisel") or "").strip()
|
||||
s = str(_merkmal(el.merkmale, "Strecke") or "").strip()
|
||||
if k and s:
|
||||
role = "aus" if el.kind == "Ausschleus" else "ein"
|
||||
per_strecke.setdefault(s, {})[role] = k
|
||||
elif el.kind == "Weiche":
|
||||
a = str(_merkmal(el.merkmale, "Kreisel A") or "").strip()
|
||||
b = str(_merkmal(el.merkmale, "Kreisel B") or "").strip()
|
||||
if a and b:
|
||||
weiche_pairs.append({a, b})
|
||||
|
||||
directed: dict[frozenset, tuple[str, str]] = {}
|
||||
for s, roles in per_strecke.items():
|
||||
if s not in nodes or nodes[s].element.is_circle:
|
||||
continue
|
||||
strecke_el = by_id.get(s)
|
||||
aus_lane = _kreisel_lane_node(roles["aus"], strecke_el, nodes) if roles.get("aus") else None
|
||||
ein_lane = _kreisel_lane_node(roles["ein"], strecke_el, nodes) if roles.get("ein") else None
|
||||
# Ruecklauf-Strecke (beide Enden am selben Kreisel): auf die zwei Bahnen verteilen
|
||||
if aus_lane and aus_lane == ein_lane:
|
||||
base = aus_lane.split("-")[0]
|
||||
lanes = [f"{base}-{L}" for L in (LANE_LEFT, LANE_RIGHT) if f"{base}-{L}" in nodes]
|
||||
if len(lanes) == 2:
|
||||
aus_lane, ein_lane = lanes
|
||||
warnings.append(
|
||||
f"Strecke {s}: Ein- und Ausschleusung am selben Kreisel {base} "
|
||||
f"(Ruecklauf) - Bahnen {lanes[0]}/{lanes[1]} angenommen"
|
||||
)
|
||||
if aus_lane: # Kreisel -> Strecke
|
||||
directed[frozenset({aus_lane, s})] = (aus_lane, s, "Ausschleusung")
|
||||
if ein_lane: # Strecke -> Kreisel
|
||||
directed[frozenset({ein_lane, s})] = (s, ein_lane, "Einschleusung")
|
||||
|
||||
# Fuer Strecken mit Transferelementen sind ALLEIN diese massgeblich fuer die
|
||||
# Kreisel<->Strecke-Verbindung. Jede weitere Kreisel<->Strecke-Kante aus den
|
||||
# rohen Nachbarn (Bounding-Box-Ueberlappung, z. B. ein VF, der zufaellig neben
|
||||
# einem dritten Kreisel liegt) ist ein Falschtreffer und wird verworfen.
|
||||
covered = set(per_strecke)
|
||||
result: list[Edge] = []
|
||||
for e in edges:
|
||||
if frozenset({e.src, e.dst}) in directed:
|
||||
continue # wird gerichtet neu gesetzt
|
||||
ends = {e.src, e.dst}
|
||||
strecke_end = ends & covered
|
||||
others = ends - strecke_end
|
||||
if strecke_end and len(others) == 1:
|
||||
other = next(iter(others))
|
||||
if other in nodes and nodes[other].kind in CIRCLE_KINDS:
|
||||
warnings.append(
|
||||
f"Kante {other} <-> {next(iter(strecke_end))}: Kreisel<->Strecke "
|
||||
f"ohne Transferelement - als Bounding-Box-Falschtreffer verworfen"
|
||||
)
|
||||
continue
|
||||
result.append(e)
|
||||
for (src, dst, note) in directed.values():
|
||||
result.append(Edge(src, dst, EDGE_FLOW, note))
|
||||
# Kreisel<->Kreisel-Kanten, die eine Weiche verbinden, markieren
|
||||
for e in result:
|
||||
if {e.src.split("-")[0], e.dst.split("-")[0]} in weiche_pairs:
|
||||
e.note = "Weiche"
|
||||
return result
|
||||
|
||||
|
||||
def build_graph(elements: list[Element], source: str, warnings: list[str]) -> Graph:
|
||||
by_id = {element.teile_id: element for element in elements}
|
||||
nodes = _build_nodes(elements, warnings)
|
||||
@@ -650,6 +887,12 @@ def build_graph(elements: list[Element], source: str, warnings: list[str]) -> Gr
|
||||
edges.append(Edge(left, right, EDGE_CIRCLE, "unbestimmt"))
|
||||
edges.append(Edge(right, left, EDGE_CIRCLE, "unbestimmt"))
|
||||
|
||||
# Ein-/Ausgabestationen (BTMT Be-/Entladung) als Quelle/Senke anbinden
|
||||
edges.extend(_attach_stations(elements, nodes, warnings))
|
||||
|
||||
# Kreisel<->Strecke gerichtet aus den Transferelementen (massgeblich)
|
||||
edges = _apply_transfers(elements, by_id, nodes, edges, warnings)
|
||||
|
||||
# Sackgassen und unerreichbare Bahnen melden. Sie entstehen, wenn ein
|
||||
# Kreisel nur in einer Richtung durchlaufen wird (Drehrichtung) und die
|
||||
# Gegenrichtung nicht ueber aeussere Verbindungen geschlossen ist.
|
||||
@@ -685,6 +928,14 @@ def _node_label(node: Node) -> str:
|
||||
element = node.element
|
||||
lines: list[str] = []
|
||||
|
||||
if element.is_station:
|
||||
role = "Materialeingang" if element.is_entry else "Materialausgang"
|
||||
lines.append(f"{element.name or element.kind} [{element.teile_id}]")
|
||||
lines.append(f"{element.kind} - {role}")
|
||||
if element.height is not None:
|
||||
lines.append(f"h = {element.height:.3f} m")
|
||||
return _dot_label(lines)
|
||||
|
||||
if node.lane:
|
||||
name = _merkmal(element.merkmale, "Name") or element.name
|
||||
lines.append(f"{name} [{node.node_id}]")
|
||||
@@ -716,12 +967,47 @@ def _node_label(node: Node) -> str:
|
||||
return _dot_label(lines)
|
||||
|
||||
|
||||
def render_dot(graph: Graph) -> str:
|
||||
# Zeichnungsmassstab fuer --use-cords: Graphviz rechnet in Punkt, die Anlage in
|
||||
# Millimeter. 0.25 pt/mm (1:4) - wie in tro_flow.py, damit Material- und TRO-Graph
|
||||
# im selben Massstab lagerichtig sind.
|
||||
COORD_SCALE = 0.25
|
||||
# Die beiden Bahnen (L/R) eines Kreisels teilen sich eine Position; im
|
||||
# lagerichtigen Layout leicht versetzen, damit sie nicht uebereinander liegen.
|
||||
LANE_NUDGE_MM = 350.0
|
||||
|
||||
|
||||
def _node_xy(node: "Node") -> tuple[float, float] | None:
|
||||
"""Anlagenkoordinate eines Knotens (Kreisel-Bahnen L/R leicht versetzt)."""
|
||||
el = node.element
|
||||
if el.x is None or el.y is None:
|
||||
return None
|
||||
x, y = el.x, el.y
|
||||
if node.lane == "L":
|
||||
y += LANE_NUDGE_MM
|
||||
elif node.lane == "R":
|
||||
y -= LANE_NUDGE_MM
|
||||
return (x, y)
|
||||
|
||||
|
||||
def _pos_attr(node: "Node", use_coords: bool) -> str:
|
||||
"""pos-Attribut fuer 'neato -n', oder leer ohne --use-cords."""
|
||||
if not use_coords:
|
||||
return ""
|
||||
xy = _node_xy(node)
|
||||
if xy is None:
|
||||
return ""
|
||||
return f', pos="{xy[0] * COORD_SCALE:.2f},{xy[1] * COORD_SCALE:.2f}"'
|
||||
|
||||
|
||||
def render_dot(graph: Graph, use_coords: bool = False) -> str:
|
||||
out: list[str] = []
|
||||
add = out.append
|
||||
|
||||
add("// Automatisch erzeugt von lib/material_flow.py - nicht manuell aendern.")
|
||||
add(f"// Quelle: {graph.source}")
|
||||
if use_coords:
|
||||
add(f"// Lagerichtig: Knoten auf Anlagenkoordinate, {COORD_SCALE} pt/mm.")
|
||||
add("// Rendern mit: neato -n -Tsvg <datei>.dot -o <datei>.svg")
|
||||
add("digraph Materialfluss {")
|
||||
add(' graph [rankdir=LR, splines=spline, nodesep=0.25, ranksep=1.6,')
|
||||
add(' fontname="Segoe UI", fontsize=11, labelloc="t",')
|
||||
@@ -755,7 +1041,8 @@ def render_dot(graph: Graph) -> str:
|
||||
style = NODE_STYLES.get(element.kind, DEFAULT_NODE_STYLE)
|
||||
for node in sorted(lanes, key=lambda n: n.lane or ""):
|
||||
attrs = ", ".join(f'{key}="{value}"' for key, value in style.items())
|
||||
add(f' "{node.node_id}" [label="{_node_label(node)}", {attrs}];')
|
||||
add(f' "{node.node_id}" [label="{_node_label(node)}", {attrs}'
|
||||
f'{_pos_attr(node, use_coords)}];')
|
||||
add(" { rank=same; " + " ".join(f'"{n.node_id}";' for n in lanes) + " }")
|
||||
add(" }")
|
||||
add("")
|
||||
@@ -763,7 +1050,8 @@ def render_dot(graph: Graph) -> str:
|
||||
for node in plain:
|
||||
style = NODE_STYLES.get(node.kind, DEFAULT_NODE_STYLE)
|
||||
attrs = ", ".join(f'{key}="{value}"' for key, value in style.items())
|
||||
add(f' "{node.node_id}" [label="{_node_label(node)}", {attrs}];')
|
||||
add(f' "{node.node_id}" [label="{_node_label(node)}", {attrs}'
|
||||
f'{_pos_attr(node, use_coords)}];')
|
||||
add("")
|
||||
|
||||
for edge in graph.edges:
|
||||
@@ -774,17 +1062,30 @@ def render_dot(graph: Graph) -> str:
|
||||
'arrowsize=0.9, fontcolor="#5b7fc7", '
|
||||
f'tooltip="Kreisel-Umlauf {edge.note}"{label}]')
|
||||
elif edge.kind == EDGE_UNRESOLVED:
|
||||
weiche = ', xlabel="Weiche", fontcolor="#cc8800"' if edge.note == "Weiche" else ""
|
||||
attrs = ('[color="#bf8f00", style=dashed, dir=both, '
|
||||
f'tooltip="Richtung {edge.note}"]')
|
||||
f'tooltip="Richtung {edge.note}"{weiche}]')
|
||||
elif edge.note == "Weiche":
|
||||
attrs = ('[color="#cc8800", penwidth=1.6, xlabel="Weiche", '
|
||||
'fontcolor="#cc8800", tooltip="Weiche (Kreisel-Uebergang)"]')
|
||||
else:
|
||||
attrs = "[]"
|
||||
add(f' "{edge.src}" -> "{edge.dst}" {attrs};')
|
||||
add("")
|
||||
|
||||
legend_pos = ""
|
||||
if use_coords:
|
||||
xs = [xy[0] for n in graph.nodes.values() if (xy := _node_xy(n))]
|
||||
ys = [xy[1] for n in graph.nodes.values() if (xy := _node_xy(n))]
|
||||
if xs and ys:
|
||||
lx, ly = min(xs), min(ys) - 3000.0 # etwas unterhalb der Anlage
|
||||
legend_pos = (f' pos="{lx * COORD_SCALE:.2f},'
|
||||
f'{ly * COORD_SCALE:.2f}",')
|
||||
|
||||
add(" subgraph cluster_legende {")
|
||||
add(' label="Legende"; style="rounded"; color="#a6a6a6";')
|
||||
add(' fontname="Segoe UI"; fontsize=10; fontcolor="#404040";')
|
||||
add(' legende [shape=plaintext, style="", fillcolor="none", label=<'
|
||||
add(f' legende [shape=plaintext, style="", fillcolor="none",{legend_pos} label=<'
|
||||
'<table border="0" cellborder="0" cellspacing="2" cellpadding="2">'
|
||||
'<tr><td align="left">durchgezogen</td><td align="left">Materialfluss</td></tr>'
|
||||
'<tr><td align="left"><font color="#8ea9db">gestrichelt blau</font></td>'
|
||||
@@ -800,6 +1101,120 @@ def render_dot(graph: Graph) -> str:
|
||||
return "\n".join(out) + "\n"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# connect.ini (Item-/Flussebene) erzeugen
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _connect_kind(edge: Edge) -> str:
|
||||
"""Kantenart fuer connect.ini."""
|
||||
if edge.kind == EDGE_CIRCLE:
|
||||
return "umlauf"
|
||||
if edge.note == "Weiche":
|
||||
return "weiche"
|
||||
if edge.kind == EDGE_UNRESOLVED:
|
||||
return "unbestimmt"
|
||||
if edge.note in ("Einschleusung", "Ausschleusung"):
|
||||
return edge.note.lower()
|
||||
return "normal"
|
||||
|
||||
|
||||
def render_connect_ini(graph: Graph, section: str = "Anlage", overrides=None) -> str:
|
||||
"""
|
||||
Materialfluss-Topologie als connect.ini (Item-/Flussebene).
|
||||
|
||||
Zwischenstand fuer die TRO-Ableitung: Knoten = Flussobjekte (Kreisel-Bahnen,
|
||||
Strecke, Gefaellestrecke), Kanten = gerichteter Materialfluss. Von Hand
|
||||
editierbar. Aufbau angelehnt an doc/TRO_Katalog/connect.ini.
|
||||
|
||||
`overrides` (aus cfg/tro_overrides.ini) wird als eigener Abschnitt
|
||||
`[<Anlage>.manual]` angehaengt. Die Datei hier wird bei jedem Lauf neu
|
||||
geschrieben - von Hand eingetragene Weichen waeren sonst weg. Der Abschnitt
|
||||
beschreibt bewusst nur, WAS vor Ort bestaetigt wurde; er erfindet keine
|
||||
Flusskanten, denn welche Kante zur Weiche gehoert, ist aus dem Layout nicht
|
||||
ableitbar.
|
||||
"""
|
||||
out: list[str] = []
|
||||
add = out.append
|
||||
add("; ============================================================")
|
||||
add("; connect.ini - Materialfluss-Topologie (Item-/Flussebene)")
|
||||
add(f"; Automatisch erzeugt von lib/material_flow.py aus '{graph.source}'.")
|
||||
add("; Zwischenstand fuer die TRO-Ableitung; von Hand editierbar.")
|
||||
add(";")
|
||||
add("; [<Anlage>.nodes] <Knoten> = <Art> | <Bezeichnung>")
|
||||
add("; [<Anlage>.connections] cNNN = <von> -> <nach> | <JamArea> | <kind>")
|
||||
add("; kind: normal | umlauf | weiche | einschleusung | ausschleusung | unbestimmt")
|
||||
add("; [<Anlage>.externals] eNNN = EXTERN -> <Knoten> | <Info> (Zulauf/Beladung)")
|
||||
add("; eNNN = <Knoten> -> EXTERN | <Info> (Ablauf/Entladung)")
|
||||
add("; [<Anlage>.manual] vor Ort bestaetigte Angaben aus cfg/tro_overrides.ini")
|
||||
add("; ============================================================")
|
||||
add("")
|
||||
add(f"[{section}.nodes]")
|
||||
for node_id in sorted(graph.nodes):
|
||||
node = graph.nodes[node_id]
|
||||
el = node.element
|
||||
if node.lane:
|
||||
name = _merkmal(el.merkmale, "Name") or el.name
|
||||
comment = f"{name} (Bahn {node.lane})"
|
||||
else:
|
||||
comment = el.name or el.kind
|
||||
add(f"{node_id} = {node.kind} | {comment}")
|
||||
add("")
|
||||
add(f"[{section}.connections]")
|
||||
for i, edge in enumerate(sorted(graph.edges, key=lambda e: (e.src, e.dst)), start=1):
|
||||
add(f"c{i:03d} = {edge.src} -> {edge.dst} | | {_connect_kind(edge)}")
|
||||
add("")
|
||||
add(f"[{section}.externals]")
|
||||
stations = [n for n in graph.nodes.values() if n.element.is_station]
|
||||
if stations:
|
||||
for j, node in enumerate(sorted(stations, key=lambda n: n.node_id), start=1):
|
||||
if node.element.is_entry:
|
||||
add(f"e{j:03d} = EXTERN -> {node.node_id} | {node.element.name}")
|
||||
else:
|
||||
add(f"e{j:03d} = {node.node_id} -> EXTERN | {node.element.name}")
|
||||
else:
|
||||
add("; Noch keine BTMT Be-/Entladestation im Layout - Ein-/Ausgang offen.")
|
||||
add("; Kandidaten (offene Enden aus dem Materialfluss):")
|
||||
incoming = {e.dst for e in graph.edges}
|
||||
outgoing = {e.src for e in graph.edges}
|
||||
for node_id in sorted(graph.nodes):
|
||||
node = graph.nodes[node_id]
|
||||
if node.element.is_station:
|
||||
continue
|
||||
if node_id not in incoming:
|
||||
add(f"; Beladung? EXTERN -> {node_id} ({node.kind}, keine Zufuhr)")
|
||||
if node_id not in outgoing:
|
||||
add(f"; Entladung? {node_id} -> EXTERN ({node.kind}, kein Abfluss)")
|
||||
add("")
|
||||
|
||||
if overrides is not None and (overrides.tros or overrides.splits
|
||||
or overrides.open_points):
|
||||
add(f"[{section}.manual]")
|
||||
add("; Vor Ort bestaetigt, im CSV-Export nicht enthalten.")
|
||||
add(f"; Quelle: {overrides.source}")
|
||||
add("; Diese Zeilen werden aus der Override-Datei erzeugt - nicht hier")
|
||||
add("; editieren, sonst sind sie beim naechsten Lauf wieder weg.")
|
||||
for section_key, override in sorted(overrides.tros.items()):
|
||||
if not override.type_name:
|
||||
continue
|
||||
wie = f"Separator {override.key}" if override.by_separator else override.key
|
||||
add(f"{section_key} = {override.type_name} | {wie} | {override.reason}")
|
||||
for spec in overrides.splits:
|
||||
add(f"{spec.name} = NEU {spec.type_name} "
|
||||
f"| Separator {', '.join(spec.separators)}"
|
||||
+ (f" + Scanner {', '.join(spec.scanners)}" if spec.scanners else "")
|
||||
+ (f" | Bahn {spec.host}" if spec.host else "")
|
||||
+ f" | {spec.reason}")
|
||||
for point in overrides.open_points:
|
||||
add(f"{point.name} = OFFEN {point.expect or '?'} "
|
||||
f"| x={point.x:.2f} y={point.y:.2f} "
|
||||
f"| heute Teil von {point.belongs_to or '-'} "
|
||||
f"| {point.note}")
|
||||
add("")
|
||||
|
||||
return "\n".join(out) + "\n"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# SVG erzeugen
|
||||
# ---------------------------------------------------------------------------
|
||||
@@ -851,9 +1266,13 @@ def graph_to_svg(
|
||||
f"'{engine}' ist mit Code {result.returncode} fehlgeschlagen:\n"
|
||||
f"{(result.stderr or result.stdout).strip()}"
|
||||
)
|
||||
# neato meldet fehlende Positionen nur als Warnung auf stderr
|
||||
if result.stderr.strip():
|
||||
raise RuntimeError(f"'{engine}' meldet: {result.stderr.strip()}")
|
||||
# 'neato -n' gibt harmlose Warnungen auf stderr aus (z. B. sich beruehrende
|
||||
# Knoten -> gerade Kanten). Nur dann scheitern, wenn wirklich keine SVG
|
||||
# entstanden ist; sonst ist die Datei gueltig und die Warnung unkritisch.
|
||||
if not svg_file.is_file() or svg_file.stat().st_size == 0:
|
||||
raise RuntimeError(
|
||||
f"'{engine}' hat keine SVG erzeugt: {result.stderr.strip() or '(keine Meldung)'}"
|
||||
)
|
||||
|
||||
|
||||
def dot_to_svg(dot_file: Path, svg_file: Path) -> None:
|
||||
@@ -1164,12 +1583,28 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
|
||||
help="Aus der erzeugten DOT-Datei zusaetzlich eine SVG-Datei erzeugen "
|
||||
"(benoetigt Graphviz 'dot').",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--use-cords",
|
||||
"--use-coords",
|
||||
dest="use_cords",
|
||||
action="store_true",
|
||||
help="Jeden Knoten auf seine Anlagenkoordinate setzen und das SVG von "
|
||||
f"Graphviz 'neato -n' erzeugen statt von 'dot' (lagerichtig, "
|
||||
f"{COORD_SCALE} pt/mm). Nur mit --tosvg wirksam.",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--doc",
|
||||
action="store_true",
|
||||
help="Dokumentation des Materialflusses als Markdown in %%SKEL_RESULTS%% "
|
||||
"erzeugen (Anlagenobjekte, Verbindungen, Pruefungen).",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--connect",
|
||||
action="store_true",
|
||||
help="Materialfluss-Topologie zusaetzlich als connect.ini (Item-/Flussebene) "
|
||||
"nach %%SKEL_RESULTS%% schreiben (Knoten, gerichtete Verbindungen, "
|
||||
"Ein-/Ausgang-Kandidaten) - Zwischenstand fuer die TRO-Ableitung.",
|
||||
)
|
||||
return parser.parse_args(argv)
|
||||
|
||||
|
||||
@@ -1210,13 +1645,26 @@ def main(argv: list[str] | None = None) -> int:
|
||||
svg_file = results / f"{stem}_material_flow.svg"
|
||||
doc_target = results / f"{stem}_material_flow.md"
|
||||
|
||||
# --use-cords braucht an jedem Knoten eine Koordinate
|
||||
if args.use_cords:
|
||||
without = [n.node_id for n in graph.nodes.values() if _node_xy(n) is None]
|
||||
if without:
|
||||
print(f"FEHLER: --use-cords nicht moeglich, {len(without)} Knoten ohne "
|
||||
f"Koordinate: {', '.join(without)}", file=sys.stderr)
|
||||
print(" Ohne pos an jedem Knoten kann 'neato -n' nicht zeichnen.",
|
||||
file=sys.stderr)
|
||||
return 1
|
||||
|
||||
existing_svg = svg_file if svg_file.is_file() else None
|
||||
dot_file.write_text(render_dot(graph), encoding="utf-8")
|
||||
dot_file.write_text(render_dot(graph, use_coords=args.use_cords), encoding="utf-8")
|
||||
|
||||
svg_error: str | None = None
|
||||
if args.tosvg:
|
||||
try:
|
||||
dot_to_svg(dot_file, svg_file)
|
||||
if args.use_cords:
|
||||
graph_to_svg(dot_file, svg_file, engine="neato", no_op=1)
|
||||
else:
|
||||
dot_to_svg(dot_file, svg_file)
|
||||
except RuntimeError as exc:
|
||||
svg_error = str(exc)
|
||||
created_svg = svg_file if args.tosvg and svg_error is None else None
|
||||
@@ -1229,6 +1677,20 @@ def main(argv: list[str] | None = None) -> int:
|
||||
)
|
||||
doc_file = doc_target
|
||||
|
||||
connect_file = None
|
||||
if args.connect:
|
||||
# Spaet importiert: tro_overrides zieht tro_catalog nach, das
|
||||
# material_flow sonst nicht braucht.
|
||||
from tro_overrides import load_overrides
|
||||
|
||||
overrides = load_overrides(csv_file.name, env_dir("SKEL_CFG", "cfg"))
|
||||
connect_target = results / f"{stem}_connect.ini"
|
||||
connect_target.write_text(
|
||||
render_connect_ini(graph, section=stem, overrides=overrides),
|
||||
encoding="utf-8",
|
||||
)
|
||||
connect_file = connect_target
|
||||
|
||||
report(
|
||||
graph,
|
||||
findings,
|
||||
|
||||
@@ -0,0 +1,431 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
# Absichtlich ohne Shebang: der Windows-py-Launcher wuerde daraus "python3"
|
||||
# ableiten und scheitert, wenn keine PythonCore-Installation registriert ist.
|
||||
"""
|
||||
scl_skeleton.py - FB_Main-Geruest als SCL aus der TRO-Liste erzeugen.
|
||||
|
||||
Letzter Schritt der Kette:
|
||||
|
||||
material_flow.py Objekte -> Materialfluss
|
||||
tro_flow.py Fluss -> TROs
|
||||
tro_annotate.py TROs -> Marker in der Zeichnung
|
||||
tro_extract.py Zeichnung -> JSON
|
||||
scl_skeleton.py JSON -> FB_Main.scl <- hier
|
||||
|
||||
Erzeugt wird ein **Geruest**, kein fertiges Programm: je TRO eine REGION mit
|
||||
Instanzaufruf und allen Parameterzeilen in der richtigen Reihenfolge, so wie es
|
||||
die realen Aufrufstellen in doc/TRO_Katalog/scl_templates/*.scl zeigen.
|
||||
|
||||
Was aus dem Layout kommt, steht drin. Was aus der E-Planung kommen muss, steht
|
||||
als Platzhalter mit `TODO(E-Planung)` da - bewusst als offensichtliche Luecke
|
||||
und nicht als erfundener Wert, denn eine plausibel aussehende falsche
|
||||
Sensoradresse ist schlimmer als eine leere. Alle Platzhalter sind ueber genau
|
||||
diesen Marker auffindbar.
|
||||
|
||||
Ableitbar und daher gefuellt:
|
||||
- Reihenfolge der REGIONs (Materialfluss), Instanzliste, FB-Typ je TRO
|
||||
- Anzahl der Separator-/Weichen-/Scanner-Parameterbloecke je TRO
|
||||
- nTo1Destinations einer Weiche = Anzahl ihrer Abgaenge
|
||||
- die Zielliste je Weichenabgang (Erreichbarkeit im TRO-Graphen)
|
||||
- Querverweis auf die Bauteile im Layout (Separatornummern, Koordinate)
|
||||
|
||||
Aufruf ueber bin/scl_skeleton.bat bzw. bin/scl_skeleton.sh.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import sys
|
||||
from collections import deque
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
|
||||
from material_flow import env_dir
|
||||
|
||||
TODO = "TODO(E-Planung)"
|
||||
|
||||
# Zeitwerte aus den Referenz-Aufrufstellen. Sie sind plausible Startwerte, KEINE
|
||||
# fuer diese Anlage ermittelten - deshalb ueberall mit Pruefhinweis.
|
||||
DEFAULT_SEP_TIMES = (
|
||||
("tDelayToNextItem", '"TimeForSepWaiting_Short"', None),
|
||||
("tTrailingTime", "t#600ms", "pruefen"),
|
||||
("tHandlingTime", "t#200ms", "pruefen"),
|
||||
("tJam", '"TimeForSepJam"', None),
|
||||
)
|
||||
|
||||
|
||||
def load_tros(path: Path) -> dict:
|
||||
return json.loads(path.read_text(encoding="utf-8"))
|
||||
|
||||
|
||||
def flow_order(tros: list[dict], start_id: str = "") -> tuple[list[dict], str]:
|
||||
"""TROs in Fliessrichtung sortieren.
|
||||
|
||||
Startpunkt ist ein TRO ohne Vorgaenger - also die Stelle, an der Material
|
||||
in die Anlage kommt. Solange Be-/Entladestation fehlen, ist der Fluss ein
|
||||
geschlossener Kreis und es gibt keinen solchen TRO; dann muss der Kreis
|
||||
irgendwo aufgetrennt werden. Das ist eine willkuerliche Wahl, deshalb wird
|
||||
sie zurueckgegeben und im Kopf der SCL vermerkt (und laesst sich mit
|
||||
--start ueberschreiben). Nicht erreichte TROs haengt die Funktion hinten
|
||||
an, damit nie einer verlorengeht.
|
||||
"""
|
||||
nach_id = {t["id"]: t for t in tros}
|
||||
ohne_vorgaenger = sorted(t["id"] for t in tros if not t.get("predecessors"))
|
||||
|
||||
if start_id and start_id in nach_id:
|
||||
start, grund = [start_id], "per --start gesetzt"
|
||||
elif ohne_vorgaenger:
|
||||
start, grund = ohne_vorgaenger, "TRO ohne Vorgaenger (Materialeingang)"
|
||||
else:
|
||||
start = [min(nach_id)]
|
||||
grund = ("Kreislauf ohne Eingang - willkuerlich aufgetrennt; sobald "
|
||||
"Be-/Entladestation im Export stehen, ergibt sich der Anfang")
|
||||
|
||||
gesehen: list[str] = []
|
||||
queue = deque(start)
|
||||
while queue:
|
||||
tro_id = queue.popleft()
|
||||
if tro_id in gesehen:
|
||||
continue
|
||||
gesehen.append(tro_id)
|
||||
for nxt in nach_id.get(tro_id, {}).get("successors", []):
|
||||
if nxt not in gesehen:
|
||||
queue.append(nxt)
|
||||
for tro_id in sorted(nach_id):
|
||||
if tro_id not in gesehen:
|
||||
gesehen.append(tro_id)
|
||||
return [nach_id[i] for i in gesehen], grund
|
||||
|
||||
|
||||
def reachable_via(start: str, first: str, tros: dict[str, dict]) -> list[str]:
|
||||
"""Welche TROs sind ueber diesen Abgang erreichbar?
|
||||
|
||||
Das ist die Zielliste eines Weichenabgangs: von `first` aus weiterlaufen,
|
||||
ohne ueber die Weiche selbst zurueckzugehen. Genau die Information, die
|
||||
FC_Direction braucht, um zu entscheiden - und sie steht im Graphen.
|
||||
"""
|
||||
gesehen: set[str] = set()
|
||||
queue = deque([first])
|
||||
while queue:
|
||||
node = queue.popleft()
|
||||
if node in gesehen or node == start:
|
||||
continue
|
||||
gesehen.add(node)
|
||||
queue.extend(tros.get(node, {}).get("successors", []))
|
||||
return sorted(gesehen)
|
||||
|
||||
|
||||
def region_title(tro: dict) -> str:
|
||||
seps = ", ".join(tro.get("separators", []))
|
||||
teile = f" Sep {seps}" if seps else ""
|
||||
return f"{tro['id']} ({tro['type']}){teile}"
|
||||
|
||||
|
||||
def emit_separators(add, tro: dict, anzahl: int, hat_scanner: bool) -> None:
|
||||
seps = tro.get("separators", [])
|
||||
for i in range(1, anzahl + 1):
|
||||
pfad = f"#{tro['id']}.stInSeparator{i}"
|
||||
layout = seps[i - 1] if i <= len(seps) else "?"
|
||||
add(f" // Separator {i} = Bauteil {layout} aus dem Layout")
|
||||
add(f' {pfad}.Settings.nSeparatorNo := "cSep_____"; // {TODO}')
|
||||
for name, wert, hinweis in DEFAULT_SEP_TIMES:
|
||||
kommentar = f" // {TODO}: {hinweis}" if hinweis else ""
|
||||
add(f" {pfad}.Settings.{name} := {wert};{kommentar}")
|
||||
add(f" {pfad}.Settings.xPluggedJam := TRUE;")
|
||||
add(f" {pfad}.Settings.xPluggedPart := FALSE;")
|
||||
add(f" {pfad}.Settings.xPluggedScanner := "
|
||||
f"{'TRUE' if hat_scanner else 'FALSE'};")
|
||||
add(f' {pfad}.stSenInSep := "DB_Inputs".Sensors["cInBG____"]; // {TODO}')
|
||||
add(f" {pfad}.Settings.tSenFree := T#1500ms; // {TODO}: pruefen")
|
||||
add(f" {pfad}.Settings.tSenWait := T#4000ms; // {TODO}: pruefen")
|
||||
add("")
|
||||
|
||||
|
||||
def emit_switches(add, tro: dict, anzahl: int, alle: dict[str, dict]) -> None:
|
||||
abgaenge = tro.get("successors", [])
|
||||
for j in range(1, anzahl + 1):
|
||||
pfad = f"#{tro['id']}.stInSwitch{j}"
|
||||
add(f' {pfad}.Settings.nSwitchNo := "cSwi_____"; // {TODO}')
|
||||
add(f" {pfad}.Settings.xPluggedExSen3 := FALSE;")
|
||||
add(f" {pfad}.Settings.xPluggedExSen4 := FALSE;")
|
||||
if anzahl == 1:
|
||||
# Eine Weiche bedient alle Abgaenge - direkt ableitbar.
|
||||
add(f" {pfad}.Settings.nTo1Destinations := {len(abgaenge)};"
|
||||
f" // Abgaenge laut Topologie")
|
||||
else:
|
||||
# Bei mehreren Weichen ist es eine Planungsentscheidung, welche
|
||||
# Weiche welche Abgaenge bedient - die Summe ist bekannt, die
|
||||
# Aufteilung nicht. Deshalb hier kein geratener Wert.
|
||||
add(f" {pfad}.Settings.nTo1Destinations := _;"
|
||||
f" // {TODO}: dieser TRO hat {len(abgaenge)} "
|
||||
f"Abgaenge auf {anzahl} Weichen - Aufteilung festlegen")
|
||||
add("")
|
||||
if abgaenge:
|
||||
add(" // Zielliste je Abgang (aus dem TRO-Graphen berechnet) -")
|
||||
add(" // Grundlage fuer die Zuordnung Ziel -> Richtung in FC_Direction:")
|
||||
for abgang in abgaenge:
|
||||
ziele = reachable_via(tro["id"], abgang, alle)
|
||||
add(f" // ueber {abgang}: {', '.join(ziele) if ziele else '-'}")
|
||||
add("")
|
||||
|
||||
|
||||
def emit_vario(add, tro: dict) -> None:
|
||||
pfad = f"#{tro['id']}.stInVario"
|
||||
add(f" {pfad}.Settings.xContinousEmptying := FALSE;")
|
||||
add(f" {pfad}.Settings.tWaitForJog := T#300ms; // {TODO}: pruefen")
|
||||
add(f" {pfad}.Settings.tConvEmpty := T#45000ms; // {TODO}: pruefen")
|
||||
add(f" {pfad}.Settings.nCountCarrier := 1; // {TODO}: pruefen")
|
||||
add(f" {pfad}.Settings.tWaitForStop := t#3s;")
|
||||
add(f" {pfad}.Settings.tCarrierInPosDelay := t#1200ms;")
|
||||
add(f" {pfad}.Settings.tCarrierInPosEdgeDelay := t#100ms;")
|
||||
add(f" {pfad}.Settings.xJamSenDelayActive := TRUE;")
|
||||
add(f" {pfad}.Settings.tJamSenDelay := T#2s;")
|
||||
add(f' {pfad}.stSenCarrInPos := "DB_Inputs".Sensors["cInBG____"]; // {TODO}')
|
||||
add(f' {pfad}.stSenFinger := "DB_Inputs".Sensors["cInBG____"]; // {TODO}')
|
||||
add(f' {pfad}.stSenLastPos := "DB_Inputs".Sensors["cInBG____"]; // {TODO}')
|
||||
add(f' {pfad}.stSenJam := "DB_Inputs".Sensors["cInBG____"]; // {TODO}')
|
||||
add("")
|
||||
|
||||
|
||||
def emit_barcode(add, tro: dict, anzahl: int) -> None:
|
||||
for i in range(1, anzahl + 1):
|
||||
suffix = "" if anzahl == 1 else str(i)
|
||||
pfad = f"#{tro['id']}.stInBarcodeReader{suffix}"
|
||||
add(f" {pfad}.Settings.xActivateCheckASCI := TRUE;")
|
||||
add(f" {pfad}.Settings.tTimeout := T#3000ms;")
|
||||
add(f' // {TODO}: Cognex-Interface verdrahten '
|
||||
f'(#arCognexInterface["cScanner____"])')
|
||||
add("")
|
||||
|
||||
|
||||
def emit_call(add, tro: dict, items: dict[str, int]) -> None:
|
||||
"""Aufrufstelle schreiben.
|
||||
|
||||
Die Zeilen werden als (Code, Kommentar) gefuehrt und das Komma erst beim
|
||||
Zusammensetzen gesetzt - sonst landet es hinter dem "//" und ist Teil des
|
||||
Kommentars, womit die Parameterliste nicht mehr uebersetzt.
|
||||
"""
|
||||
tro_id = tro["id"]
|
||||
zeilen: list[tuple[str, str]] = [
|
||||
('nInMainTroNo := "cMainTro____"', TODO),
|
||||
('xInSftyOk := "DB_InterfaceSafety".stFromSafety.EStopMain.Q', TODO),
|
||||
("xInAllRdyToStart := TRUE", ""),
|
||||
(f'xInRelease := "DB_TRO_TEST".{tro_id}', ""),
|
||||
]
|
||||
if items.get("VarioDrive"):
|
||||
zeilen.append(('xInMotorProtection := "FC____"', TODO))
|
||||
zeilen.append(("xInAutomatic := #t_automatik", ""))
|
||||
zeilen.append(("stInOutMachineState := #stInOutControlUnitCabinet", ""))
|
||||
zeilen.append(("stInOutCognexInterface := #stCognexInterfaceDummy", ""))
|
||||
|
||||
# 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.
|
||||
for n, vor in enumerate(tro.get("predecessors", []), start=1):
|
||||
zeilen.append((f'arInOutJamEntr{n} := "DB_JamArea____".stJam____.arCarrier',
|
||||
f"{TODO} - Staubereich von {vor}"))
|
||||
zeilen.append((f'stInOutJamEntr{n} := "DB_JamArea____".stJam____.stData', ""))
|
||||
for n, nach in enumerate(tro.get("successors", []), start=2):
|
||||
zeilen.append((f'arInOutJamExit{n} := "DB_JamArea____".stJam____.arCarrier',
|
||||
f"{TODO} - Staubereich nach {nach}"))
|
||||
zeilen.append((f'stInOutJamExit{n} := "DB_JamArea____".stJam____.stData', ""))
|
||||
|
||||
for i in range(1, items.get("Separator", 1) + 1):
|
||||
zeilen.append((f'xOutStopper{i} => "MB____"', TODO))
|
||||
if items.get("VarioDrive"):
|
||||
zeilen.append(('xOutVarioMotor => "MA____"', TODO))
|
||||
for j in range(1, items.get("Switch", 0) + 1):
|
||||
zeilen.append((f'xOutSw{j}ExTo3 => "MB____"', TODO))
|
||||
zeilen.append((f'stInOutHMI := "DB_Interface_HMI".stTRO.{tro_id}', ""))
|
||||
zeilen.append(('nOutStateLast => "DB_GLOB_TroState".arTroState["cMainTro____"]',
|
||||
TODO))
|
||||
|
||||
kopf = f" #{tro_id}("
|
||||
einzug = " " * len(kopf)
|
||||
for index, (code, kommentar) in enumerate(zeilen):
|
||||
trenner = ");" if index == len(zeilen) - 1 else ","
|
||||
prefix = kopf if index == 0 else einzug
|
||||
text = f"{prefix}{code}{trenner}"
|
||||
add(f"{text} // {kommentar}" if kommentar else text)
|
||||
|
||||
|
||||
def emit_direction(add, tro: dict, anzahl: int) -> None:
|
||||
"""Weichenlogik - je Weiche ein FC_Direction-Aufruf."""
|
||||
add("")
|
||||
add(" REGION Direction")
|
||||
for j in range(1, anzahl + 1):
|
||||
ziel = (f"#{tro['id']}.nInDirection" if anzahl == 1
|
||||
else f"#{tro['id']}.nInDirection{j}")
|
||||
add(f' "FC_Direction"(nInSwitchNo := "cSwi_____", // {TODO}')
|
||||
add(f" nOutDirection => {ziel},")
|
||||
add(f' InOutCarrier := "DB_JamArea____".stJam____'
|
||||
f".arCarrier[1]);")
|
||||
if j < anzahl:
|
||||
add("")
|
||||
add(" END_REGION")
|
||||
|
||||
|
||||
def render(daten: dict, quelle: str, start_id: str = "") -> str:
|
||||
tros = daten["tros"]
|
||||
alle = {t["id"]: t for t in tros}
|
||||
geordnet, startgrund = flow_order(tros, start_id)
|
||||
|
||||
out: list[str] = []
|
||||
add = out.append
|
||||
|
||||
add("// ============================================================")
|
||||
add("// FB_Main - GERUEST, erzeugt von lib/scl_skeleton.py")
|
||||
add(f"// Quelle : {quelle}")
|
||||
add(f"// Erzeugt: {datetime.now().strftime('%Y-%m-%d %H:%M')}")
|
||||
add("//")
|
||||
add("// Das ist KEIN lauffaehiges Programm, sondern ein Geruest zum")
|
||||
add("// Ausfuellen. Struktur, Reihenfolge und Bauteilzahl stammen aus dem")
|
||||
add("// Anlagenlayout und der vor Ort aufgenommenen Topologie.")
|
||||
add("//")
|
||||
add(f"// Jede Stelle, die aus der E-Planung kommen muss, ist mit")
|
||||
add(f"// '{TODO}' markiert - danach suchen, dann ist nichts vergessen.")
|
||||
add("// Offen sind durchgaengig: Sensor-/Aktoradressen (cInBG/MB/MA),")
|
||||
add("// Separator- und Weichennummern, MainTro-Nummern, die Staubereiche")
|
||||
add("// (DB_JamArea...) und die Zeiten, soweit als 'pruefen' vermerkt.")
|
||||
add("//")
|
||||
add("// Reihenfolge der REGIONs = Materialfluss.")
|
||||
add(f"// Anfang bei {geordnet[0]['id']}: {startgrund}.")
|
||||
add("// ============================================================")
|
||||
add("")
|
||||
add('FUNCTION_BLOCK "FB_Main"')
|
||||
add("{ S7_Optimized_Access := 'TRUE' }")
|
||||
add("VERSION : 0.1")
|
||||
add(" VAR ")
|
||||
for tro in geordnet:
|
||||
fb = tro.get("fbType") or "FB_UNBEKANNT"
|
||||
kommentar = f" // {tro['type']}"
|
||||
if tro.get("separators"):
|
||||
kommentar += f", Sep {', '.join(tro['separators'])}"
|
||||
add(f' {tro["id"]} : "{fb}";{kommentar}')
|
||||
add(" END_VAR")
|
||||
add("")
|
||||
add("BEGIN")
|
||||
|
||||
for tro in geordnet:
|
||||
items = tro.get("items", {})
|
||||
n_sep = items.get("Separator", 1)
|
||||
n_swi = items.get("Switch", 0)
|
||||
n_scan = items.get("Scanner", 0)
|
||||
|
||||
add("")
|
||||
add(f"REGION {region_title(tro)}")
|
||||
pos = tro.get("position")
|
||||
if pos:
|
||||
add(f" // Anlagenkoordinate x/y: {pos['x']:.0f} / {pos['y']:.0f} mm")
|
||||
add(f" // Vorgaenger: {', '.join(tro.get('predecessors', [])) or '-'}")
|
||||
add(f" // Nachfolger: {', '.join(tro.get('successors', [])) or '-'}")
|
||||
if tro.get("confidence"):
|
||||
add(f" // Einstufung: {tro['confidence']}")
|
||||
add("")
|
||||
|
||||
emit_separators(add, tro, n_sep, hat_scanner=bool(n_scan))
|
||||
if n_swi:
|
||||
emit_switches(add, tro, n_swi, alle)
|
||||
if items.get("VarioDrive"):
|
||||
emit_vario(add, tro)
|
||||
if n_scan:
|
||||
emit_barcode(add, tro, n_scan)
|
||||
if items.get("StorageLine"):
|
||||
add(f" // {TODO}: Linienspeicher mit {items['StorageLine']} Linien - "
|
||||
f"Parametrierung je Linie")
|
||||
add("")
|
||||
|
||||
add(" #%s.stInPriorityManager.xReleaseOk := TRUE;" % tro["id"])
|
||||
add("")
|
||||
emit_call(add, tro, items)
|
||||
if n_swi:
|
||||
emit_direction(add, tro, n_swi)
|
||||
add("END_REGION")
|
||||
|
||||
add("")
|
||||
add("END_FUNCTION_BLOCK")
|
||||
add("")
|
||||
return "\n".join(out)
|
||||
|
||||
|
||||
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
|
||||
if not json_file.is_file():
|
||||
print(f"FEHLER: JSON nicht gefunden: {json_file}", file=sys.stderr)
|
||||
print(" Zuerst bin/tro_extract.bat laufen lassen.", file=sys.stderr)
|
||||
return 1
|
||||
|
||||
try:
|
||||
daten = load_tros(json_file)
|
||||
except (OSError, ValueError) as exc:
|
||||
print(f"FEHLER: {json_file} nicht lesbar: {exc}", file=sys.stderr)
|
||||
return 1
|
||||
if not daten.get("tros"):
|
||||
print(f"FEHLER: {json_file} enthaelt keine TROs.", file=sys.stderr)
|
||||
return 1
|
||||
|
||||
text = render(daten, json_file.name, args.start)
|
||||
out_file = results / (args.out or f"{json_file.stem.replace('_tro', '')}_FB_Main.scl")
|
||||
out_file.write_text(text, encoding="utf-8")
|
||||
|
||||
offen = text.count(TODO)
|
||||
geordnet, startgrund = flow_order(daten["tros"], args.start)
|
||||
print("")
|
||||
print("================================================================")
|
||||
print("FB_MAIN-GERUEST")
|
||||
print("================================================================")
|
||||
print(f"Quelle = {json_file}")
|
||||
print(f"TROs = {len(daten['tros'])}")
|
||||
print(f"Reihenfolge = {' -> '.join(t['id'] for t in geordnet)}")
|
||||
print(f"Anfang = {geordnet[0]['id']} ({startgrund})")
|
||||
print(f"Offene Stellen = {offen} (Marker '{TODO}')")
|
||||
print(f"SCL = {out_file}")
|
||||
print("================================================================")
|
||||
print("")
|
||||
return 0
|
||||
|
||||
|
||||
def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
|
||||
parser = argparse.ArgumentParser(
|
||||
prog="scl_skeleton",
|
||||
description="Erzeugt aus der extrahierten TRO-Liste ein FB_Main-Geruest "
|
||||
"als SCL zum Import in TIA Portal.",
|
||||
formatter_class=argparse.RawDescriptionHelpFormatter,
|
||||
epilog="Beispiele:\n"
|
||||
" scl_skeleton.bat\n"
|
||||
" scl_skeleton.bat --json export_tro.json --out FB_Main.scl\n"
|
||||
"\n"
|
||||
"Exit-Codes:\n"
|
||||
" 0 ok\n"
|
||||
" 1 Eingabe- oder Aufruffehler\n",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--json",
|
||||
default="export_tro.json",
|
||||
metavar="NAME",
|
||||
help="TRO-JSON aus tro_extract.py in %%SKEL_RESULTS%% "
|
||||
"(oder ein vollstaendiger Pfad). Standard: %(default)s",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--out",
|
||||
metavar="NAME",
|
||||
help="Name der SCL-Datei in %%SKEL_RESULTS%%. Standard: <Quelle>_FB_Main.scl",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--start",
|
||||
default="",
|
||||
metavar="TRO",
|
||||
help="TRO, bei dem die Reihenfolge beginnt. Nur noetig, solange der "
|
||||
"Fluss ein geschlossener Kreis ohne Be-/Entladestation ist.",
|
||||
)
|
||||
return parser.parse_args(argv)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
@@ -152,6 +152,9 @@ class TroSymbol:
|
||||
HEXAGON = "hexagon" # Sechseck
|
||||
RECT = "rect" # liegendes Rechteck
|
||||
ARROW = "arrow" # Pfeil/Fahne
|
||||
# Dreieck mit zweitem Dreieck innen - fuer 2Sep2Swi, sichtbar verwandt mit
|
||||
# TRIANGLE_DOWN (2Sep1Swi) und trotzdem davon unterscheidbar.
|
||||
DOUBLE_TRIANGLE_DOWN = "double_triangle_down"
|
||||
|
||||
@classmethod
|
||||
def all(cls) -> tuple[str, ...]:
|
||||
@@ -508,6 +511,20 @@ TRO_CATALOG: tuple[TroDefinition, ...] = (
|
||||
TroItem.SWITCH: 1,
|
||||
TroItem.SCANNER: 2,
|
||||
}, STYLE_SWI, TroSymbol.TRIANGLE_DOWN),
|
||||
# ACHTUNG - NICHT aus SCL belegt. Alle Typen oberhalb stammen aus
|
||||
# vorhandenem Code bzw. der Mapping-Tabelle in TRO_Typen.md; dieser hier
|
||||
# wurde am 2026-08-12 auf Anforderung aus dem Projekt 500573 (Mubea)
|
||||
# aufgenommen, um zwei Uebergabestellen am Beruehrpunkt Kreisel2/Kreisel1
|
||||
# zu einem Steuerobjekt zusammenzufassen.
|
||||
# Der Baustein FB_ILS_MTRO_2Sep2Swi existiert in diesem Repository nicht
|
||||
# und ist auch in doc/TRO_Katalog/scl_templates/ nicht hinterlegt - er muss
|
||||
# in der ILSLib angelegt werden, sonst erzeugt ein spaeterer
|
||||
# SCL-Generator einen Aufruf auf einen fehlenden Baustein.
|
||||
TroDefinition("2Sep2Swi", "FB_ILS_MTRO_2Sep2Swi", {
|
||||
TroItem.SEPARATOR: 2,
|
||||
TroItem.SWITCH: 2,
|
||||
TroItem.SCANNER: 2,
|
||||
}, STYLE_SWI, TroSymbol.DOUBLE_TRIANGLE_DOWN),
|
||||
)
|
||||
|
||||
_BY_NAME: dict[str, TroDefinition] = {t.get_name(): t for t in TRO_CATALOG}
|
||||
|
||||
@@ -0,0 +1,331 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
# Absichtlich ohne Shebang: der Windows-py-Launcher wuerde daraus "python3"
|
||||
# ableiten und scheitert, wenn keine PythonCore-Installation registriert ist.
|
||||
"""
|
||||
tro_extract.py - TROs aus der annotierten Zeichnung zurueck nach JSON lesen.
|
||||
|
||||
Gegenrichtung zu tro_annotate.py:
|
||||
|
||||
tro_annotate.py CSV + Ableitung -> Marker in der Zeichnung
|
||||
tro_extract.py Zeichnung -> JSON
|
||||
|
||||
Gelesen wird ausschliesslich die Zeichnung, nicht der CSV-Export. Damit
|
||||
kommt zurueck, was tatsaechlich in der Zeichnung steht - also auch das, was
|
||||
in BricsCAD von Hand nachgezogen wurde (Marker verschoben, Typ ueber TRO_EDIT
|
||||
geaendert). Genau dafuer ist der Weg gedacht: die Zeichnung ist nach dem
|
||||
Review der Stand der Dinge, nicht der Export.
|
||||
|
||||
Erkannt werden die Objekte, die tro_annotate.py mit seinen XDATA (AppID
|
||||
SPS_SKEL_TRO) gekennzeichnet hat:
|
||||
|
||||
Blockreferenz TRO_SYM_<Typ> ein TRO, Attribute ID/TYPE/ITEMS/
|
||||
CONFIDENCE/SEPARATORS (+ FB_BLOCK mit --fb)
|
||||
LWPOLYLINE "<von>-><nach>" ein Flusspfeil = eine Verbindung
|
||||
Kreis/Linien "OFFEN:<Name>" ein offener Punkt (noch kein TRO)
|
||||
|
||||
Die Koordinaten stehen doppelt in der Datei: `positionDxf` ist die
|
||||
Zeichnungskoordinate, `position` die Anlagenkoordinate. Umgerechnet wird mit
|
||||
dem Lagebezug aus <stem>_registration.json (von tro_annotate.py geschrieben);
|
||||
fehlt der, bleibt `position` leer statt geraten zu werden.
|
||||
|
||||
Aufruf ueber bin/tro_extract.bat bzw. bin/tro_extract.sh.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import re
|
||||
import sys
|
||||
from collections import Counter
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
|
||||
import ezdxf
|
||||
|
||||
from dxf_registration import Transform
|
||||
from material_flow import env_dir
|
||||
from tro_catalog import get_tro
|
||||
|
||||
APPID = "SPS_SKEL_TRO"
|
||||
BLOCK_PREFIX = "TRO_SYM_"
|
||||
OPEN_PREFIX = "OFFEN:"
|
||||
|
||||
# XDATA-Nutzlast eines Flusspfeils: "<von>-><nach>"
|
||||
EDGE_RE = re.compile(r"^(?P<src>[^>]+)->(?P<dst>.+)$")
|
||||
|
||||
# "1x SSCC, 1x Scanner, 1x Separator" -> {"SSCC": 1, "Scanner": 1, ...}
|
||||
ITEM_RE = re.compile(r"(?P<count>\d+)\s*x\s*(?P<item>.+)")
|
||||
|
||||
|
||||
def xdata_payload(entity) -> str | None:
|
||||
"""Zweiter String der XDATA von tro_annotate, sonst None.
|
||||
|
||||
Aufbau: (1000, "tro_annotate"), (1000, <Nutzlast>). Die Nutzlast ist je
|
||||
nach Objekt die TRO-ID, "<von>-><nach>" oder "OFFEN:<Name>".
|
||||
"""
|
||||
try:
|
||||
tags = entity.get_xdata(APPID)
|
||||
except (ezdxf.DXFValueError, ezdxf.DXFTypeError, ezdxf.DXFError):
|
||||
return None
|
||||
strings = [str(value) for code, value in tags if code == 1000]
|
||||
if len(strings) < 2 or strings[0] != "tro_annotate":
|
||||
return None
|
||||
return strings[1]
|
||||
|
||||
|
||||
def parse_items(text: str) -> dict[str, int]:
|
||||
"""Bauteilliste aus dem ITEMS-Attribut zurueckgewinnen."""
|
||||
items: dict[str, int] = {}
|
||||
for part in (text or "").split(","):
|
||||
match = ITEM_RE.match(part.strip())
|
||||
if match:
|
||||
items[match.group("item").strip()] = int(match.group("count"))
|
||||
return items
|
||||
|
||||
|
||||
def attrib(ref, tag: str) -> str:
|
||||
"""Attributwert einer Blockreferenz (leerer String, wenn nicht da)."""
|
||||
try:
|
||||
return (ref.get_attrib_text(tag) or "").strip()
|
||||
except Exception:
|
||||
return ""
|
||||
|
||||
|
||||
def load_transform(dxf_file: Path, results: Path, warnings: list[str]) -> Transform | None:
|
||||
"""Lagebezug zu dieser Zeichnung suchen.
|
||||
|
||||
tro_annotate.py benennt die Datei nach dem CSV-Stamm (export_registration
|
||||
.json), die Zeichnung heisst aber <stamm>_annotated.dxf - daher wird das
|
||||
Suffix "_annotated" zusaetzlich ohne Endung probiert. Zuletzt entscheidet
|
||||
der Inhalt: eine Registrierung, deren `dxf_file` auf diese Zeichnung passt,
|
||||
gilt auch dann, wenn der Dateiname anders lautet.
|
||||
"""
|
||||
stem = dxf_file.stem
|
||||
namen = [stem]
|
||||
if stem.endswith("_annotated"):
|
||||
namen.append(stem[: -len("_annotated")])
|
||||
|
||||
kandidaten = [results / f"{n}_registration.json" for n in namen]
|
||||
kandidaten += [dxf_file.with_name(f"{n}_registration.json") for n in namen]
|
||||
kandidaten += sorted(results.glob("*_registration.json"))
|
||||
|
||||
for candidate in kandidaten:
|
||||
if not candidate.is_file():
|
||||
continue
|
||||
try:
|
||||
roh = json.loads(candidate.read_text(encoding="utf-8"))
|
||||
transform = Transform(**roh)
|
||||
except (OSError, ValueError, TypeError) as exc:
|
||||
warnings.append(f"{candidate.name} nicht lesbar ({exc})")
|
||||
continue
|
||||
# Bei den aus dem Glob gefundenen Dateien pruefen, ob sie ueberhaupt
|
||||
# zu dieser Zeichnung gehoeren - sonst waere der Bezug frei erfunden.
|
||||
passt = candidate.stem.split("_registration")[0] in namen
|
||||
if not passt and Path(roh.get("dxf_file", "")).stem not in (stem, *namen):
|
||||
continue
|
||||
return transform
|
||||
|
||||
warnings.append(
|
||||
f"kein Lagebezug gefunden (gesucht: "
|
||||
f"{', '.join(n + '_registration.json' for n in namen)}) - "
|
||||
f"'position' bleibt leer, 'positionDxf' ist trotzdem gefuellt"
|
||||
)
|
||||
return None
|
||||
|
||||
|
||||
def extract(doc, transform: Transform | None, warnings: list[str]) -> dict:
|
||||
"""TROs, Verbindungen und offene Punkte aus dem Modelbereich lesen."""
|
||||
msp = doc.modelspace()
|
||||
tros: list[dict] = []
|
||||
edges: list[dict] = []
|
||||
open_points: list[dict] = []
|
||||
seen_edges: set[tuple[str, str]] = set()
|
||||
|
||||
for entity in msp:
|
||||
payload = xdata_payload(entity)
|
||||
if payload is None:
|
||||
continue
|
||||
|
||||
if entity.dxftype() == "INSERT" and entity.dxf.name.startswith(BLOCK_PREFIX):
|
||||
tro_id = attrib(entity, "ID") or payload
|
||||
type_name = attrib(entity, "TYPE") or entity.dxf.name[len(BLOCK_PREFIX):]
|
||||
insert = entity.dxf.insert
|
||||
eintrag = {
|
||||
"id": tro_id,
|
||||
"type": type_name,
|
||||
"items": parse_items(attrib(entity, "ITEMS")),
|
||||
"confidence": attrib(entity, "CONFIDENCE"),
|
||||
"separators": [s.strip() for s in attrib(entity, "SEPARATORS").split(",")
|
||||
if s.strip()],
|
||||
"positionDxf": {"x": round(insert.x, 2), "y": round(insert.y, 2),
|
||||
"z": round(insert.z, 2)},
|
||||
"layer": entity.dxf.layer,
|
||||
"block": entity.dxf.name,
|
||||
}
|
||||
# FB-Baustein: als Attribut nur vorhanden, wenn mit --fb annotiert
|
||||
# wurde. Sonst aus dem Katalog nachgeschlagen und als solches
|
||||
# gekennzeichnet - er steht dann nicht in der Zeichnung.
|
||||
fb = attrib(entity, "FB_BLOCK")
|
||||
if fb and fb != "-":
|
||||
eintrag["fbType"] = fb
|
||||
eintrag["fbTypeFrom"] = "dxf"
|
||||
else:
|
||||
definition = get_tro(type_name)
|
||||
if definition:
|
||||
eintrag["fbType"] = definition.get_fb_block()
|
||||
eintrag["fbTypeFrom"] = "tro_catalog"
|
||||
else:
|
||||
eintrag["fbType"] = None
|
||||
eintrag["fbTypeFrom"] = "unbekannter Typ"
|
||||
warnings.append(
|
||||
f"{tro_id}: Typ '{type_name}' steht nicht im TRO-Katalog"
|
||||
)
|
||||
if transform is not None:
|
||||
x, y = transform.to_csv(insert.x, insert.y)
|
||||
eintrag["position"] = {"x": round(x, 2), "y": round(y, 2)}
|
||||
tros.append(eintrag)
|
||||
continue
|
||||
|
||||
if payload.startswith(OPEN_PREFIX):
|
||||
name = payload[len(OPEN_PREFIX):]
|
||||
if entity.dxftype() != "CIRCLE":
|
||||
continue # Kreis, Kreuz und Text tragen dieselbe Nutzlast
|
||||
center = entity.dxf.center
|
||||
punkt = {"name": name,
|
||||
"positionDxf": {"x": round(center.x, 2), "y": round(center.y, 2)}}
|
||||
if transform is not None:
|
||||
x, y = transform.to_csv(center.x, center.y)
|
||||
punkt["position"] = {"x": round(x, 2), "y": round(y, 2)}
|
||||
open_points.append(punkt)
|
||||
continue
|
||||
|
||||
match = EDGE_RE.match(payload)
|
||||
if match:
|
||||
# Schaft und Spitze eines Pfeils tragen dieselbe Nutzlast
|
||||
pair = (match.group("src"), match.group("dst"))
|
||||
if pair not in seen_edges:
|
||||
seen_edges.add(pair)
|
||||
edges.append({"from": pair[0], "to": pair[1]})
|
||||
|
||||
tros.sort(key=lambda t: t["id"])
|
||||
edges.sort(key=lambda e: (e["from"], e["to"]))
|
||||
open_points.sort(key=lambda p: p["name"])
|
||||
|
||||
# Vorgaenger/Nachfolger je TRO - das ist die Frage, die man an so eine
|
||||
# Datei stellt ("was kommt vor diesem TRO, was danach").
|
||||
bekannt = {t["id"] for t in tros}
|
||||
for tro in tros:
|
||||
tro["predecessors"] = sorted(e["from"] for e in edges if e["to"] == tro["id"])
|
||||
tro["successors"] = sorted(e["to"] for e in edges if e["from"] == tro["id"])
|
||||
for edge in edges:
|
||||
for seite in ("from", "to"):
|
||||
if edge[seite] not in bekannt:
|
||||
warnings.append(
|
||||
f"Verbindung {edge['from']} -> {edge['to']}: '{edge[seite]}' "
|
||||
f"ist kein TRO in dieser Zeichnung"
|
||||
)
|
||||
return {"tros": tros, "connections": edges, "openPoints": open_points}
|
||||
|
||||
|
||||
def main(argv: list[str] | None = None) -> int:
|
||||
args = parse_args(argv)
|
||||
warnings: list[str] = []
|
||||
results = env_dir("SKEL_RESULTS", "results")
|
||||
|
||||
dxf_file = Path(args.dxf)
|
||||
if not dxf_file.is_absolute() and dxf_file.parent == Path("."):
|
||||
dxf_file = results / args.dxf
|
||||
if not dxf_file.is_file():
|
||||
print(f"FEHLER: Zeichnung nicht gefunden: {dxf_file}", file=sys.stderr)
|
||||
return 1
|
||||
|
||||
try:
|
||||
doc = ezdxf.readfile(str(dxf_file))
|
||||
except (OSError, ezdxf.DXFError) as exc:
|
||||
print(f"FEHLER: {dxf_file} nicht lesbar: {exc}", file=sys.stderr)
|
||||
return 1
|
||||
|
||||
transform = load_transform(dxf_file, results, warnings)
|
||||
daten = extract(doc, transform, warnings)
|
||||
|
||||
if not daten["tros"]:
|
||||
print(f"FEHLER: keine TRO-Marker in {dxf_file} gefunden. Wurde sie mit "
|
||||
f"tro_annotate.py erzeugt?", file=sys.stderr)
|
||||
return 2
|
||||
|
||||
kopf = {
|
||||
"schema": "sps_skel-tro-extract-v1",
|
||||
"source": dxf_file.name,
|
||||
"extracted": datetime.now().strftime("%Y-%m-%d %H:%M"),
|
||||
"note": "Aus der Zeichnung gelesen, nicht aus dem CSV-Export - "
|
||||
"Aenderungen aus BricsCAD sind enthalten.",
|
||||
}
|
||||
if transform is not None:
|
||||
kopf["transform"] = {"dx": round(transform.dx, 3), "dy": round(transform.dy, 3),
|
||||
"rotation": transform.rotation,
|
||||
"note": "Anlagenkoordinate = Zeichnung + (dx, dy)"}
|
||||
|
||||
out_file = results / (args.out or f"{dxf_file.stem}_tro.json")
|
||||
out_file.write_text(
|
||||
json.dumps({**kopf, **daten}, indent=2, ensure_ascii=False) + "\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
typen = Counter(t["type"] for t in daten["tros"])
|
||||
print("")
|
||||
print("================================================================")
|
||||
print("TRO-EXTRAKT")
|
||||
print("================================================================")
|
||||
print(f"Zeichnung = {dxf_file}")
|
||||
print(f"Lagebezug = "
|
||||
+ (transform.describe() if transform else "- (nur DXF-Koordinaten)"))
|
||||
print(f"TROs = {len(daten['tros'])} "
|
||||
f"({', '.join(f'{k}: {v}' for k, v in sorted(typen.items()))})")
|
||||
print(f"Verbindungen = {len(daten['connections'])}")
|
||||
print(f"Offene Punkte = {len(daten['openPoints'])}")
|
||||
print(f"JSON = {out_file}")
|
||||
print("================================================================")
|
||||
if warnings:
|
||||
print("")
|
||||
print(f"Hinweise ({len(warnings)}):")
|
||||
for warning in warnings:
|
||||
print(f" ! {warning}")
|
||||
print("")
|
||||
return 0
|
||||
|
||||
|
||||
def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
|
||||
parser = argparse.ArgumentParser(
|
||||
prog="tro_extract",
|
||||
description="Liest die von tro_annotate.py gesetzten TRO-Marker aus "
|
||||
"einer Zeichnung zurueck und schreibt sie als JSON.",
|
||||
formatter_class=argparse.RawDescriptionHelpFormatter,
|
||||
epilog="Beispiele:\n"
|
||||
" tro_extract.bat\n"
|
||||
" tro_extract.bat --dxf export_annotated.dxf\n"
|
||||
" tro_extract.bat --dxf export_annotated.dxf --out tros.json\n"
|
||||
"\n"
|
||||
"Exit-Codes:\n"
|
||||
" 0 ok\n"
|
||||
" 1 Eingabe- oder Aufruffehler\n"
|
||||
" 2 keine TRO-Marker in der Zeichnung\n",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--dxf",
|
||||
default="export_annotated.dxf",
|
||||
metavar="NAME",
|
||||
help="Annotierte Zeichnung in %%SKEL_RESULTS%% (oder ein vollstaendiger "
|
||||
"Pfad). Standard: %(default)s",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--out",
|
||||
metavar="NAME",
|
||||
help="Name der JSON-Datei in %%SKEL_RESULTS%%. "
|
||||
"Standard: <Zeichnung>_tro.json",
|
||||
)
|
||||
return parser.parse_args(argv)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
+266
-20
@@ -58,6 +58,7 @@ from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import csv
|
||||
import math
|
||||
import sys
|
||||
from dataclasses import dataclass, field
|
||||
from datetime import datetime
|
||||
@@ -69,6 +70,8 @@ from material_flow import (
|
||||
EDGE_UNRESOLVED,
|
||||
LANE_LEFT,
|
||||
LANE_RIGHT,
|
||||
ROTATION_CCW,
|
||||
ROTATION_CW,
|
||||
UNASSIGNED,
|
||||
Element,
|
||||
Graph,
|
||||
@@ -91,6 +94,14 @@ from tro_catalog import (
|
||||
get_tro,
|
||||
mermaid_classdefs,
|
||||
)
|
||||
from tro_overrides import (
|
||||
Overrides,
|
||||
apply_connections,
|
||||
apply_merges,
|
||||
apply_overrides,
|
||||
apply_splits,
|
||||
load_overrides,
|
||||
)
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Konstanten
|
||||
@@ -169,9 +180,13 @@ class Tro:
|
||||
class Analysis:
|
||||
tros: list[Tro] = field(default_factory=list)
|
||||
edges: list[tuple[str, str]] = field(default_factory=list)
|
||||
# (von, nach) -> Kantenart (transfer / umlauf / weiche / unbestimmt / normal)
|
||||
edge_kinds: dict[tuple[str, str], str] = field(default_factory=dict)
|
||||
isolated: list[Tro] = field(default_factory=list)
|
||||
unknown_types: list[str] = field(default_factory=list)
|
||||
findings: list[str] = field(default_factory=list)
|
||||
# Manuelle Korrekturen aus cfg/tro_overrides.ini (Typkorrekturen, offene Punkte)
|
||||
overrides: Overrides = field(default_factory=Overrides)
|
||||
|
||||
def by_id(self, tro_id: str) -> Tro | None:
|
||||
for tro in self.tros:
|
||||
@@ -228,7 +243,12 @@ def _switch_exits(hosts: list[str], graph: Graph) -> list[str]:
|
||||
)
|
||||
|
||||
|
||||
def derive_tros(graph: Graph, elements: list[Element], findings: list[str]) -> list[Tro]:
|
||||
def derive_tros(
|
||||
graph: Graph,
|
||||
elements: list[Element],
|
||||
findings: list[str],
|
||||
overrides: Overrides | None = None,
|
||||
) -> list[Tro]:
|
||||
"""TRO-Liste aus den Separatoren und ihren angeschlossenen Objekten bilden."""
|
||||
separators = sorted(
|
||||
(e for e in elements if e.kind == "Separator"), key=lambda e: e.teile_id
|
||||
@@ -376,12 +396,41 @@ def derive_tros(graph: Graph, elements: list[Element], findings: list[str]) -> l
|
||||
tro.reason = f"{tro.reason}; {len(exits)} abgehende Wege -> Weiche"
|
||||
tro.confidence = "niedrig"
|
||||
|
||||
# 4b) Manuell ergaenzte Uebergabestellen (cfg/tro_overrides.ini, [.split.]).
|
||||
# Vor der Nummernvergabe, damit die neuen TROs in derselben Reihenfolge
|
||||
# eine Nummer bekommen wie die abgeleiteten.
|
||||
if overrides is not None and overrides.splits:
|
||||
def _neuer_tro(spec, hosts, label):
|
||||
return Tro(
|
||||
tro_id="",
|
||||
type_name=spec.type_name,
|
||||
label=label,
|
||||
hosts=hosts,
|
||||
separators=list(spec.separators),
|
||||
scanners=list(spec.scanners),
|
||||
reason=spec.reason or "manuell ergaenzte Uebergabestelle",
|
||||
confidence=spec.confidence,
|
||||
)
|
||||
|
||||
apply_splits(tros, overrides, _neuer_tro)
|
||||
|
||||
# 4c) Zusammenfassungen - nach den Splits, damit auch ein frisch
|
||||
# herausgeloester TRO Teil einer Zusammenfassung sein kann.
|
||||
if overrides is not None and overrides.merges:
|
||||
apply_merges(tros, overrides)
|
||||
|
||||
# 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}"
|
||||
|
||||
# 5b) Manuelle Korrekturen aus cfg/tro_overrides.ini. Sie greifen erst hier,
|
||||
# weil sie ueber die TRO-ID adressiert sind, und noch vor Schritt 6, damit
|
||||
# ein manuell gesetzter Typ den SSCC-Hinweis nicht mehr ausloest.
|
||||
if overrides is not None:
|
||||
apply_overrides(tros, overrides)
|
||||
|
||||
# 6) Scanner am Separator: 1Sep_SSCC laesst sich mechanisch nicht erkennen
|
||||
for tro in tros:
|
||||
if tro.type_name == "1Sep" and tro.scanners:
|
||||
@@ -394,53 +443,129 @@ def derive_tros(graph: Graph, elements: list[Element], findings: list[str]) -> l
|
||||
return tros
|
||||
|
||||
|
||||
def link_tros(graph: Graph, tros: list[Tro]) -> list[tuple[str, str]]:
|
||||
def _tro_edge_kind(edge: Edge) -> str:
|
||||
"""Materialkante -> Kantenart des TRO-Flusses (fuer Einfaerbung)."""
|
||||
if edge.note == "Weiche":
|
||||
return "weiche"
|
||||
if edge.kind == EDGE_CIRCLE:
|
||||
return "umlauf"
|
||||
if edge.kind == EDGE_UNRESOLVED:
|
||||
return "unbestimmt"
|
||||
if edge.note in ("Einschleusung", "Ausschleusung"):
|
||||
return "transfer"
|
||||
return "normal"
|
||||
|
||||
|
||||
def _bahn_richtung(node) -> tuple[float, float] | None:
|
||||
"""Foerderrichtung einer Kreisel-Bahn als Einheitsvektor.
|
||||
|
||||
Der Kreisel ist eine geschlossene Schleife: auf der einen Bahn laeuft das
|
||||
Material hin, auf der anderen zurueck. Welche Richtung das ist, ergibt sich
|
||||
aus der Laengsachse ("Drehung") und dem Umlaufsinn ("Drehrichtung"):
|
||||
|
||||
UZS (im Uhrzeigersinn) Bahn R entgegen der Achse, Bahn L mit ihr
|
||||
GUZ (gegen den UZS) umgekehrt
|
||||
|
||||
Rueckgabe None, wenn eines der beiden Merkmale fehlt - dann bleibt es bei
|
||||
der ungerichteten "gleiche Bahn"-Notiz.
|
||||
"""
|
||||
element = node.element
|
||||
drehung = _as_float(_merkmal(element.merkmale, "Drehung"))
|
||||
umlauf = element.rotation
|
||||
if drehung is None or umlauf not in (ROTATION_CW, ROTATION_CCW):
|
||||
return None
|
||||
rad = math.radians(drehung)
|
||||
achse = (math.cos(rad), math.sin(rad))
|
||||
mit_achse = (node.lane == LANE_LEFT) == (umlauf == ROTATION_CW)
|
||||
return achse if mit_achse else (-achse[0], -achse[1])
|
||||
|
||||
|
||||
def link_tros(
|
||||
graph: Graph, tros: list[Tro]
|
||||
) -> tuple[list[tuple[str, str]], dict[tuple[str, str], str]]:
|
||||
"""
|
||||
TROs verknuepfen.
|
||||
|
||||
Der Materialfluss laeuft von Uebergabestelle zu Uebergabestelle, ggf. durch
|
||||
passive Zwischenknoten (Strecken/Bahnen ohne eigenen TRO). Solche Knoten
|
||||
werden daher durchlaufen ("kontrahiert"), belegte Knoten beenden den Pfad.
|
||||
Jede TRO-Kante bekommt zusaetzlich die Art der *ankommenden* Materialkante
|
||||
(transfer / umlauf / weiche / unbestimmt / normal).
|
||||
"""
|
||||
owners: dict[str, list[str]] = {}
|
||||
for tro in tros:
|
||||
for host in tro.hosts:
|
||||
owners.setdefault(host, []).append(tro.tro_id)
|
||||
|
||||
forward: dict[str, list[str]] = {}
|
||||
forward: dict[str, list[tuple[str, str]]] = {}
|
||||
for edge in graph.edges:
|
||||
forward.setdefault(edge.src, []).append(edge.dst)
|
||||
kind = _tro_edge_kind(edge)
|
||||
forward.setdefault(edge.src, []).append((edge.dst, kind))
|
||||
if edge.kind == EDGE_UNRESOLVED:
|
||||
# unbestimmte Richtung: in beide Richtungen begehbar
|
||||
forward.setdefault(edge.dst, []).append(edge.src)
|
||||
forward.setdefault(edge.dst, []).append((edge.src, kind))
|
||||
|
||||
edges: set[tuple[str, str]] = set()
|
||||
edge_kinds: dict[tuple[str, str], str] = {}
|
||||
for tro in tros:
|
||||
seen = set(tro.hosts)
|
||||
queue = [n for host in tro.hosts for n in forward.get(host, [])]
|
||||
queue = [(nxt, k) for host in tro.hosts for (nxt, k) in forward.get(host, [])]
|
||||
while queue:
|
||||
node = queue.pop()
|
||||
node, arriving = queue.pop()
|
||||
if node in seen:
|
||||
continue
|
||||
seen.add(node)
|
||||
if node in owners:
|
||||
for other in owners[node]:
|
||||
if other != tro.tro_id:
|
||||
edges.add((tro.tro_id, other))
|
||||
pair = (tro.tro_id, other)
|
||||
edges.add(pair)
|
||||
edge_kinds.setdefault(pair, arriving)
|
||||
continue
|
||||
queue.extend(forward.get(node, []))
|
||||
|
||||
# gleicher Host = physisch dieselbe Strecke/Bahn
|
||||
# Mehrere TROs auf derselben Kreisel-Bahn liegen HINTEREINANDER, nicht
|
||||
# nebeneinander: das Material laeuft die Bahn in Foerderrichtung ab. Sie
|
||||
# werden daher der Reihe nach verkettet statt nur als "gleiche Bahn"
|
||||
# markiert - sonst fehlt z. B. die Kante von der einen Sperre zur naechsten.
|
||||
verkettet: set[str] = set()
|
||||
for host, ids in owners.items():
|
||||
if len(ids) < 2:
|
||||
continue
|
||||
node = graph.nodes.get(host)
|
||||
if node is None or node.kind != "Kreisel" or not node.lane:
|
||||
continue
|
||||
richtung = _bahn_richtung(node)
|
||||
if richtung is None:
|
||||
continue
|
||||
auf_bahn = [t for t in tros if t.tro_id in ids and t.position]
|
||||
if len(auf_bahn) < 2:
|
||||
continue
|
||||
# Projektion auf die Foerderrichtung: kleinster Wert kommt zuerst
|
||||
auf_bahn.sort(key=lambda t: t.position[0] * richtung[0]
|
||||
+ t.position[1] * richtung[1])
|
||||
for vorher, nachher in zip(auf_bahn, auf_bahn[1:]):
|
||||
pair = (vorher.tro_id, nachher.tro_id)
|
||||
edges.add(pair)
|
||||
edge_kinds.setdefault(pair, "bahn")
|
||||
verkettet.update(pair)
|
||||
|
||||
# gleicher Host = physisch dieselbe Strecke/Bahn. Verkettete Bahnen sind
|
||||
# oben schon gerichtet verbunden und brauchen die ungerichtete Notiz nicht.
|
||||
for host, ids in owners.items():
|
||||
for tro_id in ids:
|
||||
tro = next(t for t in tros if t.tro_id == tro_id)
|
||||
tro.colocated = sorted(set(tro.colocated) | {o for o in ids if o != tro_id})
|
||||
if tro_id in verkettet:
|
||||
continue
|
||||
tro.colocated = sorted(
|
||||
set(tro.colocated) | {o for o in ids if o != tro_id and o not in verkettet}
|
||||
)
|
||||
|
||||
for tro in tros:
|
||||
tro.successors = sorted(dst for src, dst in edges if src == tro.tro_id)
|
||||
tro.predecessors = sorted(src for src, dst in edges if dst == tro.tro_id)
|
||||
|
||||
return sorted(edges)
|
||||
return sorted(edges), edge_kinds
|
||||
|
||||
|
||||
def _centroid(
|
||||
@@ -494,12 +619,36 @@ def locate_tros(tros: list[Tro], elements: list[Element], graph: Graph) -> None:
|
||||
tro.position_from = source if tro.position else "keine Position"
|
||||
|
||||
|
||||
def analyse_tros(graph: Graph, elements: list[Element]) -> Analysis:
|
||||
def analyse_tros(
|
||||
graph: Graph, elements: list[Element], overrides: Overrides | None = None
|
||||
) -> Analysis:
|
||||
"""TROs ermitteln, verknuepfen, verorten und gegen die Typdefinition pruefen."""
|
||||
analysis = Analysis()
|
||||
analysis.tros = derive_tros(graph, elements, analysis.findings)
|
||||
analysis.edges = link_tros(graph, analysis.tros)
|
||||
analysis.overrides = overrides if overrides is not None else Overrides()
|
||||
analysis.tros = derive_tros(graph, elements, analysis.findings, analysis.overrides)
|
||||
# Verorten vor dem Verknuepfen: die Reihenfolge mehrerer TROs auf einer
|
||||
# Kreisel-Bahn wird ueber ihre Koordinate bestimmt.
|
||||
locate_tros(analysis.tros, elements, graph)
|
||||
analysis.edges, analysis.edge_kinds = link_tros(graph, analysis.tros)
|
||||
|
||||
# Vor Ort aufgenommene Topologie schlaegt die abgeleitete. Die Differenz
|
||||
# wird protokolliert, damit sichtbar bleibt, was die Ableitung falsch hatte.
|
||||
if analysis.overrides.connections:
|
||||
abgeleitet = set(analysis.edges)
|
||||
analysis.edges, analysis.edge_kinds = apply_connections(
|
||||
analysis.tros, analysis.overrides
|
||||
)
|
||||
gesetzt = set(analysis.edges)
|
||||
for src, dst in sorted(abgeleitet - gesetzt):
|
||||
analysis.findings.append(
|
||||
f"abgeleitete Verbindung {src} -> {dst} entfaellt - steht nicht "
|
||||
f"in der gesetzten Topologie"
|
||||
)
|
||||
for src, dst in sorted(gesetzt - abgeleitet):
|
||||
analysis.findings.append(
|
||||
f"Verbindung {src} -> {dst} nur gesetzt, aus dem Layout nicht "
|
||||
f"ableitbar"
|
||||
)
|
||||
|
||||
for tro in analysis.tros:
|
||||
definition = tro.definition()
|
||||
@@ -552,6 +701,21 @@ def _legend_pos(analysis: Analysis) -> tuple[float, float]:
|
||||
return (min(xs) - LEGEND_MARGIN, min(ys) - LEGEND_MARGIN)
|
||||
|
||||
|
||||
# Kantenstil je Art (an material_flow.py angelehnt: Umlauf hellblau gestrichelt,
|
||||
# Weiche orange). "transfer"/"normal" bleiben der durchgezogene Standardfluss.
|
||||
TRO_EDGE_STYLE = {
|
||||
"umlauf": 'color="#8ea9db", style=dashed, tooltip="Kreisel-Umlauf"',
|
||||
"weiche": 'color="#cc8800", penwidth=1.8, xlabel="Weiche", fontcolor="#cc8800", '
|
||||
'tooltip="Weiche (Kreisel-Uebergang)"',
|
||||
"unbestimmt": 'color="#bf8f00", style=dashed, dir=both, tooltip="Richtung unbestimmt"',
|
||||
"transfer": 'tooltip="Uebergabe (Ein-/Ausschleusung)"',
|
||||
}
|
||||
|
||||
|
||||
def _tro_edge_attrs(kind: str | None) -> str:
|
||||
return TRO_EDGE_STYLE.get(kind or "normal", "")
|
||||
|
||||
|
||||
def render_dot(analysis: Analysis, source: str, use_coords: bool = False) -> str:
|
||||
"""
|
||||
Gerichtetes Flussdiagramm der TROs als Graphviz-Quelle.
|
||||
@@ -601,7 +765,7 @@ def render_dot(analysis: Analysis, source: str, use_coords: bool = False) -> str
|
||||
add("")
|
||||
|
||||
for src, dst in analysis.edges:
|
||||
add(f' "{src}" -> "{dst}" [];')
|
||||
add(f' "{src}" -> "{dst}" [{_tro_edge_attrs(analysis.edge_kinds.get((src, dst)))}];')
|
||||
|
||||
seen: set[frozenset] = set()
|
||||
for tro in analysis.tros:
|
||||
@@ -624,9 +788,13 @@ def render_dot(analysis: Analysis, source: str, use_coords: bool = False) -> str
|
||||
)
|
||||
legend_label = ('<<table border="0" cellborder="1" cellspacing="0" cellpadding="3">'
|
||||
+ rows
|
||||
+ '<tr><td align="left">durchgezogen</td>'
|
||||
'<td align="left">Materialfluss</td></tr>'
|
||||
'<tr><td align="left">gepunktet</td>'
|
||||
+ '<tr><td align="left"><font color="#2f5597">durchgezogen</font></td>'
|
||||
'<td align="left">Uebergabe / Ein-Ausschleusung</td></tr>'
|
||||
'<tr><td align="left"><font color="#8ea9db">blau gestrichelt</font></td>'
|
||||
'<td align="left">Kreisel-Umlauf</td></tr>'
|
||||
'<tr><td align="left"><font color="#cc8800">orange</font></td>'
|
||||
'<td align="left">Weiche (Kreisel-Uebergang)</td></tr>'
|
||||
'<tr><td align="left">grau gepunktet</td>'
|
||||
'<td align="left">gleiche Bahn / Strecke</td></tr>'
|
||||
"</table>>")
|
||||
if use_coords:
|
||||
@@ -787,6 +955,11 @@ def render_doc(
|
||||
add(f"| TRO-Verbindungen | {len(analysis.edges)} |")
|
||||
add(f"| Nicht verbundene TROs | {len(analysis.isolated)} |")
|
||||
add(f"| Hinweise | {len(analysis.findings)} |")
|
||||
if analysis.overrides.applied:
|
||||
add(f"| Manuelle Typkorrekturen | {len(analysis.overrides.applied)} |")
|
||||
if analysis.overrides.open_points:
|
||||
add(f"| Offene Punkte (noch kein TRO) | "
|
||||
f"{len(analysis.overrides.open_points)} |")
|
||||
add("")
|
||||
add("| Typ | FB-Baustein | Anzahl hier | Bauteile je TRO | Farbgruppe |")
|
||||
add("|---|---|---:|---|---|")
|
||||
@@ -873,7 +1046,56 @@ def render_doc(
|
||||
add("_Keine._")
|
||||
add("")
|
||||
|
||||
add("## 6. Ableitungsregeln und Typdefinition")
|
||||
add("## 6. Manuelle Korrekturen und offene Punkte")
|
||||
add("")
|
||||
overrides = analysis.overrides
|
||||
if not overrides and not overrides.warnings:
|
||||
add("_Keine._ Alles unten Stehende ist rein aus dem Layout abgeleitet.")
|
||||
add("")
|
||||
else:
|
||||
add(f"Quelle: `{overrides.source}`" if overrides.source
|
||||
else "_Keine Override-Datei gefunden._")
|
||||
add("")
|
||||
|
||||
if overrides.applied:
|
||||
add("### 6.1 Angewandte Typkorrekturen")
|
||||
add("")
|
||||
add("Diese Typen stammen **nicht** aus dem Layout, sondern aus der "
|
||||
"Begehung bzw. dem Review - im Layout sind sie nicht erkennbar.")
|
||||
add("")
|
||||
add("| Korrektur | Begruendung |")
|
||||
add("|---|---|")
|
||||
for entry in overrides.applied:
|
||||
head, _, tail = entry.partition(" (")
|
||||
add(f"| `{head}` | {tail[:-1] if tail.endswith(')') else tail} |")
|
||||
add("")
|
||||
|
||||
if overrides.open_points:
|
||||
add("### 6.2 Offene Punkte - noch kein TRO")
|
||||
add("")
|
||||
add("Im Review gemeldete Stellen. Die TRO-Struktur ist hier "
|
||||
"**bewusst unveraendert**; die Punkte sind nur markiert (auch in der "
|
||||
"annotierten Zeichnung, Layer `TRO_OPENPOINT`), damit sie in BricsCAD "
|
||||
"geprueft werden koennen.")
|
||||
add("")
|
||||
add("| Punkt | X [mm] | Y [mm] | Z [mm] | erwartet | heute Teil von | Anmerkung |")
|
||||
add("|---|---:|---:|---:|---|---|---|")
|
||||
for point in overrides.open_points:
|
||||
z_text = f"{point.z:.0f}" if point.z is not None else "-"
|
||||
expect = f"`{point.expect}`" if point.expect else "-"
|
||||
owner = f"`{point.belongs_to}`" if point.belongs_to else "-"
|
||||
add(f"| {point.name} | {point.x:.0f} | {point.y:.0f} | {z_text} "
|
||||
f"| {expect} | {owner} | {point.note or '-'} |")
|
||||
add("")
|
||||
|
||||
if overrides.warnings:
|
||||
add("### 6.3 Probleme mit der Override-Datei")
|
||||
add("")
|
||||
for warning in overrides.warnings:
|
||||
add(f"- {warning}")
|
||||
add("")
|
||||
|
||||
add("## 7. Ableitungsregeln und Typdefinition")
|
||||
add("")
|
||||
add("Ein TRO ist die Steuerungslogik **einer Uebergabestelle**. Ausgangspunkt "
|
||||
"ist der Separator und das, was an ihm haengt (Host-Objekt, Scanner).")
|
||||
@@ -974,6 +1196,29 @@ def report(
|
||||
print("FEHLER: TRO-Typen ohne Eintrag in TRO_CATALOG: "
|
||||
+ ", ".join(analysis.unknown_types))
|
||||
|
||||
overrides = analysis.overrides
|
||||
if overrides.applied:
|
||||
print("")
|
||||
print(f"Manuelle Typkorrekturen ({len(overrides.applied)}) aus "
|
||||
f"{overrides.source}:")
|
||||
for entry in overrides.applied:
|
||||
print(f" * {entry}")
|
||||
|
||||
if overrides.open_points:
|
||||
print("")
|
||||
print(f"Offene Punkte ({len(overrides.open_points)}) - nur markiert, "
|
||||
f"TRO-Struktur unveraendert:")
|
||||
for point in overrides.open_points:
|
||||
expect = f" erwartet {point.expect}" if point.expect else ""
|
||||
owner = f", heute Teil von {point.belongs_to}" if point.belongs_to else ""
|
||||
print(f" o {point.name} bei {point.describe_position()}{expect}{owner}")
|
||||
|
||||
if overrides.warnings:
|
||||
print("")
|
||||
print(f"Probleme mit {overrides.source} ({len(overrides.warnings)}):")
|
||||
for warning in overrides.warnings:
|
||||
print(f" ! {warning}")
|
||||
|
||||
if analysis.findings:
|
||||
print("")
|
||||
print(f"Hinweise ({len(analysis.findings)}):")
|
||||
@@ -1071,7 +1316,8 @@ def main(argv: list[str] | None = None) -> int:
|
||||
print(f"FEHLER: {csv_file} enthaelt keine Flussobjekte.", file=sys.stderr)
|
||||
return 1
|
||||
|
||||
analysis = analyse_tros(graph, elements)
|
||||
overrides = load_overrides(csv_file.name, env_dir("SKEL_CFG", "cfg"))
|
||||
analysis = analyse_tros(graph, elements, overrides)
|
||||
if not analysis.tros:
|
||||
print(f"FEHLER: {csv_file} enthaelt keine Separatoren - keine TROs "
|
||||
f"ableitbar.", file=sys.stderr)
|
||||
|
||||
@@ -0,0 +1,492 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
tro_overrides.py - manuelle Korrekturen an der abgeleiteten TRO-Liste.
|
||||
|
||||
`tro_flow.py` leitet die TROs allein aus dem mechanischen Layout (CSV-Export)
|
||||
ab. Manches steht dort aber nicht drin: eine Weiche, die im Export fehlt, ein
|
||||
Scanner, der den Typ auf `1Sep_SSCC` hebt, oder eine Stelle, die vor Ort als
|
||||
eigener TRO erkannt wurde. Solche Erkenntnisse kommen aus der Begehung bzw.
|
||||
aus dem Review der erzeugten Diagramme.
|
||||
|
||||
Damit sie den naechsten Lauf ueberleben, stehen sie in einer INI-Datei
|
||||
(`%SKEL_CFG%/tro_overrides.ini`) statt in den erzeugten Dateien. `tro_flow.py`
|
||||
und `tro_annotate.py` lesen sie und wenden sie nach der Ableitung an; die Doku
|
||||
weist jede angewandte Korrektur getrennt aus, damit abgeleitet und behauptet
|
||||
unterscheidbar bleiben.
|
||||
|
||||
Aufbau der INI (Sektionsnamen mit dem CSV-Dateinamen als Praefix, damit
|
||||
mehrere Anlagen in einer Datei stehen koennen):
|
||||
|
||||
[<csv>.tro.<TRO-ID>] Korrektur an einem abgeleiteten TRO
|
||||
type = 1Sep1Swi neuer Katalogtyp
|
||||
reason = ... Begruendung (kommt in die Doku)
|
||||
confidence = bestaetigt Vertrauensangabe (Default: bestaetigt)
|
||||
|
||||
[<csv>.openpoint.<Name>] offener Punkt - noch KEIN TRO
|
||||
x = 529.95 Anlagenkoordinate in mm
|
||||
y = -3895.27
|
||||
z = 1468
|
||||
expect = 1Sep1Swi hier erwarteter TRO-Typ
|
||||
belongs_to = TRO10 TRO, der das Bauteil heute enthaelt
|
||||
note = ... Erlaeuterung
|
||||
|
||||
Offene Punkte aendern die TRO-Struktur ausdruecklich **nicht**. Sie werden nur
|
||||
in der Doku und in der annotierten Zeichnung markiert, damit sie in BricsCAD
|
||||
geprueft werden koennen, bevor daraus ein TRO wird.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import configparser
|
||||
import os
|
||||
from dataclasses import dataclass, field
|
||||
from pathlib import Path
|
||||
|
||||
from tro_catalog import get_tro
|
||||
|
||||
OVERRIDE_FILE = "tro_overrides.ini"
|
||||
|
||||
# Vertrauensangabe fuer manuell gesetzte Typen. Bewusst ein eigener Wert - so
|
||||
# ist in der Doku sichtbar, dass der Typ nicht aus dem Layout stammt.
|
||||
DEFAULT_CONFIDENCE = "bestaetigt"
|
||||
|
||||
|
||||
@dataclass
|
||||
class TroOverride:
|
||||
"""Korrektur an einem abgeleiteten TRO.
|
||||
|
||||
Adressiert entweder ueber die TRO-ID (`TRO03`) oder - stabiler - ueber die
|
||||
Nummer eines seiner Separatoren (`SEP0011`). Die TRO-IDs sind laufende
|
||||
Nummern eines Laufs und verschieben sich, sobald TROs dazukommen oder
|
||||
wegfallen; die Separator-Nummer aus dem CSV-Export tut das nicht.
|
||||
"""
|
||||
|
||||
key: str
|
||||
by_separator: bool = False
|
||||
type_name: str = ""
|
||||
reason: str = ""
|
||||
confidence: str = ""
|
||||
|
||||
def matches(self, tro) -> bool:
|
||||
if self.by_separator:
|
||||
return self.key in tro.separators
|
||||
return self.key == tro.tro_id
|
||||
|
||||
|
||||
@dataclass
|
||||
class SplitSpec:
|
||||
"""Ein neuer TRO, herausgeloest aus einem bereits abgeleiteten.
|
||||
|
||||
Fuer Uebergabestellen, die die Ableitung nicht als eigenstaendig erkennt,
|
||||
weil die "Zuordnung" des Separators auf ein Transportobjekt zeigt (Strecke,
|
||||
Gefaellestrecke) statt auf die Kreisel-Bahn, an der er tatsaechlich
|
||||
schaltet. Die genannten Separatoren/Scanner werden ihrem bisherigen TRO
|
||||
entnommen und bilden einen neuen.
|
||||
"""
|
||||
|
||||
name: str
|
||||
type_name: str
|
||||
separators: list[str] = field(default_factory=list)
|
||||
scanners: list[str] = field(default_factory=list)
|
||||
host: str = ""
|
||||
label: str = ""
|
||||
reason: str = ""
|
||||
confidence: str = DEFAULT_CONFIDENCE
|
||||
|
||||
|
||||
@dataclass
|
||||
class ConnectionSpec:
|
||||
"""Eine vor Ort bestaetigte Verbindung zwischen zwei TROs."""
|
||||
|
||||
src: str
|
||||
dst: str
|
||||
kind: str = ""
|
||||
|
||||
|
||||
@dataclass
|
||||
class MergeSpec:
|
||||
"""Mehrere abgeleitete TROs zu einem Steuerobjekt zusammenfassen.
|
||||
|
||||
Fuer Stellen, die mechanisch aus mehreren Bauteilgruppen bestehen,
|
||||
steuerungstechnisch aber eine einzige Uebergabestelle sind - etwa zwei
|
||||
Sperren beiderseits eines Kreisel-Beruehrpunkts. Adressiert wird ueber die
|
||||
Separatoren, weil deren Nummern ueber Exporte hinweg stabiler sind als die
|
||||
laufenden TRO-Nummern.
|
||||
"""
|
||||
|
||||
name: str
|
||||
type_name: str
|
||||
separators: list[str] = field(default_factory=list)
|
||||
host: str = ""
|
||||
label: str = ""
|
||||
reason: str = ""
|
||||
confidence: str = DEFAULT_CONFIDENCE
|
||||
|
||||
|
||||
@dataclass
|
||||
class OpenPoint:
|
||||
"""Eine im Review gemeldete Stelle, die noch kein eigener TRO ist."""
|
||||
|
||||
name: str
|
||||
x: float
|
||||
y: float
|
||||
z: float | None = None
|
||||
expect: str = ""
|
||||
belongs_to: str = ""
|
||||
note: str = ""
|
||||
|
||||
def describe_position(self) -> str:
|
||||
return f"{self.x:.0f}, {self.y:.0f}"
|
||||
|
||||
|
||||
@dataclass
|
||||
class Overrides:
|
||||
"""Alle Korrekturen zu einer CSV-Datei."""
|
||||
|
||||
source: Path | None = None
|
||||
tros: dict[str, TroOverride] = field(default_factory=dict)
|
||||
splits: list[SplitSpec] = field(default_factory=list)
|
||||
merges: list[MergeSpec] = field(default_factory=list)
|
||||
# Gesetzte Topologie: ersetzt die abgeleiteten Kanten vollstaendig
|
||||
connections: list[ConnectionSpec] = field(default_factory=list)
|
||||
open_points: list[OpenPoint] = field(default_factory=list)
|
||||
# Protokoll der tatsaechlich angewandten Korrekturen (fuer die Doku)
|
||||
applied: list[str] = field(default_factory=list)
|
||||
warnings: list[str] = field(default_factory=list)
|
||||
|
||||
def __bool__(self) -> bool:
|
||||
return bool(self.tros or self.splits or self.merges or self.connections
|
||||
or self.open_points)
|
||||
|
||||
|
||||
def override_path(cfg_dir: Path | None = None) -> Path:
|
||||
"""Pfad zur Override-Datei: %SKEL_CFG%, sonst <projekt>/cfg."""
|
||||
if cfg_dir is not None:
|
||||
return cfg_dir / OVERRIDE_FILE
|
||||
env = os.environ.get("SKEL_CFG")
|
||||
if env:
|
||||
return Path(env) / OVERRIDE_FILE
|
||||
return Path(__file__).resolve().parent.parent / "cfg" / OVERRIDE_FILE
|
||||
|
||||
|
||||
def _as_float(raw: str, where: str, warnings: list[str]) -> float | None:
|
||||
try:
|
||||
return float(str(raw).strip())
|
||||
except (TypeError, ValueError):
|
||||
warnings.append(f"{where}: '{raw}' ist keine Zahl - Eintrag ignoriert")
|
||||
return None
|
||||
|
||||
|
||||
def load_overrides(csv_name: str, cfg_dir: Path | None = None) -> Overrides:
|
||||
"""Korrekturen zu `csv_name` laden (leeres Ergebnis, wenn es keine gibt)."""
|
||||
path = override_path(cfg_dir)
|
||||
result = Overrides()
|
||||
if not path.is_file():
|
||||
return result
|
||||
result.source = path
|
||||
|
||||
parser = configparser.ConfigParser(inline_comment_prefixes=(";",))
|
||||
# Schluessel sind Kleinschreibung, Sektionsnamen bleiben wie geschrieben.
|
||||
parser.read(path, encoding="utf-8")
|
||||
|
||||
prefix = csv_name.lower()
|
||||
for section in parser.sections():
|
||||
# [<csv>.connections] - gesetzte Topologie, eigene Form ohne <name>
|
||||
if section.lower() == f"{prefix}.connections":
|
||||
for key, raw in parser[section].items():
|
||||
pfeil, _, rest = str(raw).partition("|")
|
||||
src, sep, dst = (t.strip() for t in pfeil.partition("->"))
|
||||
if not sep or not src or not dst:
|
||||
result.warnings.append(
|
||||
f"[{section}] {key}: '{raw}' ist keine Verbindung "
|
||||
f"'<von> -> <nach>'"
|
||||
)
|
||||
continue
|
||||
result.connections.append(
|
||||
ConnectionSpec(src=src, dst=dst, kind=rest.strip())
|
||||
)
|
||||
continue
|
||||
|
||||
parts = section.split(".")
|
||||
# <csv>.<art>.<name> - der CSV-Name kann selbst einen Punkt enthalten
|
||||
if len(parts) < 3:
|
||||
continue
|
||||
kind, name = parts[-2].lower(), parts[-1]
|
||||
owner = ".".join(parts[:-2]).lower()
|
||||
if owner != prefix:
|
||||
continue
|
||||
values = parser[section]
|
||||
|
||||
if kind == "tro":
|
||||
type_name = values.get("type", "").strip()
|
||||
if type_name and get_tro(type_name) is None:
|
||||
result.warnings.append(
|
||||
f"[{section}]: Typ '{type_name}' steht nicht im TRO-Katalog "
|
||||
f"- Korrektur ignoriert"
|
||||
)
|
||||
continue
|
||||
key = name.upper()
|
||||
by_sep = key.startswith("SEP")
|
||||
result.tros[key] = TroOverride(
|
||||
key=key[3:] if by_sep else key,
|
||||
by_separator=by_sep,
|
||||
type_name=type_name,
|
||||
reason=values.get("reason", "").strip(),
|
||||
confidence=values.get("confidence", "").strip() or DEFAULT_CONFIDENCE,
|
||||
)
|
||||
elif kind == "split":
|
||||
type_name = values.get("type", "").strip()
|
||||
if get_tro(type_name) is None:
|
||||
result.warnings.append(
|
||||
f"[{section}]: Typ '{type_name}' steht nicht im TRO-Katalog "
|
||||
f"- neuer TRO nicht angelegt"
|
||||
)
|
||||
continue
|
||||
seps = [s.strip() for s in values.get("separators", "").split(",") if s.strip()]
|
||||
if not seps:
|
||||
result.warnings.append(
|
||||
f"[{section}]: 'separators' fehlt - ohne Separator kein TRO"
|
||||
)
|
||||
continue
|
||||
result.splits.append(
|
||||
SplitSpec(
|
||||
name=name,
|
||||
type_name=type_name,
|
||||
separators=seps,
|
||||
scanners=[s.strip() for s in values.get("scanners", "").split(",")
|
||||
if s.strip()],
|
||||
host=values.get("host", "").strip(),
|
||||
label=values.get("label", "").strip(),
|
||||
reason=values.get("reason", "").strip(),
|
||||
confidence=values.get("confidence", "").strip() or DEFAULT_CONFIDENCE,
|
||||
)
|
||||
)
|
||||
elif kind == "merge":
|
||||
type_name = values.get("type", "").strip()
|
||||
if get_tro(type_name) is None:
|
||||
result.warnings.append(
|
||||
f"[{section}]: Typ '{type_name}' steht nicht im TRO-Katalog "
|
||||
f"- Zusammenfassung nicht durchgefuehrt"
|
||||
)
|
||||
continue
|
||||
seps = [s.strip() for s in values.get("separators", "").split(",") if s.strip()]
|
||||
if len(seps) < 2:
|
||||
result.warnings.append(
|
||||
f"[{section}]: 'separators' braucht mindestens zwei Nummern"
|
||||
)
|
||||
continue
|
||||
result.merges.append(
|
||||
MergeSpec(
|
||||
name=name,
|
||||
type_name=type_name,
|
||||
separators=seps,
|
||||
host=values.get("host", "").strip(),
|
||||
label=values.get("label", "").strip(),
|
||||
reason=values.get("reason", "").strip(),
|
||||
confidence=values.get("confidence", "").strip() or DEFAULT_CONFIDENCE,
|
||||
)
|
||||
)
|
||||
elif kind == "openpoint":
|
||||
x = _as_float(values.get("x", ""), f"[{section}] x", result.warnings)
|
||||
y = _as_float(values.get("y", ""), f"[{section}] y", result.warnings)
|
||||
if x is None or y is None:
|
||||
continue
|
||||
z_raw = values.get("z", "").strip()
|
||||
expect = values.get("expect", "").strip()
|
||||
if expect and get_tro(expect) is None:
|
||||
result.warnings.append(
|
||||
f"[{section}]: erwarteter Typ '{expect}' steht nicht im "
|
||||
f"TRO-Katalog"
|
||||
)
|
||||
result.open_points.append(
|
||||
OpenPoint(
|
||||
name=name,
|
||||
x=x,
|
||||
y=y,
|
||||
z=_as_float(z_raw, f"[{section}] z", result.warnings)
|
||||
if z_raw
|
||||
else None,
|
||||
expect=expect,
|
||||
belongs_to=values.get("belongs_to", "").strip(),
|
||||
note=values.get("note", "").strip(),
|
||||
)
|
||||
)
|
||||
else:
|
||||
result.warnings.append(
|
||||
f"[{section}]: unbekannte Art '{kind}' - erwartet 'tro' oder "
|
||||
f"'openpoint'"
|
||||
)
|
||||
|
||||
result.open_points.sort(key=lambda p: p.name)
|
||||
return result
|
||||
|
||||
|
||||
def apply_splits(tros, overrides: Overrides, make_tro) -> list:
|
||||
"""Neue TROs aus bestehenden herausloesen (vor der Nummernvergabe).
|
||||
|
||||
`make_tro(spec, hosts, label)` erzeugt das projektspezifische Tro-Objekt -
|
||||
so bleibt dieses Modul frei von einem Import aus tro_flow (das umgekehrt
|
||||
schon von hier importiert).
|
||||
"""
|
||||
for spec in overrides.splits:
|
||||
quellen = [t for t in tros if set(spec.separators) & set(t.separators)]
|
||||
if not quellen:
|
||||
overrides.warnings.append(
|
||||
f"Split '{spec.name}': keiner der Separatoren "
|
||||
f"{', '.join(spec.separators)} kommt in diesem Lauf vor - "
|
||||
f"nicht angelegt"
|
||||
)
|
||||
continue
|
||||
|
||||
hosts: list[str] = []
|
||||
for quelle in quellen:
|
||||
quelle.separators = [s for s in quelle.separators
|
||||
if s not in spec.separators]
|
||||
quelle.scanners = [s for s in quelle.scanners if s not in spec.scanners]
|
||||
hosts.extend(quelle.hosts)
|
||||
if not quelle.separators and quelle.type_name not in ("Vario",):
|
||||
overrides.warnings.append(
|
||||
f"Split '{spec.name}': {quelle.label} hat danach keinen "
|
||||
f"Separator mehr - bitte pruefen"
|
||||
)
|
||||
|
||||
ziel_hosts = [spec.host] if spec.host else sorted(set(hosts))[:1]
|
||||
label = spec.label or (
|
||||
f"Separator {', '.join(spec.separators)} "
|
||||
f"(manuell aus {quellen[0].label} herausgeloest)"
|
||||
)
|
||||
tros.append(make_tro(spec, ziel_hosts, label))
|
||||
overrides.applied.append(
|
||||
f"neu {spec.name}: {spec.type_name} auf {', '.join(ziel_hosts)} "
|
||||
f"mit Separator {', '.join(spec.separators)}, herausgeloest aus "
|
||||
f"{', '.join(sorted({q.label for q in quellen}))} "
|
||||
f"({spec.reason or 'manuelle Ergaenzung'})"
|
||||
)
|
||||
return tros
|
||||
|
||||
|
||||
def apply_merges(tros, overrides: Overrides) -> list:
|
||||
"""Mehrere TROs zu einem zusammenfassen (nach den Splits, vor der Nummer).
|
||||
|
||||
Der erste betroffene TRO bleibt bestehen und uebernimmt Bauteile, Hosts und
|
||||
Typ; die uebrigen werden aus der Liste entfernt.
|
||||
"""
|
||||
for spec in overrides.merges:
|
||||
betroffen = [t for t in tros if set(spec.separators) & set(t.separators)]
|
||||
if len(betroffen) < 2:
|
||||
gefunden = ", ".join(t.tro_id or t.label for t in betroffen) or "keiner"
|
||||
overrides.warnings.append(
|
||||
f"Merge '{spec.name}': nur {len(betroffen)} der Separatoren "
|
||||
f"{', '.join(spec.separators)} gefunden ({gefunden}) - "
|
||||
f"nicht zusammengefasst"
|
||||
)
|
||||
continue
|
||||
|
||||
ziel = betroffen[0]
|
||||
quellen = betroffen[1:]
|
||||
for quelle in quellen:
|
||||
ziel.separators = sorted(set(ziel.separators) | set(quelle.separators))
|
||||
ziel.scanners = sorted(set(ziel.scanners) | set(quelle.scanners))
|
||||
ziel.hosts = sorted(set(ziel.hosts) | set(quelle.hosts))
|
||||
tros.remove(quelle)
|
||||
|
||||
if spec.host:
|
||||
# Ein Host bestimmt, wo der zusammengefasste TRO im Fluss haengt;
|
||||
# die uebrigen Bahnen bleiben als weitere Hosts erhalten, damit die
|
||||
# Verknuepfung beide Seiten des Uebergangs sieht.
|
||||
ziel.hosts = [spec.host] + [h for h in ziel.hosts if h != spec.host]
|
||||
ziel.type_name = spec.type_name
|
||||
ziel.confidence = spec.confidence
|
||||
ziel.label = spec.label or ziel.label
|
||||
grund = spec.reason or "manuell zusammengefasst"
|
||||
ziel.reason = f"{ziel.reason}; manuell zusammengefasst ({grund})"
|
||||
overrides.applied.append(
|
||||
f"merge {spec.name}: {len(betroffen)} TROs -> ein {spec.type_name} "
|
||||
f"mit Separatoren {', '.join(ziel.separators)} ({grund})"
|
||||
)
|
||||
return tros
|
||||
|
||||
|
||||
def apply_connections(tros, overrides: Overrides):
|
||||
"""Gesetzte Topologie an die Stelle der abgeleiteten Kanten setzen.
|
||||
|
||||
Ersetzt vollstaendig, nicht ergaenzend: wenn die Verbindungen vor Ort
|
||||
aufgenommen wurden, sind die abgeleiteten damit erledigt - sonst blieben
|
||||
genau die Kanten stehen, die korrigiert werden sollten. Was dabei wegfaellt
|
||||
bzw. hinzukommt, steht anschliessend in overrides.applied.
|
||||
|
||||
Rueckgabe: (edges, edge_kinds) wie von link_tros.
|
||||
"""
|
||||
bekannt = {t.tro_id for t in tros}
|
||||
edges: list[tuple[str, str]] = []
|
||||
kinds: dict[tuple[str, str], str] = {}
|
||||
for spec in overrides.connections:
|
||||
for seite, wert in (("von", spec.src), ("nach", spec.dst)):
|
||||
if wert not in bekannt:
|
||||
overrides.warnings.append(
|
||||
f"Verbindung {spec.src} -> {spec.dst}: {seite} '{wert}' ist "
|
||||
f"kein TRO dieses Laufs - Verbindung ignoriert"
|
||||
)
|
||||
break
|
||||
else:
|
||||
paar = (spec.src, spec.dst)
|
||||
if paar in kinds:
|
||||
continue
|
||||
edges.append(paar)
|
||||
kinds[paar] = spec.kind or "gesetzt"
|
||||
|
||||
for tro in tros:
|
||||
tro.successors = sorted(d for s, d in edges if s == tro.tro_id)
|
||||
tro.predecessors = sorted(s for s, d in edges if d == tro.tro_id)
|
||||
tro.colocated = []
|
||||
|
||||
ohne = [t.tro_id for t in tros if not t.successors and not t.predecessors]
|
||||
if ohne:
|
||||
overrides.warnings.append(
|
||||
"ohne Verbindung in der gesetzten Topologie: " + ", ".join(sorted(ohne))
|
||||
)
|
||||
overrides.applied.append(
|
||||
f"Topologie gesetzt: {len(edges)} Verbindungen aus der Override-Datei "
|
||||
f"ersetzen die abgeleiteten"
|
||||
)
|
||||
return sorted(edges), kinds
|
||||
|
||||
|
||||
def apply_overrides(tros, overrides: Overrides) -> None:
|
||||
"""Typkorrekturen auf die abgeleitete TRO-Liste anwenden.
|
||||
|
||||
Wird aufgerufen, nachdem die TRO-IDs vergeben sind (die Korrekturen sind
|
||||
ueber die ID adressiert) und bevor FB-Baustein und Bauteilliste aus dem
|
||||
Katalog gezogen werden - der Typwechsel zieht beides mit.
|
||||
"""
|
||||
if not overrides.tros:
|
||||
return
|
||||
for section_key, override in sorted(overrides.tros.items()):
|
||||
treffer = [t for t in tros if override.matches(t)]
|
||||
if not treffer:
|
||||
wie = "Separator" if override.by_separator else "TRO-ID"
|
||||
overrides.warnings.append(
|
||||
f"[{section_key}]: kein TRO mit {wie} '{override.key}' in diesem "
|
||||
f"Lauf - Korrektur nicht angewandt"
|
||||
+ ("" if override.by_separator else
|
||||
" (Nummerierung geaendert? Besser ueber SEP<Nummer> adressieren)")
|
||||
)
|
||||
continue
|
||||
if len(treffer) > 1:
|
||||
overrides.warnings.append(
|
||||
f"[{section_key}]: trifft {len(treffer)} TROs "
|
||||
f"({', '.join(t.tro_id for t in treffer)}) - auf alle angewandt"
|
||||
)
|
||||
for tro in treffer:
|
||||
if not override.type_name or override.type_name == tro.type_name:
|
||||
continue
|
||||
vorher = tro.type_name
|
||||
tro.type_name = override.type_name
|
||||
tro.confidence = override.confidence
|
||||
grund = override.reason or "manuelle Korrektur"
|
||||
tro.reason = (f"{tro.reason}; manuell auf {override.type_name} "
|
||||
f"gesetzt ({grund})")
|
||||
overrides.applied.append(
|
||||
f"{tro.tro_id}: {vorher} -> {override.type_name} ({grund})"
|
||||
)
|
||||
Reference in New Issue
Block a user