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

Materialfluss: --graphtype fuer --chain-graph (coarsebox/compound/subgraph)

Browse Source 
--chain-graph zeichnete bisher immer nur die groben Objekte (Kreisel-Bahnen,
Strecke/Gefaellestrecke). Neuer Schalter --graphtype waehlt die Methode:
coarsebox (bisheriges Verhalten, unveraendert), compound (dieselbe
massstabsgetreue SVG, zusaetzlich die Anbauteile - Separator, Scanner,
Ausschleus-/Einschleuselement, Weiche - verschachtelt im Container an ihrer
echten Position, verbunden ueber die bereits aufgeloesten TrackGraph-Kanten)
und subgraph (dieselbe Verschachtelung als geschachtelte Graphviz-Cluster,
schematisch statt lagerichtig, <name>_chain.dot / mit --tosvg auch .svg).

_container_children() liest die Anbauteile direkt aus Element.track_ids -
derselben Quelle, die build_track_graph()/--track schon nutzt - statt sie wie
im Prototyp data/gen_chain_svg.py erneut aus der Nachbarn-Spalte herzuleiten.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
Samer Ayadi
2026-09-22 16:26:02 +02:00
co-authored by Claude Sonnet 5
parent b197bf8150
commit 0477f05180
2 changed files with 300 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/material_flow.py
+296 -17
View File
@@ -2479,6 +2479,45 @@ def _kreisel_lane_rects(
return {lane_a: half_a, lane_b: half_b}
def _lane_of_point(pt: tuple[float, float], lane_rects: dict[str, tuple[float, float, float, float]]) -> str:
"""Fallback-Bahnzuordnung ueber die Geometrie (naechstes Rechteck), falls ein
Feinelement keine auswertbare '<Kreisel>-L'/'-R'-Zuordnung traegt (siehe
_container_children)."""
px, py = pt
def dist(rect: tuple[float, float, float, float]) -> float:
x0, y0, x1, y1 = rect
cx, cy = min(max(px, x0), x1), min(max(py, y0), y1)
return (cx - px) ** 2 + (cy - py) ** 2
return min(lane_rects, key=lambda lane: dist(lane_rects[lane]))
def _container_children(element: Element, by_id: dict[str, Element]) -> list[Element]:
"""
Anbauteile (Separator, Scanner, Ausschleus-/Einschleuselement, Weiche) eines
Containers (Kreisel/Strecke/Gefaellestrecke) aus seiner TrackIds-Kette, in
Reihenfolge - fuer die verschachtelte Darstellung in --graphtype compound/
subgraph. Nutzt dieselbe Quelle (Element.track_ids) und dieselbe
Richtungs-Regel fuer Strecke/Gefaellestrecke wie build_track_graph() Schritt 1
(Kreisel-seitiges Element zuerst umdrehen), statt Nachbarn-Heuristiken wie im
Prototyp data/gen_chain_svg.py neu zu erfinden.
"""
chain = list(element.track_ids)
if element.is_transport and chain:
first_el, last_el = by_id.get(chain[0]), by_id.get(chain[-1])
first_kind = first_el.kind if first_el else ""
last_kind = last_el.kind if last_el else ""
if first_kind == "Einschleus" or last_kind == "Ausschleus":
chain = list(reversed(chain))
children: list[Element] = []
for tid in chain:
el = by_id.get(tid)
if el is not None and el.kind in TRACK_KINDS:
children.append(el)
return children
# ---------------------------------------------------------------------------
# Chain-/Lageplan-SVG der groben Objekte erzeugen (--chain-graph)
# ---------------------------------------------------------------------------
@@ -2506,7 +2545,9 @@ def _chain_marker_id(color: str) -> str:
return "chainArrow_" + color.lstrip("#")
def render_chain_svg(elements: list[Element], graph: Graph, source: str) -> str:
def render_chain_svg(
elements: list[Element], graph: Graph, source: str, track: TrackGraph | None = None
) -> str:
"""
Massstabsgetreuer Chain-/Lageplan der groben Objekte (--chain-graph): jeder
Kreisel als seine beiden Bahnen L/R (_kreisel_lane_rects(), Zuordnung aus
@@ -2515,6 +2556,16 @@ def render_chain_svg(elements: list[Element], graph: Graph, source: str) -> str:
Boundingbox. Kanten kommen direkt aus dem schon aufgeloesten groben
Materialfluss-Graphen (graph.edges) - eine Kante, deren Knoten hier nicht
gezeichnet wird (z. B. eine BTMT-Station), wird schlicht uebersprungen.
track nur fuer --graphtype compound (siehe render_chain_compound_svg): mit
angegebenem TrackGraph werden zusaetzlich die Anbauteile (Separator,
Scanner, Ausschleus-/Einschleuselement, Weiche) jedes Containers an
ihrer echten Position hinein verschachtelt gezeichnet, verbunden mit
den schon aufgeloesten Kanten aus track.edges (_container_children()
liefert die Reihenfolge aus derselben TrackIds-Quelle, die auch
build_track_graph() benutzt - keine eigene Nachbarn-Herleitung wie im
Prototyp data/gen_chain_svg.py). None (Default) = reines --graphtype
coarsebox ohne Anbauteile, wie bisher.
"""
rects: dict[str, tuple[float, float, float, float]] = {}
lane_info: dict[str, tuple[Element, str]] = {}
@@ -2629,6 +2680,58 @@ def render_chain_svg(elements: list[Element], graph: Graph, source: str) -> str:
f'fill="{style["color"]}">{_xml_escape(label)}</text>'
)
nested_svg: list[str] = []
if track is not None:
by_id = {e.teile_id: e for e in elements}
kreisel_lane_rects: dict[str, dict[str, tuple[float, float, float, float]]] = {}
for node_id, rect in rects.items():
if node_id in lane_info:
kreisel, lane = lane_info[node_id]
lane_rects = kreisel_lane_rects.setdefault(
kreisel.teile_id, _kreisel_lane_rects(kreisel, elements)
)
children = []
for el in _container_children(kreisel, by_id):
if el.x is None or el.y is None:
continue
base, _, el_lane = el.assignment.partition("-")
if base != kreisel.teile_id or el_lane not in (LANE_LEFT, LANE_RIGHT):
el_lane = _lane_of_point((el.x, el.y), lane_rects)
if el_lane == lane:
children.append(el)
else:
strecke = strecke_labels[node_id]
children = [
el for el in _container_children(strecke, by_id)
if el.x is not None and el.y is not None
]
if not children:
continue
rx0, ry0, rx1, ry1 = rect
child_ids = {el.teile_id for el in children}
edge_color = CHAIN_EDGE_COLOR[EDGE_CIRCLE if node_id in lane_info else EDGE_FLOW]
for te in track.edges:
if te.src not in child_ids or te.dst not in child_ids:
continue
src_el, dst_el = by_id[te.src], by_id[te.dst]
x1, y1 = sx(min(max(src_el.x, rx0), rx1)), sy(min(max(src_el.y, ry0), ry1))
x2, y2 = sx(min(max(dst_el.x, rx0), rx1)), sy(min(max(dst_el.y, ry0), ry1))
nested_svg.append(
f'<line x1="{x1:.1f}" y1="{y1:.1f}" x2="{x2:.1f}" y2="{y2:.1f}" '
f'stroke="{edge_color}" stroke-width="1"/>'
)
for el in children:
style = TRACK_NODE_STYLES.get(el.kind, DEFAULT_NODE_STYLE)
cx, cy = min(max(el.x, rx0), rx1), min(max(el.y, ry0), ry1)
px, py = sx(cx), sy(cy)
title = f"{el.kind} {el.name or el.teile_id} [{el.teile_id}]"
nested_svg.append(
f'<circle cx="{px:.1f}" cy="{py:.1f}" r="3.4" fill="{style["fillcolor"]}" '
f'stroke="{style["color"]}" stroke-width="1.2"><title>{_xml_escape(title)}</title></circle>'
)
heading = f"{Path(source).stem} - Materialfluss (grobe Objekte)"
out: list[str] = []
@@ -2655,10 +2758,142 @@ def render_chain_svg(elements: list[Element], graph: Graph, source: str) -> str:
f'Massstabsgetreu &#8226; Quelle: {_xml_escape(source)}</text>')
out.extend(box_svg)
out.extend(edge_svg)
out.extend(nested_svg)
add("</svg>")
return "\n".join(out) + "\n"
def render_chain_compound_svg(
elements: list[Element], graph: Graph, track: TrackGraph, source: str
) -> str:
"""--graphtype compound: wie render_chain_svg(), aber mit den Anbauteilen
(Separator, Scanner, Ausschleus-/Einschleuselement, Weiche) verschachtelt in
ihrem Container gezeichnet. Duenner Wrapper um render_chain_svg(track=...),
damit Rechteck-/Massstab-/Kanten-Aufbau nicht doppelt existiert."""
return render_chain_svg(elements, graph, source, track=track)
# ---------------------------------------------------------------------------
# Chain-/Lageplan als verschachtelte Graphviz-Cluster (--graphtype subgraph)
# ---------------------------------------------------------------------------
#
# Schematische (nicht massstabsgetreue) Alternative zu render_chain_svg(track=...):
# dieselbe Containment - Kreisel/Bahn/Strecke enthaelt seine Anbauteile - aber als
# verschachtelte Graphviz "subgraph cluster_*"-Bloecke statt handgebauter SVG-
# Rechtecke, gerendert von Graphviz' eigenem Layout (dot). Nuetzlich, wenn die
# Verschachtelung selbst interessiert und nicht die reale Geometrie.
CHAIN_SUBGRAPH_EDGE = {
EDGE_FLOW: '[color="#2f5597", penwidth=1.4]',
EDGE_CIRCLE: '[color="#8ea9db", style=dashed, penwidth=1.2, constraint=false]',
EDGE_UNRESOLVED: '[color="#bf8f00", style=dashed, dir=both]',
}
def _chain_child_dot(node_id: str, element: Element) -> str:
style = TRACK_NODE_STYLES.get(element.kind, DEFAULT_NODE_STYLE)
attrs = ", ".join(f'{key}="{value}"' for key, value in style.items())
label = _dot_label([f"{element.name or element.teile_id}", f"{element.kind} [{element.teile_id}]"])
return f' "{node_id}" [label="{label}", {attrs}];'
def render_chain_subgraph_dot(
elements: list[Element], graph: Graph, track: TrackGraph, source: str
) -> str:
"""
--graphtype subgraph: dieselbe Containment wie render_chain_compound_svg(),
aber als verschachtelte Graphviz-Cluster (Kreisel-Cluster mit einem
Bahn-L/R-Cluster je Bahn darin, Strecke/Gefaellestrecke-Cluster) statt
massstabsgetreuer SVG. Rendern mit dot -Tsvg (siehe dot_to_svg()) - NICHT
lagerichtig, das Layout kommt von Graphviz.
"""
by_id = {e.teile_id: e for e in elements}
out: list[str] = []
add = out.append
add("// Automatisch erzeugt von lib/material_flow.py - nicht manuell aendern.")
add(f"// Quelle: {source}")
add("// --chain-graph --graphtype subgraph: verschachtelte Cluster, nicht lagerichtig.")
add("digraph Materialfluss_Chain {")
add(' graph [rankdir=LR, splines=spline, nodesep=0.22, ranksep=1.2,')
add(' fontname="Segoe UI", fontsize=11, labelloc="t",')
add(f' label="Chain-Graph (verschachtelt) - {_dot_escape(source)}"];')
add(' node [shape=box, style="rounded,filled", fontname="Segoe UI", fontsize=8];')
add(' edge [fontname="Segoe UI", fontsize=8, arrowsize=0.8];')
add("")
top_ids: set[str] = set()
for index, element in enumerate(
sorted((e for e in elements if e.is_circle and e.position and e.bbox), key=lambda e: e.teile_id)
):
add(f" subgraph cluster_kreisel_{index} {{")
add(f' label="{_dot_escape(element.kind)} {_dot_escape(element.name or element.teile_id)} '
f'[{element.teile_id}]"; style="rounded,filled"; fillcolor="#eef2fa"; color="#2f5597";')
add(' fontname="Segoe UI"; fontsize=10; fontcolor="#1f3864"; margin=16;')
lane_rects = _kreisel_lane_rects(element, elements)
for lane in sorted(lane_rects):
lane_id = f"{element.teile_id}-{lane}"
top_ids.add(lane_id)
add(f' subgraph "cluster_{lane_id}" {{')
add(f' label="Bahn {lane}"; style="rounded,filled"; fillcolor="#ffffff"; '
f'color="#8ea9db"; fontname="Segoe UI"; fontsize=9; fontcolor="#375623"; margin=10;')
add(f' "{lane_id}" [label="Bahn {lane}", shape=box, style="rounded,filled", '
f'fillcolor="#2f5597", fontcolor="#ffffff", color="#1f3864"];')
children = []
for el in _container_children(element, by_id):
if el.x is None or el.y is None:
continue
base, _, el_lane = el.assignment.partition("-")
if base != element.teile_id or el_lane not in (LANE_LEFT, LANE_RIGHT):
el_lane = _lane_of_point((el.x, el.y), lane_rects)
if el_lane == lane:
children.append(el)
child_ids = {el.teile_id for el in children}
for el in children:
add(_chain_child_dot(el.teile_id, el))
for te in track.edges:
if te.src in child_ids and te.dst in child_ids:
add(f' "{te.src}" -> "{te.dst}" [color="#8ea9db"];')
add(" }")
add(" }")
add("")
for index, element in enumerate(
sorted((e for e in elements if e.is_transport and e.position and e.bbox), key=lambda e: e.teile_id)
):
node_id = element.teile_id
top_ids.add(node_id)
add(f" subgraph cluster_strecke_{index} {{")
add(f' label="{_dot_escape(element.name or node_id)} [{node_id}]"; '
f'style="rounded,filled"; fillcolor="#f5faf0"; color="#548235";')
add(' fontname="Segoe UI"; fontsize=10; fontcolor="#375623"; margin=14;')
style = NODE_STYLES.get(element.kind, DEFAULT_NODE_STYLE)
attrs = ", ".join(f'{key}="{value}"' for key, value in style.items())
add(f' "{node_id}" [label="{_dot_escape(element.name or node_id)}", {attrs}];')
children = [
el for el in _container_children(element, by_id)
if el.x is not None and el.y is not None
]
child_ids = {el.teile_id for el in children}
for el in children:
add(_chain_child_dot(el.teile_id, el))
for te in track.edges:
if te.src in child_ids and te.dst in child_ids:
add(f' "{te.src}" -> "{te.dst}" [color="#2f5597"];')
add(" }")
add("")
for edge in graph.edges:
if edge.src not in top_ids or edge.dst not in top_ids:
continue
attrs = CHAIN_SUBGRAPH_EDGE.get(edge.kind, CHAIN_SUBGRAPH_EDGE[EDGE_FLOW])
add(f' "{edge.src}" -> "{edge.dst}" {attrs};')
add("}")
return "\n".join(out) + "\n"
# ---------------------------------------------------------------------------
# SVG erzeugen
# ---------------------------------------------------------------------------
@@ -2935,6 +3170,7 @@ def report(
bbs_file: Path | None = None,
track_dot_file: Path | None = None,
track_svg_file: Path | None = None,
chain_dot_file: Path | None = None,
chain_file: Path | None = None,
) -> None:
kinds: dict[str, int] = {}
@@ -2968,7 +3204,11 @@ def report(
print(f"Bahnzusammenhang = {track_dot_file if track_dot_file else '- (ohne --track)'}")
if track_dot_file:
print(f" SVG = {track_svg_file if track_svg_file else '- (ohne --tosvg)'}")
print(f"Chain-Graph = {chain_file if chain_file else '- (ohne --chain-graph)'}")
if chain_dot_file:
print(f"Chain-Graph = {chain_dot_file}")
print(f" SVG = {chain_file if chain_file else '- (ohne --tosvg)'}")
else:
print(f"Chain-Graph = {chain_file if chain_file else '- (ohne --chain-graph)'}")
print("================================================================")
if stale_svg is not None:
@@ -3020,6 +3260,8 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
" material_flow.bat --file HundM05.csv --show-bbs\n"
" material_flow.bat --file mubea.csv --track --tosvg\n"
" material_flow.bat --file HundM05.csv --chain-graph\n"
" material_flow.bat --file HundM05.csv --chain-graph --graphtype compound\n"
" material_flow.bat --file HundM05.csv --chain-graph --graphtype subgraph --tosvg\n"
"\n"
"Exit-Codes:\n"
" 0 ok\n"
@@ -3093,13 +3335,24 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
"--chain-graph",
dest="chain_graph",
action="store_true",
help="Massstabsgetreuen Chain-/Lageplan der groben Objekte als "
"<name>_chain.svg nach %%SKEL_RESULTS%% schreiben: jeder Kreisel "
"(je Bahn L/R) und jede Strecke/Gefaellestrecke (VF/GF) als "
"Rechteck an ihrer echten Position/Boundingbox, verbunden durch "
"die bereits aufgeloesten Materialfluss-Kanten. Schritt 1 - noch "
"ohne Anbauteile (Separator, Scanner, Weiche, ...) oder Ein-/"
"Ausgabestationen. Braucht kein Graphviz.",
help="Chain-/Lageplan der groben Objekte (Kreisel je Bahn L/R, Strecke/"
"Gefaellestrecke) nach %%SKEL_RESULTS%% schreiben, verbunden durch "
"die bereits aufgeloesten Materialfluss-Kanten. Noch ohne Ein-/"
"Ausgabestationen. Methode/Ausgabedatei je --graphtype.",
)
parser.add_argument(
"--graphtype",
choices=("coarsebox", "compound", "subgraph"),
default="coarsebox",
help="Nur zusammen mit --chain-graph: Darstellungsmethode. 'coarsebox' "
"(Default) = nur die groben Objekte, massstabsgetreu, wie bisher "
"(<name>_chain.svg, kein Graphviz). 'compound' = zusaetzlich die "
"Anbauteile (Separator, Scanner, Ausschleus-/Einschleuselement, "
"Weiche) massstabsgetreu in ihrem Container verschachtelt "
"(<name>_chain.svg, kein Graphviz). 'subgraph' = dieselbe "
"Verschachtelung als geschachtelte Graphviz-Cluster, schematisch "
"statt lagerichtig (<name>_chain.dot, mit --tosvg zusaetzlich "
"<name>_chain.svg).",
)
return parser.parse_args(argv)
@@ -3139,6 +3392,11 @@ def main(argv: list[str] | None = None) -> int:
file=sys.stderr)
return 1
if args.graphtype != "coarsebox" and not args.chain_graph:
print("FEHLER: --graphtype ist nur zusammen mit --chain-graph moeglich.",
file=sys.stderr)
return 1
bbs_file = None
if args.show_bbs:
bbs_target = results / f"{stem}_bbs.svg"
@@ -3214,8 +3472,10 @@ def main(argv: list[str] | None = None) -> int:
track: TrackGraph | None = None
track_dot_file = track_svg_file = None
if args.track:
needs_track = args.track or (args.chain_graph and args.graphtype != "coarsebox")
if needs_track:
track = build_track_graph(elements, by_id, graph, warnings)
if args.track:
track_dot_target = results / f"{stem}_track.dot"
track_dot_target.write_text(render_track_dot(track), encoding="utf-8")
track_dot_file = track_dot_target
@@ -3228,15 +3488,33 @@ def main(argv: list[str] | None = None) -> int:
print(f"FEHLER: Bahnzusammenhang-SVG nicht erzeugt: {exc}", file=sys.stderr)
chain_file = None
chain_dot_file = None
if args.chain_graph:
try:
svg = render_chain_svg(elements, graph, csv_file.name)
except ValueError as exc:
print(f"WARNUNG: Chain-Graph: {exc} - nicht erzeugt.", file=sys.stderr)
if args.graphtype == "subgraph":
chain_dot_target = results / f"{stem}_chain.dot"
chain_dot_target.write_text(
render_chain_subgraph_dot(elements, graph, track, csv_file.name), encoding="utf-8"
)
chain_dot_file = chain_dot_target
if args.tosvg:
chain_svg_target = results / f"{stem}_chain.svg"
try:
dot_to_svg(chain_dot_file, chain_svg_target)
chain_file = chain_svg_target
except RuntimeError as exc:
print(f"FEHLER: Chain-Graph-SVG nicht erzeugt: {exc}", file=sys.stderr)
else:
chain_target = results / f"{stem}_chain.svg"
chain_target.write_text(svg, encoding="utf-8")
chain_file = chain_target
try:
if args.graphtype == "compound":
svg = render_chain_compound_svg(elements, graph, track, csv_file.name)
else:
svg = render_chain_svg(elements, graph, csv_file.name)
except ValueError as exc:
print(f"WARNUNG: Chain-Graph: {exc} - nicht erzeugt.", file=sys.stderr)
else:
chain_target = results / f"{stem}_chain.svg"
chain_target.write_text(svg, encoding="utf-8")
chain_file = chain_target
report(
graph,
@@ -3249,6 +3527,7 @@ def main(argv: list[str] | None = None) -> int:
bbs_file=bbs_file,
track_dot_file=track_dot_file,
track_svg_file=track_svg_file,
chain_dot_file=chain_dot_file,
chain_file=chain_file,
)