dxfmakros/lib/export_neighbors.py

#!/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 configparser

from export_blockpatterns import cfg_path_from_env

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 = configparser.ConfigParser(inline_comment_prefixes=("#",))
    parser.read(cfg_path, encoding="utf-8")
    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 = configparser.ConfigParser(inline_comment_prefixes=("#",))
    parser.read(cfg_path, encoding="utf-8")
    return parser.getfloat("Nachbarschaft", "omniflo_zellgroesse_mm", fallback=3000.0)


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 _test_group_pairs(group, result):
    """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]
            if _overlaps(bounds_a, bounds_b):
                _add_neighbor(result, idx_a, idx_b, id_a, id_b)


def _test_group_against_group(group_a, group_b, result):
    """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):
                _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):
    """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)

    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)
                if _overlaps(bounds_a, bounds_b):
                    _add_neighbor(result, idx_a, idx_b, id_a, id_b)


def compute_neighbor_ids(items, tolerance_mm, omniflo_cell_size_mm=3000.0):
    """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.

    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 = []

    for idx, item in enumerate(items):
        bbox = item.get("_bbox")
        if not bbox:
            continue
        teileart = item.get("teileart", "")
        teileid = item.get("teileid", "")
        bounds = _bounds(bbox, half_tol)
        if teileart in KREISEL_TEILEARTEN:
            kreisel_group.append((idx, bounds, teileid))
        elif teileart in STRECKEN_TEILEARTEN:
            strecken_group.append((idx, bounds, teileid))
        elif teileart in OMNIFLO_TEILEARTEN:
            omniflo_group.append((idx, bounds, teileid, bbox))

    # 1. Kreisel/Eckrad gegen Kreisel/Eckrad
    _test_group_pairs(kreisel_group, result)

    # 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)

    # 3. Omniflo gegen Omniflo, rasterbasiert vorgefiltert
    _test_omniflo_grid(omniflo_group, omniflo_cell_size_mm, result)

    return [", ".join(neighbor_ids) for neighbor_ids in result]


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