#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ export_neighbors.py - Ermittelt benachbarte Elemente ueber Bounding-Box- Ueberschneidung in der x/y-Ebene (Grundriss, Z/Hoehe wird ignoriert). Reines Python (keine externe Abhaengigkeit wie shapely) - die Bloecke werden dazu in drei Gruppen eingeteilt und NUR innerhalb dieser Kombinationen gegeneinander geprueft (nicht mehr alle Elemente einer Zeichnung gemeinsam): 1. Kreisel/Eckrad gegen Kreisel/Eckrad 2. Kreisel/Eckrad gegen Gefaellestrecke/Foerderer/Strecke-Modul - diese drei Kategorien werden NICHT gegeneinander getestet (nur relevant, wo sie an einen Kreisel/Eckrad anschliessen) 3. Omniflo-Elemente (Bogen/Weiche/Gerade) gegen Omniflo-Elemente Omniflo-Anlagen koennen mehrere hundert Elemente enthalten - ein Test jedes gegen jedes (O(n^2)) waere dort zu teuer. Die Omniflo-Gruppe wird daher vorab per Raster (Zellgroesse siehe [Nachbarschaft] -> omniflo_zellgroesse_mm in cfg/export.cfg) nach x/y-Koordinate in Quadranten/Zellen eingeteilt; jedes Element wird nur gegen Elemente in derselben und den 8 angrenzenden Zellen geprueft (3x3-Nachbarschaft, damit auch Elemente knapp jenseits einer Zellgrenze noch erkannt werden). Zwei Bounding-Boxes gelten als benachbart, wenn sie sich (nach Erweiterung um die halbe Toleranz je Seite) in x/y ueberschneiden. Toleranz aus cfg/export.cfg ([Nachbarschaft] -> toleranz_mm), Wert in mm. compute_neighbor_errors() prueft zusaetzlich, ob jedes Element die fuer seine TeileArt noetige Mindestanzahl an Partnern (Nachbarn) hat (siehe 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"} STRECKEN_TEILEARTEN = {"ILS 2.0 Gefaellestrecke", "ILS 2.0 Strecke", "ILS 2.0 Strecke - Modul"} OMNIFLO_TEILEARTEN = {"Omniflo Kurve", "Omniflo Weiche", "Omniflo Gerade"} # Mindestanzahl Partner (Nachbarn) je TeileArt fuer die Fehlerspalte "Fehler" # (siehe compute_neighbor_errors): Kreisel/Eckrad und Omniflo-Elemente # brauchen mindestens einen Partner, Gefaellestrecke/Foerderer/Strecke-Modul # muessen an zwei Seiten anschliessen (mindestens zwei Partner). TeileArten, # die hier nicht auftauchen (z.B. die synthetische "Omniflo Sum"-Zeile), # werden nicht geprueft. MIN_PARTNER = {} MIN_PARTNER.update({t: 1 for t in KREISEL_TEILEARTEN}) MIN_PARTNER.update({t: 1 for t in OMNIFLO_TEILEARTEN}) MIN_PARTNER.update({t: 2 for t in STRECKEN_TEILEARTEN}) def load_neighbor_tolerance_mm(cfg_path=None): """Liest die Nachbarschafts-Toleranz (mm) aus export.cfg. Default: 0.""" if cfg_path is None: cfg_path = cfg_path_from_env() parser = load_export_cfg(cfg_path) return parser.getfloat("Nachbarschaft", "toleranz_mm", fallback=0.0) def load_omniflo_cell_size_mm(cfg_path=None): """Liest die Rastergroesse (mm) fuer die Omniflo-Nachbarschaftspruefung. Default 3000mm: groesser als die ueblichen Omniflo-Bauteile (AP110- Geraden ca. 2000mm), damit ein Element nicht durch ein zu kleines Raster in mehr als eine Zelle ueberlappt. """ if cfg_path is None: cfg_path = cfg_path_from_env() parser = load_export_cfg(cfg_path) 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 ( bbox["cx"] - bbox["dx"] / 2.0 - half_tol, bbox["cx"] + bbox["dx"] / 2.0 + half_tol, bbox["cy"] - bbox["dy"] / 2.0 - half_tol, bbox["cy"] + bbox["dy"] / 2.0 + half_tol, ) def _overlaps(bounds_a, bounds_b): """True, wenn sich zwei (minx, maxx, miny, maxy)-Rechtecke in x/y ueberschneiden.""" a_minx, a_maxx, a_miny, a_maxy = bounds_a b_minx, b_maxx, b_miny, b_maxy = bounds_b return a_minx <= b_maxx and b_minx <= a_maxx and a_miny <= b_maxy and b_miny <= a_maxy def _add_neighbor(result, idx_a, idx_b, id_a, id_b): result[idx_a].append(id_b) result[idx_b].append(id_a) 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) for a in range(n): idx_a, bounds_a, id_a = group[a] for b in range(a + 1, n): idx_b, bounds_b, id_b = group[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, 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: 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 _grid_cell(bbox, cell_size_mm): """Rasterzelle (Quadrant) einer Bounding-Box nach ihrem Mittelpunkt.""" return (int(bbox["cx"] // cell_size_mm), int(bbox["cy"] // cell_size_mm)) 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 Ueberschneidungen an Zellgrenzen zu verpassen.""" buckets = {} for entry in group: 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 = [] for dx in (-1, 0, 1): for dy in (-1, 0, 1): candidates.extend(buckets.get((cx + dx, cy + dy), [])) for idx_a, bounds_a, id_a, _ in cell_items: for idx_b, bounds_b, id_b, _ in candidates: if idx_a >= idx_b: continue pair = (idx_a, idx_b) if pair in checked: continue checked.add(pair) 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, dbg=None): """Ermittelt je Item die IDs (item["teileid"]) benachbarter Elemente. items = Liste von dict mit "teileart", "teileid" und optional "_bbox" ({"cx","cy","dx","dy",...}, siehe export_csv.py bbox_columns). Items 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). """ half_tol = tolerance_mm / 2.0 result = [[] for _ in items] 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, 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, 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, dbg) 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): """Prueft je Item die Mindestanzahl an Partnern (siehe MIN_PARTNER). items = dieselbe Liste wie bei compute_neighbor_ids (braucht "teileart"). neighbor_ids = Rueckgabe von compute_neighbor_ids (positionsgleich zu items). Rueckgabe: Liste von Fehlertexten, positionsgleich zu items: "unverbunden" - keine Partner, obwohl mindestens einer noetig waere "nur ein Partner" - genau ein Partner, obwohl mindestens zwei noetig waeren "" - Mindestanzahl erreicht, oder TeileArt wird nicht geprueft """ errors = [] for item, nachbarn in zip(items, neighbor_ids): min_partner = MIN_PARTNER.get(item.get("teileart", "")) if min_partner is None: errors.append("") continue anzahl = len([p for p in nachbarn.split(",") if p.strip()]) if anzahl == 0: errors.append("unverbunden") elif anzahl == 1 and min_partner >= 2: errors.append("nur ein Partner") else: errors.append("") return errors