From f697ee8db541f342c8d1bc217b7f53e0bbf0d3a2 Mon Sep 17 00:00:00 2001 From: Michael Stangl Date: Mon, 27 Jul 2026 13:43:29 +0200 Subject: [PATCH] =?UTF-8?q?debug=20lib=20f=C3=BCr=20Kollisionspr=C3=BCfung?= =?UTF-8?q?=20impl?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- lib/export_csv.py | 32 ++++++++++- lib/export_neighbors.py | 118 ++++++++++++++++++++++++++++++++++++---- 2 files changed, 136 insertions(+), 14 deletions(-) diff --git a/lib/export_csv.py b/lib/export_csv.py index 7debb7a..0622f5d 100644 --- a/lib/export_csv.py +++ b/lib/export_csv.py @@ -55,6 +55,7 @@ from export_planquadrat import load_planquadrat_config, resolve_origins, compute from export_neighbors import ( load_neighbor_tolerance_mm, load_omniflo_cell_size_mm, + load_collision_debug_target, compute_neighbor_ids, compute_neighbor_errors, ) @@ -63,6 +64,7 @@ BLOCKPATTERNS = load_patterns() PLANQUADRAT_CFG = load_planquadrat_config() NEIGHBOR_TOLERANCE_MM = load_neighbor_tolerance_mm() NEIGHBOR_OMNIFLO_CELL_SIZE_MM = load_omniflo_cell_size_mm() +NEIGHBOR_DEBUG_TARGET = load_collision_debug_target() # --------------------------------------------------------------------------- @@ -311,7 +313,7 @@ SKIP_BLOCKS = set(BLOCKPATTERNS.get("pattern_ks_subblocks", ["K1", "K2", "K3", " # Bloecke verarbeiten (einfache Liste, keine Summierung) # --------------------------------------------------------------------------- -def process_blocks(blocks, lookup): +def process_blocks(blocks, lookup, dbg=None): # x_origin_mm/y_origin_mm einmalig fuer diesen Exportlauf aufloesen (siehe # export_planquadrat.resolve_origins): automatisch aus den Bloecken, falls # nicht explizit in [Planquadrate] konfiguriert. @@ -498,7 +500,7 @@ def process_blocks(blocks, lookup): }) neighbor_ids = compute_neighbor_ids( - items, NEIGHBOR_TOLERANCE_MM, NEIGHBOR_OMNIFLO_CELL_SIZE_MM) + items, NEIGHBOR_TOLERANCE_MM, NEIGHBOR_OMNIFLO_CELL_SIZE_MM, dbg=dbg) neighbor_errors = compute_neighbor_errors(items, neighbor_ids) for item, nachbarn, fehler in zip(items, neighbor_ids, neighbor_errors): item["nachbarn"] = nachbarn @@ -558,7 +560,31 @@ def main(): print(f"[export_csv] {len(blocks)} Bloecke geladen, " f"{len(boegen)} Boegen, {len(weichen)} Weichen im Katalog.") - items = process_blocks(blocks, lookup) + # Optionales Kollisions-/Nachbarschafts-Debuglog (siehe [Nachbarschaft] -> + # debug_log in cfg/export.cfg). dbg ist ein Callable(str), das eine Zeile + # in die .dbg-Datei schreibt - oder None, wenn das Log deaktiviert ist. + dbg_file = None + dbg = None + if NEIGHBOR_DEBUG_TARGET: + try: + os.makedirs(os.path.dirname(NEIGHBOR_DEBUG_TARGET), exist_ok=True) + dbg_file = open(NEIGHBOR_DEBUG_TARGET, "w", encoding="utf-8") + dbg = lambda msg: dbg_file.write(msg + "\n") + dbg(f"# Kollisions-/Nachbarschafts-Debuglog (export_csv.py)") + dbg(f"# Quelle: {raw_json_path}") + dbg(f"# Bloecke im Roh-JSON: {len(blocks)}") + dbg("") + except OSError as exc: + print(f"[export_csv] WARNUNG: Debuglog nicht schreibbar " + f"({NEIGHBOR_DEBUG_TARGET}): {exc}") + dbg_file = None + dbg = None + + items = process_blocks(blocks, lookup, dbg=dbg) + + if dbg_file: + dbg_file.close() + print(f"[export_csv] Kollisions-Debuglog geschrieben: {NEIGHBOR_DEBUG_TARGET}") header = ("Elementnummer;TeileArt;TeileId;Bezeichnung;Planquadrat;Anzahl;Position;Boundingbox;" "Insertpoint;K1;K2;K3;K4;Nachbarn;Fehler;Merkmale") diff --git a/lib/export_neighbors.py b/lib/export_neighbors.py index 76a84d0..e442182 100644 --- a/lib/export_neighbors.py +++ b/lib/export_neighbors.py @@ -33,6 +33,8 @@ MIN_PARTNER) - Ergebnis ist die CSV-Spalte "Fehler" in export_csv.py. Nur von export_csv.py genutzt (EXPORTCSV), nicht von export_sivas.py. """ +import os + from export_blockpatterns import cfg_path_from_env, load_export_cfg KREISEL_TEILEARTEN = {"ILS 2.0 Kreisel", "ILS 2.0 Eckrad"} @@ -72,6 +74,30 @@ def load_omniflo_cell_size_mm(cfg_path=None): return parser.getfloat("Nachbarschaft", "omniflo_zellgroesse_mm", fallback=3000.0) +def load_collision_debug_target(cfg_path=None): + """Zielpfad fuer das Kollisions-/Nachbarschafts-Debuglog aus export.cfg. + + [Nachbarschaft] -> debug_log: leer/0/false/no/off = aus (Rueckgabe None). + 1/true/yes/on = DXFM_LOG/export_collision.dbg (bzw. /logs/, falls + DXFM_LOG nicht gesetzt ist). Jeder andere Wert wird als expliziter + Dateipfad interpretiert. + """ + if cfg_path is None: + cfg_path = cfg_path_from_env() + parser = load_export_cfg(cfg_path) + raw = parser.get("Nachbarschaft", "debug_log", fallback="").strip() + if not raw or raw.lower() in ("0", "false", "no", "off", "nein", "aus"): + return None + if raw.lower() in ("1", "true", "yes", "on", "ja", "ein"): + log_dir = os.environ.get("DXFM_LOG") + if not log_dir: + # cfg_path zeigt auf /cfg/export.cfg -> /logs + log_dir = os.path.join( + os.path.dirname(os.path.dirname(os.path.abspath(cfg_path))), "logs") + return os.path.join(log_dir, "export_collision.dbg") + return raw # expliziter Pfad + + def _bounds(bbox, half_tol): """(minx, maxx, miny, maxy) einer Bounding-Box, je Seite um half_tol erweitert.""" return ( @@ -94,7 +120,17 @@ def _add_neighbor(result, idx_a, idx_b, id_a, id_b): result[idx_b].append(id_a) -def _test_group_pairs(group, result): +def _fmt_bounds(b): + """(minx, maxx, miny, maxy) kompakt fuer das Debuglog.""" + return f"[x {b[0]:.0f}..{b[1]:.0f} | y {b[2]:.0f}..{b[3]:.0f}]" + + +def _dbg_pair(dbg, id_a, id_b, overlap): + if dbg: + dbg(f" {id_a} <-> {id_b}: {'NACHBARN' if overlap else 'kein Kontakt'}") + + +def _test_group_pairs(group, result, dbg=None): """Alle Paare EINER Gruppe gegeneinander testen (O(n^2) - fuer Kreisel/ Eckrad-Stueckzahlen pro Zeichnung unkritisch).""" n = len(group) @@ -102,16 +138,20 @@ def _test_group_pairs(group, result): idx_a, bounds_a, id_a = group[a] for b in range(a + 1, n): idx_b, bounds_b, id_b = group[b] - if _overlaps(bounds_a, bounds_b): + overlap = _overlaps(bounds_a, bounds_b) + _dbg_pair(dbg, id_a, id_b, overlap) + if overlap: _add_neighbor(result, idx_a, idx_b, id_a, id_b) -def _test_group_against_group(group_a, group_b, result): +def _test_group_against_group(group_a, group_b, result, dbg=None): """Jedes Element aus group_a gegen jedes Element aus group_b testen. group_b wird NICHT gegen sich selbst getestet (Aufgabe des Aufrufers).""" for idx_a, bounds_a, id_a in group_a: for idx_b, bounds_b, id_b in group_b: - if _overlaps(bounds_a, bounds_b): + overlap = _overlaps(bounds_a, bounds_b) + _dbg_pair(dbg, id_a, id_b, overlap) + if overlap: _add_neighbor(result, idx_a, idx_b, id_a, id_b) @@ -120,7 +160,7 @@ def _grid_cell(bbox, cell_size_mm): return (int(bbox["cx"] // cell_size_mm), int(bbox["cy"] // cell_size_mm)) -def _test_omniflo_grid(group, cell_size_mm, result): +def _test_omniflo_grid(group, cell_size_mm, result, dbg=None): """Omniflo-Elemente nur gegen Elemente in derselben oder einer der 8 angrenzenden Rasterzellen testen (3x3-Nachbarschaft) statt gegen alle - haelt den Aufwand bei mehreren hundert Omniflo-Elementen niedrig, ohne @@ -130,6 +170,10 @@ def _test_omniflo_grid(group, cell_size_mm, result): idx, bounds, teileid, bbox = entry buckets.setdefault(_grid_cell(bbox, cell_size_mm), []).append(entry) + if dbg: + dbg(f" {len(buckets)} belegte Rasterzelle(n): " + + ", ".join(f"{cell}={len(v)}" for cell, v in sorted(buckets.items()))) + checked = set() for (cx, cy), cell_items in buckets.items(): candidates = [] @@ -145,11 +189,13 @@ def _test_omniflo_grid(group, cell_size_mm, result): if pair in checked: continue checked.add(pair) - if _overlaps(bounds_a, bounds_b): + overlap = _overlaps(bounds_a, bounds_b) + _dbg_pair(dbg, id_a, id_b, overlap) + if overlap: _add_neighbor(result, idx_a, idx_b, id_a, id_b) -def compute_neighbor_ids(items, tolerance_mm, omniflo_cell_size_mm=3000.0): +def compute_neighbor_ids(items, tolerance_mm, omniflo_cell_size_mm=3000.0, dbg=None): """Ermittelt je Item die IDs (item["teileid"]) benachbarter Elemente. items = Liste von dict mit "teileart", "teileid" und optional "_bbox" @@ -157,6 +203,9 @@ def compute_neighbor_ids(items, tolerance_mm, omniflo_cell_size_mm=3000.0): ohne _bbox (z.B. die synthetische Omniflo-Sum-Zeile) bleiben ohne Nachbarn. + dbg = optionale Callable(str) fuer ein Debugprotokoll (Klassifikation, + Gruppen, jeder Ueberschneidungstest, Ergebnis). None = kein Protokoll. + Rueckgabe: Liste von kommaseparierten Nachbar-ID-Strings, positionsgleich zu items (nicht ueber die ID dedupliziert, da TeileId nicht zwingend eindeutig ist). @@ -167,32 +216,79 @@ def compute_neighbor_ids(items, tolerance_mm, omniflo_cell_size_mm=3000.0): kreisel_group = [] strecken_group = [] omniflo_group = [] + skipped = 0 + + if dbg: + dbg("=== Nachbarschafts-/Kollisionserkennung ===") + dbg(f"Toleranz={tolerance_mm}mm (Bounding-Box je Seite +{half_tol}mm erweitert), " + f"Omniflo-Zellgroesse={omniflo_cell_size_mm}mm") + dbg(f"Elemente gesamt: {len(items)}") + dbg("") + dbg("Klassifikation je Element:") for idx, item in enumerate(items): bbox = item.get("_bbox") if not bbox: + skipped += 1 + if dbg: + dbg(f" idx={idx} teileid={item.get('teileid', '')!r} " + f"teileart={item.get('teileart', '')!r} -> UEBERSPRUNGEN " + f"(keine Bounding-Box)") continue teileart = item.get("teileart", "") teileid = item.get("teileid", "") bounds = _bounds(bbox, half_tol) if teileart in KREISEL_TEILEARTEN: + gruppe = "Kreisel/Eckrad" kreisel_group.append((idx, bounds, teileid)) elif teileart in STRECKEN_TEILEARTEN: + gruppe = "Strecke/Foerderer" strecken_group.append((idx, bounds, teileid)) elif teileart in OMNIFLO_TEILEARTEN: + gruppe = "Omniflo" omniflo_group.append((idx, bounds, teileid, bbox)) + else: + gruppe = "(keine Gruppe - wird nicht geprueft)" + if dbg: + dbg(f" idx={idx} teileid={teileid!r} teileart={teileart!r} " + f"-> {gruppe}; bbox(cx={bbox.get('cx', 0):.0f},cy={bbox.get('cy', 0):.0f}," + f"dx={bbox.get('dx', 0):.0f},dy={bbox.get('dy', 0):.0f}) " + f"bounds={_fmt_bounds(bounds)}") + if dbg: + dbg("") + dbg(f"Gruppengroessen: Kreisel/Eckrad={len(kreisel_group)}, " + f"Strecke/Foerderer={len(strecken_group)}, Omniflo={len(omniflo_group)}, " + f"ohne Bounding-Box={skipped}") + dbg("") + dbg("[1] Kreisel/Eckrad gegen Kreisel/Eckrad:") # 1. Kreisel/Eckrad gegen Kreisel/Eckrad - _test_group_pairs(kreisel_group, result) + _test_group_pairs(kreisel_group, result, dbg) + if dbg: + dbg("[2] Kreisel/Eckrad gegen Strecke/Foerderer/Gefaellestrecke:") # 2. Kreisel/Eckrad gegen Gefaellestrecke/Foerderer/Strecke-Modul - # diese drei Kategorien nicht gegeneinander (siehe Modul-Docstring) - _test_group_against_group(kreisel_group, strecken_group, result) + _test_group_against_group(kreisel_group, strecken_group, result, dbg) + if dbg: + dbg("[3] Omniflo gegen Omniflo (rasterbasiert vorgefiltert):") # 3. Omniflo gegen Omniflo, rasterbasiert vorgefiltert - _test_omniflo_grid(omniflo_group, omniflo_cell_size_mm, result) + _test_omniflo_grid(omniflo_group, omniflo_cell_size_mm, result, dbg) - return [", ".join(neighbor_ids) for neighbor_ids in result] + neighbor_id_lists = [", ".join(neighbor_ids) for neighbor_ids in result] + + if dbg: + dbg("") + dbg("Ergebnis je Element:") + for idx, item in enumerate(items): + dbg(f" idx={idx} teileid={item.get('teileid', '')!r}: " + f"{len(result[idx])} Nachbar(n) [{neighbor_id_lists[idx]}]") + dbg("") + dbg(f"Summe gerichteter Nachbarschaftsbeziehungen: " + f"{sum(len(r) for r in result)}") + + return neighbor_id_lists def compute_neighbor_errors(items, neighbor_ids):