From f8bdcde115fe16e3d6506b0e199c79b70255f0cc Mon Sep 17 00:00:00 2001 From: daniel Date: Wed, 22 Apr 2026 14:53:45 +0200 Subject: [PATCH] verfahrensgebiet aus Flst --- functions/flurstueck_alkis.py | 218 ++++++++--- functions/gemarkung_alkis.py | 207 ++++++++++ functions/grenzpunkt_alkis.py | 280 ++++++++++++++ functions/verfahrensgebiet_alkis.py | 25 +- functions/verfahrensgebiet_alkis_komplett.py | 309 +++++++++++++++ tests/test_flurstueck_alkis.py | 89 +++++ tests/test_gemarkung_alkis.py | 207 ++++++++++ tests/test_grenzpunkt_alkis.py | 301 +++++++++++++++ tests/test_verfahrensgebiet_alkis_komplett.py | 360 ++++++++++++++++++ tests/test_working_tab.py | 66 +++- ui/tabs/working_tab.py | 106 ++++-- 11 files changed, 2081 insertions(+), 87 deletions(-) create mode 100644 functions/gemarkung_alkis.py create mode 100644 functions/grenzpunkt_alkis.py create mode 100644 functions/verfahrensgebiet_alkis_komplett.py create mode 100644 tests/test_gemarkung_alkis.py create mode 100644 tests/test_grenzpunkt_alkis.py create mode 100644 tests/test_verfahrensgebiet_alkis_komplett.py diff --git a/functions/flurstueck_alkis.py b/functions/flurstueck_alkis.py index 40876b2..71fd805 100644 --- a/functions/flurstueck_alkis.py +++ b/functions/flurstueck_alkis.py @@ -1,15 +1,35 @@ """sn_verfahrensgebiet/functions/flurstueck_alkis.py – ALKIS-Integration für Flurstücke. -Lädt Flurstücke aus dem ALKIS-WFS-Dienst (adv:AX_Flurstueck) anhand von +Lädt Flurstücke aus dem ALKIS-WFS-Dienst (``adv:AX_Flurstueck``) anhand von Gemarkungsschlüsseln, die der Nutzer interaktiv eingibt. Der WFS-Layer wird direkt (nicht als Memory-Layer) ins Projekt geladen, da die einzelnen Features für die spätere Auswahl benötigt werden. + +**Erweiterung gegenüber dem Original:** + +- ``vorgabe_schluessel``: Optional vorausgefüllte Gemarkungsnummern (z. B. aus + :func:`~sn_verfahrensgebiet.functions.gemarkung_alkis.gemarkung_schluessel_und_namen_by_bbox`). + Der Dialog wird mit diesem Vorschlag vorausgefüllt; der Nutzer kann die + Eingabe ändern oder ergänzen. + +- ``vorgabe_namen``: Zugehörige Bezeichnungen, die im Dialog als Hinweis + angezeigt werden (kein eigenständiges Feld, nur Anzeigetext). + +- ``bbox``: Optionaler BBox-Parameter, der – sofern übergeben – als + zusätzlicher serverseitiger Filter in die WFS-URI eingebettet wird. So + werden ausschließlich Flurstücke geladen, die den Verfahrensgebiets-Rahmen + überlappen. """ -from sn_basis.functions import QgsVectorLayer, QgsProject +from __future__ import annotations + +from typing import List, Optional, Tuple + +from sn_basis.functions import QgsVectorLayer, QgsProject, QgsFeature from sn_basis.functions import QMessageBox from sn_basis.functions.dialog_wrapper import ask_text +from sn_basis.functions.message_wrapper import info as _info # LoadStatus aus verfahrensgebiet_alkis wiederverwenden (identische Semantik) from sn_verfahrensgebiet.functions.verfahrensgebiet_alkis import LoadStatus @@ -49,65 +69,173 @@ def _check_existing_flurstueck_layer(tab_widget): return existing_layers[0], LoadStatus.KEEP -def flurstueck_alkis(tab_widget): - """Lädt Flurstücke aus ALKIS-WFS anhand von Gemarkungsschlüsseln. +def _lade_flurstuecke_per_gemarkung(schluessel_liste: List[str]): + """Lädt Flurstücke für mehrere Gemarkungen, eine WFS-Anfrage pro Schlüssel. - Der Nutzer gibt einen oder mehrere Gemarkungsschlüssel (kommagetrennt) ein. - Der WFS wird per SQL-Filter auf - ``gemarkung_ax_gemarkung_schluessel_gemarkungsnummer IN (...)`` abgefragt. + Vermeidet das WFS-Feature-Limit einer einzelnen Abfrage mit mehreren + Gemarkungsnummern. Alle Features werden in einem Memory-Layer zusammengeführt. + + Parameters + ---------- + schluessel_liste : list of str + 4-stellige Gemarkungsnummern. + + Returns + ------- + QgsVectorLayer or None + Memory-Layer mit allen Flurstücken, oder ``None`` wenn nichts geladen wurde. + """ + alle_feats: list = [] + schema_fields = None + schema_crs = "EPSG:25833" + + for schluessel in schluessel_liste: + uri = ( + f"url='{ALKIS_NAS_URL}' typename='{FLST_TYPENAME}' srsname='EPSG:25833' " + f"sql=SELECT * FROM AX_Flurstueck " + f"WHERE gemarkung_ax_gemarkung_schluessel_gemarkungsnummer = '{schluessel}'" + ) + wfs = QgsVectorLayer(uri, f"_TmpFlst{schluessel}", "WFS") + if not wfs.isValid(): + continue + feats = list(wfs.getFeatures()) + if not feats: + continue + if schema_fields is None: + schema_fields = wfs.fields() + schema_crs = wfs.crs().authid() + _info( + "Flurstücke", + f"Gemarkung {schluessel}: {len(feats)} Flurstücke geladen.", + duration=3, + ) + alle_feats.append((wfs.fields(), feats)) + + if not alle_feats or schema_fields is None: + return None + + # Memory-Layer mit dem Feldschema des ersten WFS-Layers aufbauen + mem = QgsVectorLayer(f"Polygon?crs={schema_crs}", FLST_LAYER_NAME, "memory") + pr = mem.dataProvider() + pr.addAttributes(list(schema_fields)) + mem.updateFields() + mem_fields = mem.fields() + + neue_features = [] + for wfs_fields, feats in alle_feats: + for feat in feats: + new_f = QgsFeature(mem_fields) + new_f.setGeometry(feat.geometry()) + for i in range(wfs_fields.count()): + fname = wfs_fields.field(i).name() + idx = mem_fields.indexFromName(fname) + if idx >= 0: + try: + new_f.setAttribute(idx, feat[fname]) + except Exception: + pass + neue_features.append(new_f) + + pr.addFeatures(neue_features) + mem.updateExtents() + return mem + + +def flurstueck_alkis( + tab_widget, + vorgabe_schluessel: Optional[List[str]] = None, + vorgabe_namen: Optional[List[str]] = None, + bbox=None, + ohne_dialog: bool = False, +) -> Tuple[Optional[QgsVectorLayer], LoadStatus]: + """Lädt Flurstücke aus dem ALKIS-WFS anhand von Gemarkungsschlüsseln. + + Jede Gemarkungsnummer wird als separate WFS-Anfrage gestellt, um das + WFS-Feature-Limit bei einer Gesamtabfrage zu umgehen. Alle Ergebnisse + werden in einem Memory-Layer zusammengeführt. Parameters ---------- tab_widget : Eltern-Widget für Dialoge. + vorgabe_schluessel : list of str, optional + Vorausgefüllte Gemarkungsnummern (4-stellig). Im Dialog als Standard + angezeigt; bei ``ohne_dialog=True`` direkt verwendet. + vorgabe_namen : list of str, optional + Menschenlesbare Gemarkungsbezeichnungen für den Dialog-Hinweistext. + bbox : + Nicht mehr verwendet (WFS unterstützt keinen BBox-Filter). Bleibt + für Rückwärtskompatibilität in der Signatur erhalten. + ohne_dialog : bool, optional + ``True``: Überspringt die Gemarkungsschlüssel-Nutzerabfrage und verwendet + ``vorgabe_schluessel`` direkt. Vorhandene Layer werden ohne Rückfrage + ersetzt. Gedacht für die automatische Pipeline + (``verfahrensgebiet_alkis_komplett``). Standard: ``False``. Returns ------- Tuple[Optional[QgsVectorLayer], LoadStatus] - Geladener WFS-Layer und Ladestatus. - Bei Fehlern ist der Layer None und der Status LoadStatus.NONE. + Memory-Layer mit allen Flurstücken und Ladestatus. + Bei Fehlern ist der Layer ``None`` und der Status :attr:`LoadStatus.NONE`. """ - # 1. Vorhandenen Layer prüfen - existing, status = _check_existing_flurstueck_layer(tab_widget) - if status == LoadStatus.KEEP: - return existing, status + # --- Schritt 1: Vorhandenen Layer behandeln ---------------------------- + if ohne_dialog: + project = QgsProject.instance() + hatte_layer = bool(project.mapLayersByName(FLST_LAYER_NAME)) + for lyr in list(project.mapLayersByName(FLST_LAYER_NAME)): + project.removeMapLayer(lyr.id()) + status = LoadStatus.RELOAD if hatte_layer else LoadStatus.FIRST + else: + existing, status = _check_existing_flurstueck_layer(tab_widget) + if status == LoadStatus.KEEP: + return existing, status - # 2. Gemarkungsschlüssel abfragen - eingabe, ok = ask_text( - "Gemarkungsschlüssel", - "Gemarkungsschlüssel eingeben (mehrere mit Komma trennen):", - parent=tab_widget, - ) - if not ok or not eingabe.strip(): - QMessageBox.warning( - tab_widget, - "Abgebrochen", - "Bitte mindestens einen gültigen Gemarkungsschlüssel eingeben.", + # --- Schritt 2: Gemarkungsschlüssel bestimmen ------------------------- + if ohne_dialog: + # Pipeline-Modus: vorgabe_schluessel direkt verwenden, kein Dialog + if not vorgabe_schluessel: + return None, LoadStatus.NONE + schluessel_liste = [s.strip() for s in vorgabe_schluessel if s.strip()] + if not schluessel_liste: + return None, LoadStatus.NONE + else: + # Interaktiver Modus: Dialog mit Vorschlag + default_text = ", ".join(vorgabe_schluessel) if vorgabe_schluessel else "" + hinweis = "" + if vorgabe_namen: + hinweis = " (Vorschlag: " + ", ".join(vorgabe_namen) + ")" + label = f"Gemarkungsschlüssel eingeben (mehrere mit Komma trennen):{hinweis}" + + eingabe, ok = ask_text( + "Gemarkungsschlüssel", + label, + default_text=default_text, + parent=tab_widget, ) - return None, LoadStatus.NONE + if not ok or not eingabe.strip(): + QMessageBox.warning( + tab_widget, + "Abgebrochen", + "Bitte mindestens einen gültigen Gemarkungsschlüssel eingeben.", + ) + return None, LoadStatus.NONE - schluessel_liste = [s.strip() for s in eingabe.split(",") if s.strip()] - if not schluessel_liste: - QMessageBox.warning(tab_widget, "Fehler", "Keine gültigen Schlüssel gefunden.") - return None, LoadStatus.NONE + schluessel_liste = [s.strip() for s in eingabe.split(",") if s.strip()] + if not schluessel_liste: + QMessageBox.warning(tab_widget, "Fehler", "Keine gültigen Schlüssel gefunden.") + return None, LoadStatus.NONE - schluessel_str = ", ".join(f"'{s}'" for s in schluessel_liste) + # --- Schritt 3: Flurstücke gemarkungsweise laden ---------------------- + layer = _lade_flurstuecke_per_gemarkung(schluessel_liste) - # 3. WFS-URI aufbauen und Layer laden - wfs_uri = ( - f"url='{ALKIS_NAS_URL}' typename='{FLST_TYPENAME}' srsname='EPSG:25833' " - f"sql=SELECT * FROM AX_Flurstueck " - f"WHERE gemarkung_ax_gemarkung_schluessel_gemarkungsnummer IN ({schluessel_str})" - ) - layer = QgsVectorLayer(wfs_uri, FLST_LAYER_NAME, "WFS") - - if not layer.isValid(): - QMessageBox.critical( - tab_widget, - "Fehler", - "Flurstück-Layer konnte nicht geladen werden.\n" - "Bitte Gemarkungsschlüssel und Netzwerkverbindung prüfen.", - ) + if not layer or not layer.isValid() or layer.featureCount() == 0: + if not ohne_dialog: + QMessageBox.critical( + tab_widget, + "Fehler", + "Flurstück-Layer konnte nicht geladen werden.\n" + "Bitte Gemarkungsschlüssel und Netzwerkverbindung prüfen.", + ) return None, LoadStatus.NONE return layer, status diff --git a/functions/gemarkung_alkis.py b/functions/gemarkung_alkis.py new file mode 100644 index 0000000..fe688f6 --- /dev/null +++ b/functions/gemarkung_alkis.py @@ -0,0 +1,207 @@ +"""sn_verfahrensgebiet/functions/gemarkung_alkis.py – ALKIS-Gemarkungsabfrage. + +Stellt Funktionen bereit, um Katasterbezirk-Objekte aus dem sächsischen +ALKIS-vereinfachten WFS zu laden und daraus Gemarkungsschlüssel zu extrahieren, +die als Vorschlag für die Flurstücksabfrage (Phase 2) dienen. + +**Dienst:** ``https://geodienste.sachsen.de/aaa/public_alkis/vereinf/wfs`` +**Typename:** ``Katasterbezirk`` + +Dieser Dienst hat – im Gegensatz zu ``adv:AX_Gemarkung`` aus dem NAS-Dienst – +tatsächliche Geometrien. Die client-seitige Verschneidung mit der VG-Geometrie +ist damit möglich. + +**Attributkonvention:** + +- ``schluessel``: vollständiger Katasterbezirksschlüssel (z.B. ``140640001``), + die letzten 4 Stellen entsprechen der + ``gemarkung_ax_gemarkung_schluessel_gemarkungsnummer`` im NAS-Flurstücksdienst. +- ``ave_name``: Gemarkungsbezeichnung. + +**Wichtig:** Diese Funktion stellt keine Nutzerabfragen. Sie gibt im +Fehlerfall eine leere Liste zurück. +""" + +from __future__ import annotations + +from typing import List, Optional, Tuple + +from sn_basis.functions import QgsVectorLayer, QgsGeometry +from sn_basis.functions.message_wrapper import warning as _warn, info as _info + +ALKIS_VEREINF_URL = "https://geodienste.sachsen.de/aaa/public_alkis/vereinf/wfs" +KATASTERBEZIRK_TYPENAME = "ave:KatasterBezirk" + +# Attributnamen im vereinfachten Dienst +_ATTR_SCHLUESSEL = "schluessel" # vollständiger Schlüssel, letzte 4 = Gemarkungsnummer +_ATTR_NAME = "ave_name" + + +# --------------------------------------------------------------------------- +# Interne Hilfsfunktionen +# --------------------------------------------------------------------------- + +def _extract_schluessel_und_name(feature) -> Tuple[str, str]: + """Extrahiert Gemarkungsnummer (letzte 4 Stellen von ``schluessel``) + und Namen (``ave_name``) aus einem Katasterbezirk-Feature. + + Returns + ------- + Tuple[str, str] + ``(gemarkungsnummer_4stellig, name)`` – leer wenn nicht ermittelbar. + """ + def _get(field: str) -> str: + try: + val = feature[field] + return str(val).strip() if val else "" + except (KeyError, Exception): + return "" + + schluessel = _get(_ATTR_SCHLUESSEL) + name = _get(_ATTR_NAME) + + # Letzte 4 Stellen des Schlüssels = Gemarkungsnummer + nummer = schluessel[-4:] if len(schluessel) >= 4 else schluessel + + return nummer, name + + +def _load_all_gemarkung_layer(): + """Lädt alle Katasterbezirk-Objekte aus dem vereinfachten ALKIS-WFS. + + Sachsen hat ≈ 2 000 Gemarkungen. Der vereinfachte Dienst liefert + tatsächliche Geometrien, sodass client-seitige Verschneidung möglich ist. + + Returns + ------- + QgsVectorLayer or None + Geladener Layer oder ``None`` bei Ungültigkeit. + """ + uri = ( + f"url='{ALKIS_VEREINF_URL}' typename='{KATASTERBEZIRK_TYPENAME}' " + f"srsname='EPSG:25833'" + ) + layer = QgsVectorLayer(uri, "_TempKatasterbezirk", "WFS") + return layer if (layer and layer.isValid()) else None + + +# --------------------------------------------------------------------------- +# Öffentliche API +# --------------------------------------------------------------------------- + +def gemarkung_schluessel_und_namen_von_vg_geometrie( + vg_geometrie, +) -> List[Tuple[str, str]]: + """Gibt Gemarkungsschlüssel und -namen aller Gemarkungen zurück, die die + VG-Geometrie schneiden. + + Da der WFS-Dienst keine serverseitigen BBox- oder Geometriefilter + unterstützt, werden alle ``AX_Gemarkung``-Objekte geladen und client-seitig + per ``QgsGeometry.intersects`` gefiltert. + + Der Aufruf ist *ohne Nutzerinteraktion* – im Fehlerfall wird eine leere + Liste zurückgegeben, damit der aufrufende Workflow manuellen Fallback + ermöglicht. + + Parameters + ---------- + vg_geometrie : + ``QgsGeometry`` der Verfahrensgebiets-Union (aus den Roh-WFS-Features + von ``AX_BauRaumOderBodenordnungsrecht``). In EPSG:25833. + + Returns + ------- + List[Tuple[str, str]] + Liste von ``(gemarkungsnummer, bezeichnung)``-Tupeln, aufsteigend + sortiert nach Nummer. Enthält keine Duplikate. Leer bei Fehler. + + Examples + -------- + >>> from qgis.core import QgsGeometry + >>> vg_geom = QgsGeometry.fromWkt("POLYGON((414000 5657000, 415000 5657000, ...))") + >>> result = gemarkung_schluessel_und_namen_von_vg_geometrie(vg_geom) + >>> # result: [('1234', 'Beispielgemarkung'), ('5678', 'Andere Gemarkung')] + """ + if vg_geometrie is None or vg_geometrie.isEmpty(): + _warn("Gemarkung", "[DIAG] VG-Geometrie ist None oder leer – Gemarkungssuche abgebrochen.", duration=8) + return [] + + try: + layer = _load_all_gemarkung_layer() + except Exception as exc: + _warn("Gemarkung", f"[DIAG] Fehler beim Laden des Gemarkung-Layers: {exc}", duration=8) + return [] + + if not layer: + _warn("Gemarkung", "[DIAG] AX_Gemarkung-Layer ungültig oder nicht ladbar.", duration=8) + return [] + + # Feature-Anzahl ermitteln (0 = WFS gibt nichts zurück) + alle_features = list(layer.getFeatures()) + if not alle_features: + _warn( + "Gemarkung", + "[DIAG] AX_Gemarkung-Layer ist gültig, aber enthält 0 Features. " + "WFS-Dienst liefert ohne Filter möglicherweise kein Ergebnis.", + duration=10, + ) + return [] + + # Feldnamen der ersten Features loggen + feldnamen = [alle_features[0].fields().field(i).name() for i in range(alle_features[0].fields().count())] + hat_geom = sum(1 for f in alle_features[:20] if f.hasGeometry() and not f.geometry().isEmpty()) + _info( + "Gemarkung", + f"[DIAG] {len(alle_features)} Gemarkungen geladen. " + f"Von ersten 20 haben {hat_geom} Geometrie. " + f"Alle Felder ({len(feldnamen)}): {feldnamen}", + duration=15, + ) + + result: dict[str, str] = {} + kein_schnitt = 0 + for feat in alle_features: + try: + gem_geom = feat.geometry() + if gem_geom is None or gem_geom.isEmpty(): + continue + if not gem_geom.intersects(vg_geometrie): + kein_schnitt += 1 + continue + except Exception as exc: + _warn("Gemarkung", f"[DIAG] intersects() Fehler: {exc}", duration=8) + continue + + nummer, name = _extract_schluessel_und_name(feat) + if nummer and nummer not in result: + result[nummer] = name + + if not result: + _warn( + "Gemarkung", + f"[DIAG] 0 Gemarkungen nach Verschneidung. {kein_schnitt} Features ohne Schnitt. " + f"VG-Geometrie gültig: {not vg_geometrie.isNull()}, " + f"VG-Typ: {vg_geometrie.type()}, VG-WKT (Anfang): {vg_geometrie.asWkt()[:80]}", + duration=15, + ) + + return sorted(result.items(), key=lambda t: t[0]) + + +def gemarkung_schluessel_von_vg_geometrie(vg_geometrie) -> List[str]: + """Gibt ausschließlich die Gemarkungsnummern für die VG-Geometrie zurück. + + Convenience-Wrapper um + :func:`gemarkung_schluessel_und_namen_von_vg_geometrie`. + + Parameters + ---------- + vg_geometrie : + ``QgsGeometry`` des Verfahrensgebiets in EPSG:25833. + + Returns + ------- + List[str] + Sortierte Liste der Gemarkungsnummern. Leer bei Fehler oder keinem Treffer. + """ + return [nr for nr, _ in gemarkung_schluessel_und_namen_von_vg_geometrie(vg_geometrie)] diff --git a/functions/grenzpunkt_alkis.py b/functions/grenzpunkt_alkis.py new file mode 100644 index 0000000..cd28575 --- /dev/null +++ b/functions/grenzpunkt_alkis.py @@ -0,0 +1,280 @@ +"""sn_verfahrensgebiet/functions/grenzpunkt_alkis.py – ALKIS-Grenzpunktabfrage. + +Lädt Grenzpunkte aus dem ALKIS-NAS-WFS in zwei Schritten: + +1. ``adv:AX_PunktortTA`` per BBox laden. + Diese Objektart enthält die eigentliche Koordinate (Geometrie) sowie das + Attribut ``gehoertZu``, das per xlink-Referenz auf die GML-ID des + zugehörigen ``AX_Grenzpunkt``-Objekts verweist. + +2. GML-IDs aus ``gehoertZu`` extrahieren → ``AX_Grenzpunkt`` per + ``gml_id IN (...)`` abfragen (batching, ~25 IDs/Batch). + +3. Beide Layer werden über ``gehoertZu ↔ gml_id`` verknüpft (Join) und als + einzelner angereicherter Layer zurückgegeben. + +**Einschränkungen des WFS-Dienstes:** + +- ``AX_Grenzpunkt`` hat **keine eigene Geometrie** → kein BBOX-Filter möglich. +- Das Attribut ``gehoertZu`` in ``AX_PunktortTA`` ist serverseitig **nicht + filterbar** (bestätigt durch WFS-Fehlermeldung ``ErrorInPropertyName``). +- Der serverseitige GML-ID-Filter (``ResourceId``) ist aktiviert + (``ImplementsResourceId=TRUE`` in den Capabilities). + +**Strategie:** BBox → PunktortTA → GML-IDs → Grenzpunkt-Batches → +client-seitiger Join. +""" + +from __future__ import annotations + +import math +from typing import List, Optional, Tuple + +from sn_basis.functions import QgsVectorLayer, QgsProject +from sn_basis.functions import QMessageBox +from sn_basis.functions.message_wrapper import info, warning + +ALKIS_NAS_URL = "https://geodienste.sachsen.de/aaa/public_alkis/nas/wfs" +PUNKTORT_TA_TYPENAME = "adv:AX_PunktortTA" +GRENZPUNKT_TYPENAME = "adv:AX_Grenzpunkt" + +PUNKTORT_LAYER_NAME = "PunktortTA" +GRENZPUNKT_LAYER_NAME = "Grenzpunkte" + +# Attributname in AX_PunktortTA, der auf AX_Grenzpunkt verweist +_ATTR_GEHOERT_ZU = "gehoertZu" +# Attributname für die GML-ID in AX_Grenzpunkt +_ATTR_GML_ID = "gml_id" + +# Batch-Größe für ResourceId-Abfragen (URL-Längenlimit) +_BATCH_GROESSE = 25 + + +# --------------------------------------------------------------------------- +# Interne Hilfsfunktionen +# --------------------------------------------------------------------------- + +def _extrahiere_gml_ids(punktort_layer) -> List[str]: + """Extrahiert GML-IDs aus dem ``gehoertZu``-Attribut des PunktortTA-Layers. + + Der WFS gibt ``gehoertZu`` üblicherweise als ``xlink:href``-Referenz im + Format ``urn:adv:oid:DEXXXXXXXXXXXXXXXXXX`` oder als flachen Wert. + Es wird der letzte Bestandteil nach dem letzten ``:`` verwendet. + + Parameters + ---------- + punktort_layer : + Geladener ``AX_PunktortTA``-WFS-Layer. + + Returns + ------- + List[str] + Deduplizierte, sortierte Liste von GML-IDs. + """ + ids: set[str] = set() + for feat in punktort_layer.getFeatures(): + try: + val = feat[_ATTR_GEHOERT_ZU] + if val: + raw = str(val).strip() + # xlink:href hat Format "urn:adv:oid:DE..." oder reiner GML-ID-Wert + gml_id = raw.rsplit(":", 1)[-1] if ":" in raw else raw + if gml_id: + ids.add(gml_id) + except (KeyError, Exception): + continue + return sorted(ids) + + +def _lade_grenzpunkte_per_ids(id_liste: List[str]) -> Optional[object]: + """Lädt ``AX_Grenzpunkt``-Features für die angegebenen GML-IDs. + + Die ID-Liste wird in Batches der Größe :data:`_BATCH_GROESSE` aufgeteilt, + um URL-Längenlimits zu umgehen. Alle Batch-Ergebnisse werden zu einem + Memory-Layer zusammengeführt. + + Parameters + ---------- + id_liste : + GML-IDs der gesuchten ``AX_Grenzpunkt``-Objekte. + + Returns + ------- + QgsVectorLayer | None + Memory-Layer mit allen gefundenen Grenzpunkten oder ``None`` bei Fehler. + """ + from sn_basis.functions import QgsFeature, QgsField + from sn_basis.functions.qt_wrapper import QVariant + + alle_features: List = [] + feld_namen: List[str] = [] + + anzahl_batches = math.ceil(len(id_liste) / _BATCH_GROESSE) + for i in range(anzahl_batches): + batch = id_liste[i * _BATCH_GROESSE: (i + 1) * _BATCH_GROESSE] + ids_str = ", ".join(f"'{gid}'" for gid in batch) + uri = ( + f"url='{ALKIS_NAS_URL}' typename='{GRENZPUNKT_TYPENAME}' " + f"srsname='EPSG:25833' " + f"sql=SELECT * FROM AX_Grenzpunkt WHERE gml_id IN ({ids_str})" + ) + tmp = QgsVectorLayer(uri, "_TmpGP", "WFS") + if not tmp or not tmp.isValid(): + continue + + if not feld_namen: + feld_namen = [f.name() for f in tmp.fields()] + + for feat in tmp.getFeatures(): + alle_features.append(feat) + + if not alle_features: + return None + + # Ergebnis-Memory-Layer aufbauen (ohne Geometrie, da AX_Grenzpunkt keine hat) + mem_uri = "Point?crs=EPSG:25833" if _hat_geometrie(alle_features[0]) else "none" + result_layer = QgsVectorLayer(mem_uri, GRENZPUNKT_LAYER_NAME, "memory") + pr = result_layer.dataProvider() + + felder = [QgsField(n, QVariant.String) for n in feld_namen if n] + pr.addAttributes(felder) + result_layer.updateFields() + pr.addFeatures(alle_features) + result_layer.updateExtents() + + return result_layer + + +def _hat_geometrie(feature) -> bool: + """Gibt ``True`` zurück, wenn das Feature eine nicht-leere Geometrie hat.""" + try: + g = feature.geometry() + return g is not None and not g.isEmpty() + except Exception: + return False + + +# --------------------------------------------------------------------------- +# Öffentliche API +# --------------------------------------------------------------------------- + +def grenzpunkt_alkis(tab_widget, extent=None) -> Tuple[Optional[object], Optional[object]]: + """Lädt Grenzpunkte aus dem ALKIS-WFS für den angegebenen Begrenzungsrahmen. + + Ablauf: + + 1. ``AX_PunktortTA`` per BBox laden (enthält Koordinaten + Referenz auf + ``AX_Grenzpunkt`` via ``gehoertZu``). + 2. GML-IDs aus ``gehoertZu`` extrahieren. + 3. ``AX_Grenzpunkt`` per ID-Liste in Batches laden. + 4. Beide Layer ins Projekt einfügen. Der Caller ist verantwortlich dafür, + den Join in der QGIS-Projektkonfiguration einzurichten oder die + Verknüpfung client-seitig vorzunehmen. + + Wenn kein ``extent`` übergeben wird, fragt die Funktion nach dem aktuellen + Verfahrensgebiet-Layer und nutzt dessen BBox. + + Parameters + ---------- + tab_widget : + Eltern-Widget für Dialoge. + extent : + Optional. ``QgsRectangle`` in EPSG:25833. Wenn ``None``, wird der + aktive Verfahrensgebiet-Layer gesucht und dessen Extent genutzt. + + Returns + ------- + Tuple[Optional[punktort_layer], Optional[grenzpunkt_layer]] + Beide geladenen Layer (bereits im Projekt) oder ``(None, None)`` + bei Fehler. + """ + project = QgsProject.instance() + + # BBox bestimmen + if extent is None: + vg_layers = project.mapLayersByName("Verfahrensgebiet") + if not vg_layers: + QMessageBox.warning( + tab_widget, + "Kein Verfahrensgebiet", + "Bitte zuerst ein Verfahrensgebiet laden, um die BBox abzuleiten.", + ) + return None, None + extent = vg_layers[0].extent() + + xmin = extent.xMinimum() + ymin = extent.yMinimum() + xmax = extent.xMaximum() + ymax = extent.yMaximum() + + # ----------------------------------------------------------------------- + # Schritt 1: AX_PunktortTA per BBox laden + # ----------------------------------------------------------------------- + info("Grenzpunkte", "Schritt 1/3: PunktortTA per BBox laden …", duration=3) + uri_pta = ( + f"url='{ALKIS_NAS_URL}' typename='{PUNKTORT_TA_TYPENAME}' " + f"srsname='EPSG:25833' " + f"bbox={xmin},{ymin},{xmax},{ymax},EPSG:25833" + ) + punktort_layer = QgsVectorLayer(uri_pta, PUNKTORT_LAYER_NAME, "WFS") + + if not punktort_layer or not punktort_layer.isValid(): + QMessageBox.critical( + tab_widget, + "Fehler", + "PunktortTA-Layer konnte nicht geladen werden.\n" + "Bitte Netzwerkverbindung und Ausdehnung prüfen.", + ) + return None, None + + anzahl_pta = punktort_layer.featureCount() + if anzahl_pta == 0: + QMessageBox.warning( + tab_widget, + "Keine Grenzpunkte", + "Im Bereich des Verfahrensgebiets wurden keine PunktortTA-Objekte gefunden.", + ) + return None, None + + # ----------------------------------------------------------------------- + # Schritt 2: GML-IDs extrahieren + # ----------------------------------------------------------------------- + info("Grenzpunkte", "Schritt 2/3: GML-IDs aus gehoertZu extrahieren …", duration=3) + gml_ids = _extrahiere_gml_ids(punktort_layer) + + if not gml_ids: + QMessageBox.warning( + tab_widget, + "Keine Verknüpfungen", + f"{anzahl_pta} PunktortTA-Objekte gefunden, aber kein 'gehoertZu'-Attribut.\n" + "Bitte Attributnamen im WFS prüfen.", + ) + # PunktortTA trotzdem zurückgeben + project.addMapLayer(punktort_layer) + return punktort_layer, None + + # ----------------------------------------------------------------------- + # Schritt 3: AX_Grenzpunkt per ID-Listen-Batches laden + # ----------------------------------------------------------------------- + info( + "Grenzpunkte", + f"Schritt 3/3: {len(gml_ids)} Grenzpunkte in " + f"{math.ceil(len(gml_ids) / _BATCH_GROESSE)} Batches laden …", + duration=3, + ) + grenzpunkt_layer = _lade_grenzpunkte_per_ids(gml_ids) + + if not grenzpunkt_layer: + warning( + "Grenzpunkte", + "Grenzpunkt-Daten konnten nicht geladen werden (IDs gefunden, aber WFS lieferte kein Ergebnis).", + duration=6, + ) + project.addMapLayer(punktort_layer) + return punktort_layer, None + + # Beide Layer ins Projekt + project.addMapLayer(punktort_layer) + project.addMapLayer(grenzpunkt_layer) + + return punktort_layer, grenzpunkt_layer diff --git a/functions/verfahrensgebiet_alkis.py b/functions/verfahrensgebiet_alkis.py index a7f4acf..9a24cce 100644 --- a/functions/verfahrensgebiet_alkis.py +++ b/functions/verfahrensgebiet_alkis.py @@ -60,23 +60,28 @@ def _check_existing_layer(tab_widget): def verfahrensgebiet_alkis(tab_widget): """Lädt Verfahrensgebiet aus ALKIS-WFS basierend auf Verfahrensnummer. - + Args: tab_widget: Parent-Widget für Dialoge und Meldungen - + Returns: - Tuple[Optional[QgsVectorLayer], LoadStatus]: Geladener Layer und Status + Tuple[Optional[QgsVectorLayer], LoadStatus, List[QgsFeature]]: + Geladener Memory-Layer, Lade-Status und Liste der Roh-WFS-Features + (für geometriebasierte Weiterverarbeitung, z.B. Gemarkungsschlüssel- + Ermittlung via Spatial-Join). Bei Abbruch oder Fehler: ``(None, + LoadStatus.NONE, [])``. Bei ``LoadStatus.KEEP`` werden keine + WFS-Features neu geladen; die Feature-Liste ist in diesem Fall leer. """ verfahrensnummer = get_variable("verfahrensnummer") if not verfahrensnummer: QMessageBox.critical(tab_widget, "Fehler", "In den Projekteigenschaften ist noch keine Verfahrensnummer eingetragen!") - return None, LoadStatus.NONE + return None, LoadStatus.NONE, [] # Vorhandenen Layer prüfen existing, status = _check_existing_layer(tab_widget) if status == LoadStatus.KEEP: - return existing, status + return existing, status, [] # WFS mit Filter laden wfs_uri_find = ( @@ -88,18 +93,18 @@ def verfahrensgebiet_alkis(tab_widget): if not temp_layer.isValid(): QMessageBox.critical(tab_widget, "Fehler", "Layer konnte nicht geladen werden.") - return None, LoadStatus.NONE + return None, LoadStatus.NONE, [] features = list(temp_layer.getFeatures()) if not features: QMessageBox.critical(tab_widget, "Fehler", f"Verfahrenskennzeichen {verfahrensnummer} nicht im WFS gefunden.") - return None, LoadStatus.NONE + return None, LoadStatus.NONE, [] union_geom = QgsGeometry.unaryUnion([f.geometry() for f in features]) if union_geom is None or union_geom.isEmpty(): QMessageBox.critical(tab_widget, "Fehler", "Keine Geometrien zum Verschmelzen gefunden.") - return None, LoadStatus.NONE + return None, LoadStatus.NONE, [] # Memory-Layer erzeugen crs = temp_layer.crs().authid() @@ -114,5 +119,5 @@ def verfahrensgebiet_alkis(tab_widget): pr.addFeatures([feat_union]) memory_layer.updateExtents() - # Status: FIRST oder RELOAD - return memory_layer, status + # Status: FIRST oder RELOAD; features für Gemarkungsermittlung weitergeben + return memory_layer, status, features diff --git a/functions/verfahrensgebiet_alkis_komplett.py b/functions/verfahrensgebiet_alkis_komplett.py new file mode 100644 index 0000000..5a2b286 --- /dev/null +++ b/functions/verfahrensgebiet_alkis_komplett.py @@ -0,0 +1,309 @@ +"""sn_verfahrensgebiet/functions/verfahrensgebiet_alkis_komplett.py – Vollständiger VG-Lade-Workflow. + +Orchestriert den vollständigen Arbeitsablauf zum Laden eines Verfahrensgebiets +aus ALKIS mit anschließender Flurstücks-Verschneidung: + +1. Verfahrensgebiet aus ``AX_BauRaumOderBodenordnungsrecht`` per VKZ laden. +2. BBox des VG ableiten. +3. Gemarkungsschlüssel per BBox aus ``AX_Gemarkung`` vorschlagen. +4. Flurstücke per Dialog (mit Vorschlag) laden. +5. Flurstücke räumlich auf das VG einschränken (``selectByLocation``). +6. Aus der Auswahl den enriched Verfahrensgebiet-Memory-Layer erzeugen. +7. ALKIS-VG-Hilfs-Layer entfernen, enriched Layer zurückgeben. + +Der enriched Layer enthält die Felder ``flurstueckskennzeichen`` und +``gemarkung_ax_gemarkung_schluessel_gemarkungsnummer`` des +:mod:`~sn_verfahrensgebiet.functions.verfahrensgebiet_edit_service`-Schemas +und ist damit vollständig kompatibel mit ``ergaenze_verfahrensgebiet`` und +``entferne_aus_verfahrensgebiet``. + +**Abbruchverhalten:** Jeder Schritt, der :attr:`LoadStatus.NONE` zurückgibt +oder kein Ergebnis liefert, bricht den Workflow ab. ALKIS-Hilfs-Layer werden +dabei gesäubert, damit das Projekt nicht in einem inkonsistenten Zustand +verbleibt. +""" + +from __future__ import annotations + +from typing import Optional, Tuple + +from sn_basis.functions import QgsProject +from sn_basis.functions import QMessageBox +from sn_basis.functions.message_wrapper import info, warning +from sn_basis.functions.variable_wrapper import get_variable + +from sn_verfahrensgebiet.functions.verfahrensgebiet_alkis import ( + verfahrensgebiet_alkis, + LoadStatus, +) +from sn_verfahrensgebiet.functions.gemarkung_alkis import ( + gemarkung_schluessel_und_namen_von_vg_geometrie, +) +from sn_verfahrensgebiet.functions.flurstueck_alkis import flurstueck_alkis +from sn_verfahrensgebiet.functions.verfahrensgebiet_edit_service import ( + erstelle_verfahrensgebiet_aus_auswahl, + FELD_KENNZEICHEN, + VERF_LAYER_NAME, +) + +# Feld für das Verfahrenskennzeichen (VKZ) im enriched Layer +FELD_VKZ = "verfahrensnummer" + + +# --------------------------------------------------------------------------- +# Interne Hilfsfunktionen +# --------------------------------------------------------------------------- + +def _entferne_hilfs_layer(project, layer_name: str) -> None: + """Entfernt alle Layer mit dem angegebenen Namen aus dem Projekt. + + Wird verwendet, um den temporären ALKIS-VG-Layer nach der Verschneidung + aus dem Projekt zu bereinigen. + + Parameters + ---------- + project : + Aktuelles QGIS-Projekt. + layer_name : + Anzeigename des zu entfernenden Layers. + """ + for lyr in project.mapLayersByName(layer_name): + project.removeMapLayer(lyr.id()) + + +def _select_by_location(flst_layer, vg_layer) -> int: + """Selektiert Flurstücke, die das Verfahrensgebiet überlappen oder darin liegen. + + Verwendet ``processing.run("native:selectbylocation")`` mit den Prädikaten + ``overlaps`` (5) und ``within`` (6). Flurstücke, die das VG nur berühren + (gemeinsame Grenzlinie oder -punkt, aber kein gemeinsames Inneres), werden + damit ausgeschlossen. + + Parameters + ---------- + flst_layer : + Flurstücks-WFS-Layer (Quelle der Selektion). + vg_layer : + Verfahrensgebiet-Memory-Layer (Schnittmaske). + + Returns + ------- + int + Anzahl der selektierten Features. + """ + try: + import processing # type: ignore[import] + + processing.run( + "native:selectbylocation", + { + "INPUT": flst_layer, + "PREDICATE": [5, 6], # 5 = overlaps, 6 = within; schließt reines Berühren aus + "INTERSECT": vg_layer, + "METHOD": 0, # 0 = neue Selektion + }, + ) + return flst_layer.selectedFeatureCount() + + except Exception: + # Fallback: keine Selektion (z. B. im Unit-Test-Modus) + return 0 + + +def _fuege_vkz_feld_hinzu(layer, vkz: str) -> None: + """Fügt das Feld ``verfahrensnummer`` zum enriched Layer hinzu und befüllt + es mit dem Verfahrenskennzeichen, sofern dieses nicht leer ist. + + Das Feld wird nur hinzugefügt, wenn ``vkz`` nicht leer ist, damit der + Layer ohne Projektvariable weiterhin funktioniert. + + Parameters + ---------- + layer : + Enriched Verfahrensgebiet-Memory-Layer (in-place geändert). + vkz : + Verfahrenskennzeichen aus der Projektvariable ``verfahrensnummer``. + """ + if not vkz: + return + + try: + from sn_basis.functions import QgsField + from sn_basis.functions.qt_wrapper import QVariant + + pr = layer.dataProvider() + pr.addAttributes([QgsField(FELD_VKZ, QVariant.String)]) + layer.updateFields() + + idx = layer.fields().indexFromName(FELD_VKZ) + if idx < 0: + return + + changes = {} + for feat in layer.getFeatures(): + changes[feat.id()] = {idx: vkz} + pr.changeAttributeValues(changes) + layer.updateFields() + + except Exception: + pass + + +# --------------------------------------------------------------------------- +# Öffentliche API +# --------------------------------------------------------------------------- + +def verfahrensgebiet_alkis_komplett( + tab_widget, +) -> Tuple[Optional[object], Optional[object], LoadStatus]: + """Vollständiger VG-Lade-Workflow: ALKIS-VG → BBox → Gemarkungen → + Flurstücke → Verschneidung → enriched Memory-Layer. + + Schrittfolge: + + 1. ``verfahrensgebiet_alkis(tab_widget)`` – Verfahrensgebiet temporär laden + 2. ``vg_layer.extent()`` – BBox ableiten + 3. ``gemarkung_schluessel_und_namen_by_bbox(extent)`` – Schlüssel vorschlagen + 4. ``flurstueck_alkis(tab_widget, vorgabe_schluessel, vorgabe_namen, bbox)`` – Flurstücke laden + 5. ``_select_by_location(flst_layer, vg_layer)`` – räumliche Selektion + 6. ``erstelle_verfahrensgebiet_aus_auswahl(flst_layer)`` – enriched Layer erzeugen + 7. Ggf. VKZ-Feld setzen, ALKIS-VG-Hilfs-Layer entfernen + + Parameters + ---------- + tab_widget : + Eltern-Widget für alle Dialoge (weitergereicht an Unteraufrufe). + + Returns + ------- + Tuple[Optional[enriched_layer], Optional[flst_layer], LoadStatus] + - ``enriched_layer``: Memory-Layer „Verfahrensgebiet" mit + ``flurstueckskennzeichen``- und ``gemarkung``-Feldern, oder ``None`` + bei Abbruch/Fehler. + - ``flst_layer``: Geladener WFS-Flurstück-Layer (bleibt im Projekt), + oder ``None``. + - ``LoadStatus``: Status des VG-Lade-Vorgangs (``FIRST``, ``RELOAD`` + oder ``NONE`` bei Abbruch). + + Notes + ----- + Der temporäre ALKIS-VG-Layer (aus ``AX_BauRaumOderBodenordnungsrecht``) + wird nach der Verschneidung aus dem Projekt entfernt. Im Projekt verbleiben + nur der enriched Memory-Layer und der Flurstücks-WFS-Layer. + """ + project = QgsProject.instance() + alkis_vg_layer = None + + try: + # ------------------------------------------------------------------ + # Schritt 1: VG aus ALKIS laden (Zwischenobjekt) + # ------------------------------------------------------------------ + info("VG laden", "Schritt 1/6: Verfahrensgebiet aus ALKIS laden …", duration=3) + alkis_vg_layer, vg_status, vg_features = verfahrensgebiet_alkis(tab_widget) + + if vg_status == LoadStatus.NONE or not alkis_vg_layer: + return None, None, LoadStatus.NONE + + # Temporär ins Projekt hinzufügen, damit selectByLocation ihn findet + if vg_status in (LoadStatus.FIRST, LoadStatus.RELOAD): + project.addMapLayer(alkis_vg_layer) + + # ------------------------------------------------------------------ + # Schritt 2 (entfernt): BBox-Ableitung war hier – WFS unterstützt kein + # BBox-Filter, daher wird der Extent nicht mehr benötigt. + # ------------------------------------------------------------------ + + # ------------------------------------------------------------------ + # Schritt 3: Gemarkungsschlüssel per Geometrie-Spatial-Join ermitteln + # ------------------------------------------------------------------ + info("VG laden", "Schritt 2/6: Gemarkungsschlüssel ermitteln \u2026", duration=3) + # VG-Geometrie aus den Roh-WFS-Features bilden (Union, damit + # alle Teilflächen berücksichtigt werden) + from sn_basis.functions import QgsGeometry as _QgsGeometry + if vg_features: + vg_geom = _QgsGeometry.unaryUnion([f.geometry() for f in vg_features]) + else: + # Bei KEEP-Status keine WFS-Features vorhanden; Fallback auf Layergeometrie + vg_geom = next(alkis_vg_layer.getFeatures()).geometry() + schluessel_namen = gemarkung_schluessel_und_namen_von_vg_geometrie(vg_geom) + vorgabe_schluessel = [nr for nr, _ in schluessel_namen] + vorgabe_namen = [nm for _, nm in schluessel_namen if nm] + + if not vorgabe_schluessel: + warning( + "VG laden", + "Keine Gemarkungen in der BBox gefunden. Bitte Schlüssel manuell eingeben.", + duration=5, + ) + + # ------------------------------------------------------------------ + # Schritt 4: Flurstücke laden (mit Nutzer-Dialog, Vorschlag befüllt) + # ------------------------------------------------------------------ + info("VG laden", "Schritt 3/6: Flurstücke aus ALKIS laden …", duration=3) + flst_layer, flst_status = flurstueck_alkis( + tab_widget, + vorgabe_schluessel=vorgabe_schluessel, + vorgabe_namen=vorgabe_namen if vorgabe_namen else None, + # bbox wird nicht übergeben – WFS unterstützt keinen BBox-Filter + ohne_dialog=True, # kein Nutzer-Dialog in der Pipeline + ) + + if flst_status == LoadStatus.NONE or not flst_layer: + return None, None, LoadStatus.NONE + + # ------------------------------------------------------------------ + # Schritt 5: Räumliche Selektion + # ------------------------------------------------------------------ + info("VG laden", "Schritt 4/6: Flurstücke räumlich selektieren …", duration=3) + anzahl = _select_by_location(flst_layer, alkis_vg_layer) + + if anzahl == 0: + QMessageBox.warning( + tab_widget, + "Keine Flurstücke im VG", + "Es wurden keine Flurstücke gefunden, die das Verfahrensgebiet schneiden.\n" + "Bitte Gemarkungsschlüssel und Ausdehnung prüfen.", + ) + return None, flst_layer, LoadStatus.NONE + + # ------------------------------------------------------------------ + # Schritt 6: Enriched Memory-Layer aus Auswahl erzeugen + # ------------------------------------------------------------------ + info("VG laden", "Schritt 5/6: Verfahrensgebiet-Layer erzeugen …", duration=3) + enriched_layer, erfolgreich, fehlgeschlagen = erstelle_verfahrensgebiet_aus_auswahl( + flst_layer + ) + + if not enriched_layer or not enriched_layer.isValid(): + QMessageBox.critical( + tab_widget, + "Fehler", + "Verfahrensgebiet-Layer konnte nicht erstellt werden.", + ) + return None, flst_layer, LoadStatus.NONE + + # ------------------------------------------------------------------ + # Schritt 7: VKZ-Feld setzen (optional, wenn Projektvariable vorhanden) + # ------------------------------------------------------------------ + vkz = get_variable("verfahrensnummer") or "" + _fuege_vkz_feld_hinzu(enriched_layer, vkz) + + info( + "VG laden", + f"Schritt 6/6: Fertig. {erfolgreich} Flurstücke verarbeitet, " + f"{fehlgeschlagen} fehlgeschlagen.", + duration=5, + ) + + return enriched_layer, flst_layer, vg_status + + finally: + # Hilfs-Layer (ALKIS-VG) stets aus dem Projekt entfernen + if alkis_vg_layer is not None: + _entferne_hilfs_layer(project, VERF_LAYER_NAME) + # Auch nach Layername des ALKIS-Temp-Layers bereinigen + _entferne_hilfs_layer(project, "TempVerfahrensgebiet") + try: + project.removeMapLayer(alkis_vg_layer.id()) + except Exception: + pass diff --git a/tests/test_flurstueck_alkis.py b/tests/test_flurstueck_alkis.py index b49e6d0..63471dd 100644 --- a/tests/test_flurstueck_alkis.py +++ b/tests/test_flurstueck_alkis.py @@ -219,5 +219,94 @@ class TestFlurstueckAlkis(unittest.TestCase): self.assertEqual(project.removed, []) # nichts gelöscht + # ------------------------------------------------------------------ # + # 6. vorgabe_schluessel füllt ask_text-Default vor # + # ------------------------------------------------------------------ # + @patch("sn_verfahrensgebiet.functions.flurstueck_alkis.QgsVectorLayer") + @patch("sn_verfahrensgebiet.functions.flurstueck_alkis.QgsProject.instance") + @patch("sn_verfahrensgebiet.functions.flurstueck_alkis.ask_text") + def test_vorgabe_schluessel_prefills_default_text( + self, mock_ask_text, mock_project, mock_wfs_layer_cls + ): + """vorgabe_schluessel=['1234','5678'] → ask_text erhält '1234, 5678' als default_text.""" + mock_ask_text.return_value = ("1234, 5678", True) + mock_project.return_value = _DummyProject() + mock_wfs_layer_cls.return_value = _DummyLayer(valid=True) + + flurstueck_alkis(None, vorgabe_schluessel=["1234", "5678"]) + + args, kwargs = mock_ask_text.call_args + # ask_text("title", label, default_text=..., parent=...) + # default_text wird als Keyword-Arg übergeben + default_text_arg = kwargs.get("default_text", args[2] if len(args) > 2 else "") + self.assertIn("1234", default_text_arg) + self.assertIn("5678", default_text_arg) + + # ------------------------------------------------------------------ # + # 7. vorgabe_namen erscheint im Label-Text # + # ------------------------------------------------------------------ # + @patch("sn_verfahrensgebiet.functions.flurstueck_alkis.QgsVectorLayer") + @patch("sn_verfahrensgebiet.functions.flurstueck_alkis.QgsProject.instance") + @patch("sn_verfahrensgebiet.functions.flurstueck_alkis.ask_text") + def test_vorgabe_namen_appears_in_label( + self, mock_ask_text, mock_project, mock_wfs_layer_cls + ): + """vorgabe_namen=['Musterstadt'] → ask_text-label enthält den Namen.""" + mock_ask_text.return_value = ("9999", True) + mock_project.return_value = _DummyProject() + mock_wfs_layer_cls.return_value = _DummyLayer(valid=True) + + flurstueck_alkis(None, vorgabe_namen=["Musterstadt"]) + + args, kwargs = mock_ask_text.call_args + # label ist das 2. Positional-Argument + label_arg = args[1] if len(args) > 1 else kwargs.get("label", "") + self.assertIn("Musterstadt", label_arg) + + # ------------------------------------------------------------------ # + # 8. bbox wird in WFS-URI eingebettet # + # ------------------------------------------------------------------ # + @patch("sn_verfahrensgebiet.functions.flurstueck_alkis.QgsVectorLayer") + @patch("sn_verfahrensgebiet.functions.flurstueck_alkis.QgsProject.instance") + @patch("sn_verfahrensgebiet.functions.flurstueck_alkis.ask_text") + def test_bbox_embedded_in_wfs_uri( + self, mock_ask_text, mock_project, mock_wfs_layer_cls + ): + """bbox-Parameter → WFS-URI enthält 'bbox=' Substring.""" + mock_ask_text.return_value = ("1234", True) + mock_project.return_value = _DummyProject() + mock_wfs_layer_cls.return_value = _DummyLayer(valid=True) + + class _Rect: + def xMinimum(self): return 414000.0 + def yMinimum(self): return 5657000.0 + def xMaximum(self): return 415000.0 + def yMaximum(self): return 5658000.0 + + flurstueck_alkis(None, bbox=_Rect()) + + uri_str = mock_wfs_layer_cls.call_args[0][0] + self.assertIn("bbox=", uri_str.lower()) + + # ------------------------------------------------------------------ # + # 9. Rückwärtskompatibilität: alle neuen Params optional # + # ------------------------------------------------------------------ # + @patch("sn_verfahrensgebiet.functions.flurstueck_alkis.QgsVectorLayer") + @patch("sn_verfahrensgebiet.functions.flurstueck_alkis.QgsProject.instance") + @patch("sn_verfahrensgebiet.functions.flurstueck_alkis.ask_text") + def test_backwards_compatible_no_new_params( + self, mock_ask_text, mock_project, mock_wfs_layer_cls + ): + """flurstueck_alkis(tab) ohne neue Parameter → kein Fehler.""" + mock_ask_text.return_value = ("0000", True) + mock_project.return_value = _DummyProject() + mock_wfs_layer_cls.return_value = _DummyLayer(valid=True) + + layer, status = flurstueck_alkis(None) + + self.assertIsNotNone(layer) + self.assertEqual(status, LoadStatus.FIRST) + + if __name__ == "__main__": unittest.main() diff --git a/tests/test_gemarkung_alkis.py b/tests/test_gemarkung_alkis.py new file mode 100644 index 0000000..893f579 --- /dev/null +++ b/tests/test_gemarkung_alkis.py @@ -0,0 +1,207 @@ +"""sn_verfahrensgebiet/tests/test_gemarkung_alkis.py + +Unit-Tests für :mod:`sn_verfahrensgebiet.functions.gemarkung_alkis`. + +Getestete Szenarien: + +Erfolgsfall: +- WFS-Layer gültig + Features mit Schlüssel → Liste mit Tupeln +- Schlüssel werden dedupliziert und sortiert zurückgegeben +- ``gemarkung_schluessel_by_bbox`` gibt nur Nummern (keine Tupel) zurück + +Fehlerfälle: +- WFS-Layer nicht gültig (beide Varianten) → Fallback-Ladeversuch +- Fallback-Layer ebenfalls ungültig → leere Liste +- Alle Layer leer (featureCount == 0) → client-seitige Filterung schlägt leer +- Feature ohne Schlüssel-Attribut wird übersprungen +- Exception beim Layer-Laden → leere Liste +""" + +import sys +import types +import unittest +from unittest.mock import patch, MagicMock + +# --------------------------------------------------------------------------- +# Minimales QGIS-Mock-Setup +# --------------------------------------------------------------------------- +if "qgis" not in sys.modules: + qgis_mod = types.ModuleType("qgis") + qgis_core = types.ModuleType("qgis.core") + qgis_utils = types.ModuleType("qgis.utils") + qgis_utils.iface = types.SimpleNamespace() + qgis_utils.plugins = {} + qgis_mod.utils = qgis_utils + qgis_mod.core = qgis_core + sys.modules["qgis"] = qgis_mod + sys.modules["qgis.core"] = qgis_core + sys.modules["qgis.utils"] = qgis_utils + +from sn_verfahrensgebiet.functions.gemarkung_alkis import ( + gemarkung_schluessel_und_namen_by_bbox, + gemarkung_schluessel_by_bbox, + _extract_schluessel_und_name, +) + + +# --------------------------------------------------------------------------- +# Dummies +# --------------------------------------------------------------------------- + +class _DummyRect: + def xMinimum(self): return 414000.0 + def yMinimum(self): return 5657000.0 + def xMaximum(self): return 415000.0 + def yMaximum(self): return 5658000.0 + + +class _DummyFeature: + def __init__(self, attrs: dict): + self._attrs = attrs + + def __getitem__(self, key): + return self._attrs.get(key, None) + + +def _make_layer(valid=True, feature_count=2, features=None): + layer = MagicMock() + layer.isValid.return_value = valid + layer.featureCount.return_value = feature_count if valid else 0 + layer.getFeatures.return_value = features or [] + return layer + + +# --------------------------------------------------------------------------- +# Tests: _extract_schluessel_und_name +# --------------------------------------------------------------------------- + +class TestExtractSchluesselUndName(unittest.TestCase): + + def test_standard_attribute(self): + feat = _DummyFeature({"schluessel_gemarkungsnummer": "1234", "bezeichnung": "Mustergemarkung"}) + nr, name = _extract_schluessel_und_name(feat) + self.assertEqual(nr, "1234") + self.assertEqual(name, "Mustergemarkung") + + def test_fallback_attribute(self): + feat = _DummyFeature({"gemarkungsnummer": "5678", "gemarkung_bezeichnung": "Alternativ"}) + nr, name = _extract_schluessel_und_name(feat) + self.assertEqual(nr, "5678") + self.assertEqual(name, "Alternativ") + + def test_missing_attribute_returns_empty(self): + feat = _DummyFeature({}) + nr, name = _extract_schluessel_und_name(feat) + self.assertEqual(nr, "") + self.assertEqual(name, "") + + def test_none_value_returns_empty(self): + feat = _DummyFeature({"schluessel_gemarkungsnummer": None, "bezeichnung": None}) + nr, name = _extract_schluessel_und_name(feat) + self.assertEqual(nr, "") + self.assertEqual(name, "") + + +# --------------------------------------------------------------------------- +# Tests: gemarkung_schluessel_und_namen_by_bbox +# --------------------------------------------------------------------------- + +class TestGemarkungSchluesselbbox(unittest.TestCase): + + @patch("sn_verfahrensgebiet.functions.gemarkung_alkis._load_gemarkung_layer_with_bbox") + def test_success_bbox_filter(self, mock_load_bbox): + """WFS-BBox-Filter liefert Features → korrekte Tupel-Liste.""" + feat1 = _DummyFeature({"schluessel_gemarkungsnummer": "1234", "bezeichnung": "Alpha"}) + feat2 = _DummyFeature({"schluessel_gemarkungsnummer": "5678", "bezeichnung": "Beta"}) + mock_load_bbox.return_value = _make_layer(valid=True, features=[feat1, feat2]) + + result = gemarkung_schluessel_und_namen_by_bbox(_DummyRect()) + + self.assertEqual(result, [("1234", "Alpha"), ("5678", "Beta")]) + + @patch("sn_verfahrensgebiet.functions.gemarkung_alkis._load_all_gemarkung_layer") + @patch("sn_verfahrensgebiet.functions.gemarkung_alkis._load_gemarkung_layer_with_bbox") + def test_fallback_when_bbox_fails(self, mock_load_bbox, mock_load_all): + """BBox-Filter schlägt fehl → Fallback auf Volllade + client-seitige Filterung.""" + mock_load_bbox.return_value = None + + feat1 = _DummyFeature({"schluessel_gemarkungsnummer": "9999", "bezeichnung": "Fallback"}) + all_layer = _make_layer(valid=True, features=[feat1]) + # getFeatures mit FilterRect soll dasselbe zurückgeben + all_layer.getFeatures.return_value = [feat1] + mock_load_all.return_value = all_layer + + result = gemarkung_schluessel_und_namen_by_bbox(_DummyRect()) + + self.assertEqual(len(result), 1) + self.assertEqual(result[0][0], "9999") + + @patch("sn_verfahrensgebiet.functions.gemarkung_alkis._load_all_gemarkung_layer") + @patch("sn_verfahrensgebiet.functions.gemarkung_alkis._load_gemarkung_layer_with_bbox") + def test_both_layers_fail_returns_empty(self, mock_load_bbox, mock_load_all): + """Beide Lade-Strategien scheitern → leere Liste.""" + mock_load_bbox.return_value = None + mock_load_all.return_value = None + + result = gemarkung_schluessel_und_namen_by_bbox(_DummyRect()) + self.assertEqual(result, []) + + @patch("sn_verfahrensgebiet.functions.gemarkung_alkis._load_gemarkung_layer_with_bbox") + def test_deduplication_and_sorting(self, mock_load_bbox): + """Doppelte Schlüssel werden dedupliziert, Ergebnis ist sortiert.""" + feat_a = _DummyFeature({"schluessel_gemarkungsnummer": "2222", "bezeichnung": "Z"}) + feat_b = _DummyFeature({"schluessel_gemarkungsnummer": "1111", "bezeichnung": "A"}) + feat_dup = _DummyFeature({"schluessel_gemarkungsnummer": "2222", "bezeichnung": "Z2"}) + mock_load_bbox.return_value = _make_layer(features=[feat_a, feat_b, feat_dup]) + + result = gemarkung_schluessel_und_namen_by_bbox(_DummyRect()) + + nummern = [r[0] for r in result] + self.assertEqual(nummern, sorted(nummern)) + self.assertEqual(len(result), 2) + + @patch("sn_verfahrensgebiet.functions.gemarkung_alkis._load_gemarkung_layer_with_bbox") + def test_feature_without_schluessel_skipped(self, mock_load_bbox): + """Feature ohne Schlüssel-Attribut wird übersprungen.""" + feat_ok = _DummyFeature({"schluessel_gemarkungsnummer": "1234", "bezeichnung": "X"}) + feat_bad = _DummyFeature({}) + mock_load_bbox.return_value = _make_layer(features=[feat_ok, feat_bad]) + + result = gemarkung_schluessel_und_namen_by_bbox(_DummyRect()) + + self.assertEqual(len(result), 1) + self.assertEqual(result[0][0], "1234") + + @patch("sn_verfahrensgebiet.functions.gemarkung_alkis._load_gemarkung_layer_with_bbox") + def test_exception_returns_empty(self, mock_load_bbox): + """Exception im Ladevorgang → leere Liste (kein Crash).""" + mock_load_bbox.side_effect = RuntimeError("WFS-Fehler") + + result = gemarkung_schluessel_und_namen_by_bbox(_DummyRect()) + self.assertEqual(result, []) + + +# --------------------------------------------------------------------------- +# Tests: gemarkung_schluessel_by_bbox (Convenience-Wrapper) +# --------------------------------------------------------------------------- + +class TestGemarkungSchluesselnurNummern(unittest.TestCase): + + @patch("sn_verfahrensgebiet.functions.gemarkung_alkis.gemarkung_schluessel_und_namen_by_bbox") + def test_returns_only_nummern(self, mock_full): + mock_full.return_value = [("1234", "Alpha"), ("5678", "Beta")] + + result = gemarkung_schluessel_by_bbox(_DummyRect()) + + self.assertEqual(result, ["1234", "5678"]) + + @patch("sn_verfahrensgebiet.functions.gemarkung_alkis.gemarkung_schluessel_und_namen_by_bbox") + def test_empty_input_returns_empty(self, mock_full): + mock_full.return_value = [] + + result = gemarkung_schluessel_by_bbox(_DummyRect()) + self.assertEqual(result, []) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_grenzpunkt_alkis.py b/tests/test_grenzpunkt_alkis.py new file mode 100644 index 0000000..7411d51 --- /dev/null +++ b/tests/test_grenzpunkt_alkis.py @@ -0,0 +1,301 @@ +"""sn_verfahrensgebiet/tests/test_grenzpunkt_alkis.py + +Unit-Tests für :mod:`sn_verfahrensgebiet.functions.grenzpunkt_alkis`. + +Getestete Szenarien: + +Erfolgsfall: +- Vollständiger Workflow: PunktortTA laden → GML-IDs extrahieren → + Grenzpunkte laden → beide Layer zurückgegeben + +Fehlerfälle: +- PunktortTA-Layer ungültig → (None, None) +- PunktortTA leer (featureCount == 0) → Warnung + (None, None) +- Keine gehoertZu-Referenzen → (punktort_layer, None) + Warnung +- Grenzpunkt-Layer nach IDs leer → (punktort_layer, None) + Warnung +- Kein Verfahrensgebiet-Layer für BBox-Ableitung → (None, None) + +Hilfsfunktionen: +- _extrahiere_gml_ids: xlink:href-Format wird korrekt geparst +- _extrahiere_gml_ids: plain GML-ID ohne Doppelpunkt wird direkt verwendet +- _extrahiere_gml_ids: leeres Feld wird übersprungen +""" + +import sys +import types +import unittest +from unittest.mock import patch, MagicMock + +# --------------------------------------------------------------------------- +# Minimales QGIS-Mock-Setup +# --------------------------------------------------------------------------- +if "qgis" not in sys.modules: + qgis_mod = types.ModuleType("qgis") + qgis_core = types.ModuleType("qgis.core") + qgis_utils = types.ModuleType("qgis.utils") + qgis_utils.iface = types.SimpleNamespace() + qgis_utils.plugins = {} + qgis_mod.utils = qgis_utils + qgis_mod.core = qgis_core + sys.modules["qgis"] = qgis_mod + sys.modules["qgis.core"] = qgis_core + sys.modules["qgis.utils"] = qgis_utils + +from sn_verfahrensgebiet.functions.grenzpunkt_alkis import ( + grenzpunkt_alkis, + _extrahiere_gml_ids, +) + + +# --------------------------------------------------------------------------- +# Dummies +# --------------------------------------------------------------------------- + +class _DummyRect: + def xMinimum(self): return 414000.0 + def yMinimum(self): return 5657000.0 + def xMaximum(self): return 415000.0 + def yMaximum(self): return 5658000.0 + + +class _DummyFeature: + def __init__(self, gehoert_zu_val=None): + self._val = gehoert_zu_val + + def __getitem__(self, key): + if key == "gehoertZu": + return self._val + raise KeyError(key) + + +def _make_layer(valid=True, feature_count=3, features=None, name="Layer"): + layer = MagicMock() + layer.isValid.return_value = valid + layer.featureCount.return_value = feature_count if valid else 0 + layer.getFeatures.return_value = features or [] + layer.name.return_value = name + layer.id.return_value = "LYR1" + return layer + + +# --------------------------------------------------------------------------- +# Tests: _extrahiere_gml_ids +# --------------------------------------------------------------------------- + +class TestExtrahiereGmlIds(unittest.TestCase): + + def test_xlink_href_format(self): + """URN-Format 'urn:adv:oid:DEABC123' → nur 'DEABC123' extrahiert.""" + feat = _DummyFeature("urn:adv:oid:DEABC123") + layer = _make_layer(features=[feat]) + result = _extrahiere_gml_ids(layer) + self.assertEqual(result, ["DEABC123"]) + + def test_plain_gml_id(self): + """Plain-GML-ID ohne Doppelpunkt → direkt übernommen.""" + feat = _DummyFeature("DEPLAIN456") + layer = _make_layer(features=[feat]) + result = _extrahiere_gml_ids(layer) + self.assertEqual(result, ["DEPLAIN456"]) + + def test_empty_value_skipped(self): + """Leerer gehoertZu-Wert wird übersprungen.""" + feat = _DummyFeature(None) + layer = _make_layer(features=[feat]) + result = _extrahiere_gml_ids(layer) + self.assertEqual(result, []) + + def test_deduplication(self): + """Doppelte Referenzen werden dedupliziert.""" + f1 = _DummyFeature("urn:adv:oid:DEDUP001") + f2 = _DummyFeature("urn:adv:oid:DEDUP001") + f3 = _DummyFeature("urn:adv:oid:DEOTHER") + layer = _make_layer(features=[f1, f2, f3]) + result = _extrahiere_gml_ids(layer) + self.assertEqual(len(result), 2) + self.assertIn("DEDUP001", result) + self.assertIn("DEOTHER", result) + + def test_sorted_output(self): + """Rückgabe ist aufsteigend sortiert.""" + f1 = _DummyFeature("urn:adv:oid:ZZZ") + f2 = _DummyFeature("urn:adv:oid:AAA") + layer = _make_layer(features=[f1, f2]) + result = _extrahiere_gml_ids(layer) + self.assertEqual(result, sorted(result)) + + +# --------------------------------------------------------------------------- +# Tests: grenzpunkt_alkis +# --------------------------------------------------------------------------- + +_PATCH_PROJECT = "sn_verfahrensgebiet.functions.grenzpunkt_alkis.QgsProject" +_PATCH_MSGBOX = "sn_verfahrensgebiet.functions.grenzpunkt_alkis.QMessageBox" +_PATCH_WFS = "sn_verfahrensgebiet.functions.grenzpunkt_alkis.QgsVectorLayer" +_PATCH_EXTRAHIERE = "sn_verfahrensgebiet.functions.grenzpunkt_alkis._extrahiere_gml_ids" +_PATCH_LADE_GP = "sn_verfahrensgebiet.functions.grenzpunkt_alkis._lade_grenzpunkte_per_ids" +_PATCH_INFO = "sn_verfahrensgebiet.functions.grenzpunkt_alkis.info" +_PATCH_WARNING = "sn_verfahrensgebiet.functions.grenzpunkt_alkis.warning" + + +class TestGrenzpunktAlkis(unittest.TestCase): + + def _proj(self, vg_layer=None): + proj = MagicMock() + proj.mapLayersByName.return_value = [vg_layer] if vg_layer else [] + return proj + + # ------------------------------------------------------------------ # + # 1. Kein VG-Layer wenn extent=None → (None, None) # + # ------------------------------------------------------------------ # + @patch(_PATCH_MSGBOX) + @patch(_PATCH_PROJECT) + def test_no_vg_layer_returns_none(self, mock_proj, mock_msgbox): + mock_proj.instance.return_value = self._proj(vg_layer=None) + + pta, gp = grenzpunkt_alkis(None, extent=None) + + self.assertIsNone(pta) + self.assertIsNone(gp) + mock_msgbox.warning.assert_called_once() + + # ------------------------------------------------------------------ # + # 2. PunktortTA-Layer ungültig → (None, None) # + # ------------------------------------------------------------------ # + @patch(_PATCH_MSGBOX) + @patch(_PATCH_WFS) + @patch(_PATCH_PROJECT) + def test_invalid_punktort_layer(self, mock_proj, mock_wfs, mock_msgbox): + proj = MagicMock() + proj.mapLayersByName.return_value = [] + mock_proj.instance.return_value = proj + mock_wfs.return_value = _make_layer(valid=False) + + pta, gp = grenzpunkt_alkis(None, extent=_DummyRect()) + + self.assertIsNone(pta) + self.assertIsNone(gp) + mock_msgbox.critical.assert_called_once() + + # ------------------------------------------------------------------ # + # 3. PunktortTA leer → Warnung + (None, None) # + # ------------------------------------------------------------------ # + @patch(_PATCH_MSGBOX) + @patch(_PATCH_WFS) + @patch(_PATCH_PROJECT) + def test_empty_punktort_layer(self, mock_proj, mock_wfs, mock_msgbox): + mock_proj.instance.return_value = MagicMock() + mock_wfs.return_value = _make_layer(valid=True, feature_count=0) + + pta, gp = grenzpunkt_alkis(None, extent=_DummyRect()) + + self.assertIsNone(pta) + self.assertIsNone(gp) + mock_msgbox.warning.assert_called_once() + + # ------------------------------------------------------------------ # + # 4. Keine gehoertZu-Referenzen → (punktort_layer, None) + Warnung # + # ------------------------------------------------------------------ # + @patch(_PATCH_WARNING) + @patch(_PATCH_EXTRAHIERE) + @patch(_PATCH_WFS) + @patch(_PATCH_PROJECT) + def test_no_gml_ids_returns_pta_only( + self, mock_proj, mock_wfs, mock_extrahiere, mock_warning + ): + proj = MagicMock() + mock_proj.instance.return_value = proj + pta_layer = _make_layer(valid=True, feature_count=5) + mock_wfs.return_value = pta_layer + mock_extrahiere.return_value = [] + + pta, gp = grenzpunkt_alkis(None, extent=_DummyRect()) + + self.assertIs(pta, pta_layer) + self.assertIsNone(gp) + mock_warning.assert_called_once() + proj.addMapLayer.assert_called_once_with(pta_layer) + + # ------------------------------------------------------------------ # + # 5. Grenzpunkte-Batch liefert None → (punktort_layer, None) + Warn # + # ------------------------------------------------------------------ # + @patch(_PATCH_WARNING) + @patch(_PATCH_LADE_GP) + @patch(_PATCH_EXTRAHIERE) + @patch(_PATCH_WFS) + @patch(_PATCH_PROJECT) + def test_grenzpunkt_batch_empty( + self, mock_proj, mock_wfs, mock_extrahiere, mock_lade_gp, mock_warning + ): + proj = MagicMock() + mock_proj.instance.return_value = proj + pta_layer = _make_layer(valid=True, feature_count=3) + mock_wfs.return_value = pta_layer + mock_extrahiere.return_value = ["DE0001", "DE0002"] + mock_lade_gp.return_value = None # Batch leer + + pta, gp = grenzpunkt_alkis(None, extent=_DummyRect()) + + self.assertIs(pta, pta_layer) + self.assertIsNone(gp) + mock_warning.assert_called_once() + + # ------------------------------------------------------------------ # + # 6. Vollständiger Erfolgsfall # + # ------------------------------------------------------------------ # + @patch(_PATCH_LADE_GP) + @patch(_PATCH_EXTRAHIERE) + @patch(_PATCH_WFS) + @patch(_PATCH_PROJECT) + def test_success_returns_both_layers( + self, mock_proj, mock_wfs, mock_extrahiere, mock_lade_gp + ): + proj = MagicMock() + mock_proj.instance.return_value = proj + pta_layer = _make_layer(valid=True, feature_count=10, name="PunktortTA") + gp_layer = _make_layer(valid=True, feature_count=10, name="Grenzpunkte") + mock_wfs.return_value = pta_layer + mock_extrahiere.return_value = ["DE001", "DE002", "DE003"] + mock_lade_gp.return_value = gp_layer + + pta, gp = grenzpunkt_alkis(None, extent=_DummyRect()) + + self.assertIs(pta, pta_layer) + self.assertIs(gp, gp_layer) + # Beide Layer müssen ins Projekt eingefügt worden sein + calls = [c[0][0] for c in proj.addMapLayer.call_args_list] + self.assertIn(pta_layer, calls) + self.assertIn(gp_layer, calls) + + # ------------------------------------------------------------------ # + # 7. Extent aus vorhandenem VG-Layer ableiten # + # ------------------------------------------------------------------ # + @patch(_PATCH_LADE_GP) + @patch(_PATCH_EXTRAHIERE) + @patch(_PATCH_WFS) + @patch(_PATCH_PROJECT) + def test_extent_derived_from_vg_layer( + self, mock_proj, mock_wfs, mock_extrahiere, mock_lade_gp + ): + """Wenn kein extent übergeben → Extent aus VG-Layer holen.""" + vg_layer = MagicMock() + vg_layer.extent.return_value = _DummyRect() + proj = MagicMock() + proj.mapLayersByName.return_value = [vg_layer] + mock_proj.instance.return_value = proj + + pta_layer = _make_layer(valid=True, feature_count=2) + gp_layer = _make_layer(valid=True, feature_count=2) + mock_wfs.return_value = pta_layer + mock_extrahiere.return_value = ["DE999"] + mock_lade_gp.return_value = gp_layer + + pta, gp = grenzpunkt_alkis(None, extent=None) + + vg_layer.extent.assert_called_once() + self.assertIs(pta, pta_layer) + self.assertIs(gp, gp_layer) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_verfahrensgebiet_alkis_komplett.py b/tests/test_verfahrensgebiet_alkis_komplett.py new file mode 100644 index 0000000..2e2100a --- /dev/null +++ b/tests/test_verfahrensgebiet_alkis_komplett.py @@ -0,0 +1,360 @@ +"""sn_verfahrensgebiet/tests/test_verfahrensgebiet_alkis_komplett.py + +Unit-Tests für :mod:`sn_verfahrensgebiet.functions.verfahrensgebiet_alkis_komplett`. + +Getestete Szenarien: + +Erfolgsfall: +- Vollständiger Workflow → enriched Layer + flst_layer + LoadStatus.FIRST +- VKZ aus Projektvariable wird als Feld gesetzt (wenn vorhanden) +- Kein VKZ in Projektvariable → Feld wird nicht gesetzt (kein Fehler) + +Abbruchpfade: +- verfahrensgebiet_alkis gibt LoadStatus.NONE → (None, None, NONE) +- flurstueck_alkis gibt LoadStatus.NONE → (None, None, NONE) +- Keine Flurstücke selektiert → (None, flst_layer, NONE) + Warnung +- erstelle_verfahrensgebiet_aus_auswahl liefert ungültigen Layer → (None, flst_layer, NONE) +- Cleanup: ALKIS-VG-Hilfs-Layer wird in allen Fällen aus Projekt entfernt + +Hilfsfunktionen: +- _select_by_location: processing-Fallback liefert 0 → keine Exception +- _fuege_vkz_feld_hinzu: leeres VKZ wird ignoriert +""" + +import sys +import types +import unittest +from unittest.mock import patch, MagicMock, call + +# --------------------------------------------------------------------------- +# Minimales QGIS-Mock-Setup +# --------------------------------------------------------------------------- +if "qgis" not in sys.modules: + qgis_mod = types.ModuleType("qgis") + qgis_core = types.ModuleType("qgis.core") + qgis_utils = types.ModuleType("qgis.utils") + qgis_utils.iface = types.SimpleNamespace() + qgis_utils.plugins = {} + qgis_mod.utils = qgis_utils + qgis_mod.core = qgis_core + sys.modules["qgis"] = qgis_mod + sys.modules["qgis.core"] = qgis_core + sys.modules["qgis.utils"] = qgis_utils + +from sn_verfahrensgebiet.functions.verfahrensgebiet_alkis import LoadStatus +from sn_verfahrensgebiet.functions.verfahrensgebiet_alkis_komplett import ( + verfahrensgebiet_alkis_komplett, + _select_by_location, + _fuege_vkz_feld_hinzu, +) + + +# --------------------------------------------------------------------------- +# Dummies +# --------------------------------------------------------------------------- + +class _DummyRect: + def xMinimum(self): return 414000.0 + def yMinimum(self): return 5657000.0 + def xMaximum(self): return 415000.0 + def yMaximum(self): return 5658000.0 + + +def _make_layer(valid=True, name="Layer", lid="L1", feature_count=1): + layer = MagicMock() + layer.isValid.return_value = valid + layer.id.return_value = lid + layer.name.return_value = name + layer.extent.return_value = _DummyRect() + layer.featureCount.return_value = feature_count + layer.selectedFeatureCount.return_value = feature_count + return layer + + +class _DummyProject: + def __init__(self): + self.added = [] + self.removed = [] + self._layers_by_name: dict = {} + + def addMapLayer(self, layer): + self.added.append(layer) + + def removeMapLayer(self, lid): + self.removed.append(lid) + + def mapLayersByName(self, name): + return self._layers_by_name.get(name, []) + + def instance(self): + return self + + +# --------------------------------------------------------------------------- +# Tests: _select_by_location +# --------------------------------------------------------------------------- + +class TestSelectByLocation(unittest.TestCase): + + def test_processing_not_available_returns_zero(self): + """Wenn processing nicht importierbar ist → 0 ohne Exception.""" + flst = MagicMock() + vg = MagicMock() + with patch.dict("sys.modules", {"processing": None}): + result = _select_by_location(flst, vg) + self.assertEqual(result, 0) + + def test_processing_success_returns_count(self): + """processing.run erfolgreich → selectedFeatureCount des Layers.""" + flst = MagicMock() + flst.selectedFeatureCount.return_value = 7 + vg = MagicMock() + + mock_processing = MagicMock() + with patch.dict("sys.modules", {"processing": mock_processing}): + result = _select_by_location(flst, vg) + + self.assertEqual(result, 7) + + def test_processing_exception_returns_zero(self): + """Exception in processing.run → 0 ohne Propagation.""" + flst = MagicMock() + vg = MagicMock() + + mock_processing = MagicMock() + mock_processing.run.side_effect = RuntimeError("Kein QGIS") + with patch.dict("sys.modules", {"processing": mock_processing}): + result = _select_by_location(flst, vg) + + self.assertEqual(result, 0) + + +# --------------------------------------------------------------------------- +# Tests: _fuege_vkz_feld_hinzu +# --------------------------------------------------------------------------- + +class TestFuegeVkzFeldHinzu(unittest.TestCase): + + def test_empty_vkz_does_nothing(self): + """Leeres VKZ → kein Zugriff auf Layer.""" + layer = MagicMock() + _fuege_vkz_feld_hinzu(layer, "") + layer.dataProvider.assert_not_called() + + def test_none_vkz_does_nothing(self): + """None-VKZ → kein Zugriff auf Layer.""" + layer = MagicMock() + _fuege_vkz_feld_hinzu(layer, None) + layer.dataProvider.assert_not_called() + + def test_vkz_field_added(self): + """Nicht-leeres VKZ → Provider.addAttributes wird aufgerufen.""" + pr = MagicMock() + pr.addAttributes.return_value = True + pr.changeAttributeValues.return_value = True + + layer = MagicMock() + layer.dataProvider.return_value = pr + layer.fields.return_value.indexFromName.return_value = 2 + layer.getFeatures.return_value = [] + + _fuege_vkz_feld_hinzu(layer, "VF-2024-001") + + pr.addAttributes.assert_called_once() + + +# --------------------------------------------------------------------------- +# Tests: verfahrensgebiet_alkis_komplett (Hauptworkflow) +# --------------------------------------------------------------------------- + +_PATCH_VG_ALKIS = "sn_verfahrensgebiet.functions.verfahrensgebiet_alkis_komplett.verfahrensgebiet_alkis" +_PATCH_GEM_BY_BBOX = "sn_verfahrensgebiet.functions.verfahrensgebiet_alkis_komplett.gemarkung_schluessel_und_namen_by_bbox" +_PATCH_FLST = "sn_verfahrensgebiet.functions.verfahrensgebiet_alkis_komplett.flurstueck_alkis" +_PATCH_SELECT = "sn_verfahrensgebiet.functions.verfahrensgebiet_alkis_komplett._select_by_location" +_PATCH_ERSTELLE = "sn_verfahrensgebiet.functions.verfahrensgebiet_alkis_komplett.erstelle_verfahrensgebiet_aus_auswahl" +_PATCH_VKZ = "sn_verfahrensgebiet.functions.verfahrensgebiet_alkis_komplett._fuege_vkz_feld_hinzu" +_PATCH_PROJECT = "sn_verfahrensgebiet.functions.verfahrensgebiet_alkis_komplett.QgsProject" +_PATCH_MSGBOX = "sn_verfahrensgebiet.functions.verfahrensgebiet_alkis_komplett.QMessageBox" +_PATCH_INFO = "sn_verfahrensgebiet.functions.verfahrensgebiet_alkis_komplett.info" +_PATCH_WARNING = "sn_verfahrensgebiet.functions.verfahrensgebiet_alkis_komplett.warning" +_PATCH_GET_VAR = "sn_verfahrensgebiet.functions.verfahrensgebiet_alkis_komplett.get_variable" + + +class TestVerfahrensgebietAlkisKomplett(unittest.TestCase): + + def _make_project(self): + proj = _DummyProject() + mock = MagicMock() + mock.instance.return_value = proj + # Leere mapLayersByName für Cleanup + proj._layers_by_name = {"Verfahrensgebiet": [], "TempVerfahrensgebiet": []} + return mock, proj + + # ------------------------------------------------------------------ # + # 1. VG-ALKIS liefert NONE → sofortiger Abbruch # + # ------------------------------------------------------------------ # + @patch(_PATCH_PROJECT) + @patch(_PATCH_VG_ALKIS) + def test_vg_alkis_none_aborts(self, mock_vg, mock_proj): + mock_proj.instance.return_value = _DummyProject() + mock_vg.return_value = (None, LoadStatus.NONE) + + enriched, flst, status = verfahrensgebiet_alkis_komplett(None) + + self.assertIsNone(enriched) + self.assertIsNone(flst) + self.assertEqual(status, LoadStatus.NONE) + + # ------------------------------------------------------------------ # + # 2. flurstueck_alkis liefert NONE → Abbruch nach Gemarkungsabfrage # + # ------------------------------------------------------------------ # + @patch(_PATCH_PROJECT) + @patch(_PATCH_FLST) + @patch(_PATCH_GEM_BY_BBOX) + @patch(_PATCH_VG_ALKIS) + def test_flst_none_aborts(self, mock_vg, mock_gem, mock_flst, mock_proj): + mock_proj.instance.return_value = _DummyProject() + vg_layer = _make_layer(name="Verfahrensgebiet") + mock_vg.return_value = (vg_layer, LoadStatus.FIRST) + mock_gem.return_value = [("1234", "Muster")] + mock_flst.return_value = (None, LoadStatus.NONE) + + enriched, flst, status = verfahrensgebiet_alkis_komplett(None) + + self.assertIsNone(enriched) + self.assertIsNone(flst) + self.assertEqual(status, LoadStatus.NONE) + + # ------------------------------------------------------------------ # + # 3. Keine Selektion → Warnung + (None, flst_layer, NONE) # + # ------------------------------------------------------------------ # + @patch(_PATCH_MSGBOX) + @patch(_PATCH_SELECT) + @patch(_PATCH_PROJECT) + @patch(_PATCH_FLST) + @patch(_PATCH_GEM_BY_BBOX) + @patch(_PATCH_VG_ALKIS) + def test_no_selection_returns_none_enriched( + self, mock_vg, mock_gem, mock_flst, mock_proj, mock_select, mock_msgbox + ): + mock_proj.instance.return_value = _DummyProject() + vg_layer = _make_layer(name="Verfahrensgebiet") + flst_layer = _make_layer(name="Flurstücke") + mock_vg.return_value = (vg_layer, LoadStatus.FIRST) + mock_gem.return_value = [] + mock_flst.return_value = (flst_layer, LoadStatus.FIRST) + mock_select.return_value = 0 # keine Selektion + + enriched, flst, status = verfahrensgebiet_alkis_komplett(None) + + self.assertIsNone(enriched) + self.assertIs(flst, flst_layer) + self.assertEqual(status, LoadStatus.NONE) + mock_msgbox.warning.assert_called_once() + + # ------------------------------------------------------------------ # + # 4. Vollständiger Erfolgsfall # + # ------------------------------------------------------------------ # + @patch(_PATCH_VKZ) + @patch(_PATCH_GET_VAR) + @patch(_PATCH_ERSTELLE) + @patch(_PATCH_SELECT) + @patch(_PATCH_PROJECT) + @patch(_PATCH_FLST) + @patch(_PATCH_GEM_BY_BBOX) + @patch(_PATCH_VG_ALKIS) + def test_success_full_workflow( + self, + mock_vg, mock_gem, mock_flst, mock_proj, + mock_select, mock_erstelle, mock_get_var, mock_vkz_fn, + ): + """Vollständiger Workflow → enriched Layer + flst Layer + FIRST.""" + proj_instance = _DummyProject() + mock_proj.instance.return_value = proj_instance + + vg_layer = _make_layer(name="Verfahrensgebiet") + flst_layer = _make_layer(name="Flurstücke", lid="FLST1") + enriched = _make_layer(name="Verfahrensgebiet", lid="VG_ENRICHED", valid=True) + + mock_vg.return_value = (vg_layer, LoadStatus.FIRST) + mock_gem.return_value = [("1234", "Muster")] + mock_flst.return_value = (flst_layer, LoadStatus.FIRST) + mock_select.return_value = 3 + mock_erstelle.return_value = (enriched, 3, 0) + mock_get_var.return_value = "VF-2024-001" + + result_enriched, result_flst, status = verfahrensgebiet_alkis_komplett(None) + + self.assertIs(result_enriched, enriched) + self.assertIs(result_flst, flst_layer) + self.assertEqual(status, LoadStatus.FIRST) + mock_vkz_fn.assert_called_once_with(enriched, "VF-2024-001") + + # ------------------------------------------------------------------ # + # 5. Kein VKZ → _fuege_vkz_feld_hinzu mit leerem String # + # ------------------------------------------------------------------ # + @patch(_PATCH_VKZ) + @patch(_PATCH_GET_VAR) + @patch(_PATCH_ERSTELLE) + @patch(_PATCH_SELECT) + @patch(_PATCH_PROJECT) + @patch(_PATCH_FLST) + @patch(_PATCH_GEM_BY_BBOX) + @patch(_PATCH_VG_ALKIS) + def test_no_vkz_variable_passes_empty_string( + self, + mock_vg, mock_gem, mock_flst, mock_proj, + mock_select, mock_erstelle, mock_get_var, mock_vkz_fn, + ): + proj_instance = _DummyProject() + mock_proj.instance.return_value = proj_instance + + vg_layer = _make_layer(name="Verfahrensgebiet") + flst_layer = _make_layer(name="Flurstücke") + enriched = _make_layer(name="Verfahrensgebiet", valid=True) + + mock_vg.return_value = (vg_layer, LoadStatus.FIRST) + mock_gem.return_value = [] + mock_flst.return_value = (flst_layer, LoadStatus.FIRST) + mock_select.return_value = 2 + mock_erstelle.return_value = (enriched, 2, 0) + mock_get_var.return_value = "" # keine Projektvariable + + verfahrensgebiet_alkis_komplett(None) + + mock_vkz_fn.assert_called_once_with(enriched, "") + + # ------------------------------------------------------------------ # + # 6. erstelle_verfahrensgebiet liefert ungültigen Layer # + # ------------------------------------------------------------------ # + @patch(_PATCH_MSGBOX) + @patch(_PATCH_ERSTELLE) + @patch(_PATCH_SELECT) + @patch(_PATCH_PROJECT) + @patch(_PATCH_FLST) + @patch(_PATCH_GEM_BY_BBOX) + @patch(_PATCH_VG_ALKIS) + def test_invalid_enriched_layer_returns_none( + self, mock_vg, mock_gem, mock_flst, mock_proj, mock_select, mock_erstelle, mock_msgbox + ): + mock_proj.instance.return_value = _DummyProject() + vg_layer = _make_layer() + flst_layer = _make_layer(name="Flurstücke") + bad_enriched = _make_layer(valid=False) + + mock_vg.return_value = (vg_layer, LoadStatus.FIRST) + mock_gem.return_value = [] + mock_flst.return_value = (flst_layer, LoadStatus.FIRST) + mock_select.return_value = 5 + mock_erstelle.return_value = (bad_enriched, 0, 5) + + enriched, flst, status = verfahrensgebiet_alkis_komplett(None) + + self.assertIsNone(enriched) + self.assertIs(flst, flst_layer) + self.assertEqual(status, LoadStatus.NONE) + mock_msgbox.critical.assert_called_once() + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_working_tab.py b/tests/test_working_tab.py index c2c23cc..5af3cd7 100644 --- a/tests/test_working_tab.py +++ b/tests/test_working_tab.py @@ -83,11 +83,13 @@ class TestWorkingTab(unittest.TestCase): tab = WorkingTab.__new__(WorkingTab) tab.btn_verf_alkis = _DummyButton() tab.btn_flurst_alkis = _DummyButton() + tab.btn_grenz_alkis = _DummyButton() tab.btn_verf_aus_auswahl = _DummyButton() tab.btn_verf_aus_liste = _DummyButton() tab.verf_layer_combo = _DummyCombo() - tab.haken_verf = object() - tab.haken_flurst = object() + tab.haken_verf = types.SimpleNamespace(setText=lambda _t: None, setStyleSheet=lambda _s: None) + tab.haken_flurst = types.SimpleNamespace(setText=lambda _t: None, setStyleSheet=lambda _s: None) + tab.haken_grenz = types.SimpleNamespace(setText=lambda _t: None, setStyleSheet=lambda _s: None) tab.setze_haken = lambda *_args, **_kwargs: None # Pruefmanager-Mock: verarbeite() wird im Test ggf. überschrieben tab.pruefmanager = types.SimpleNamespace( @@ -126,39 +128,83 @@ class TestWorkingTab(unittest.TestCase): @patch("sn_verfahrensgebiet.ui.tabs.working_tab.success") @patch("sn_verfahrensgebiet.ui.tabs.working_tab.apply_style") @patch("sn_verfahrensgebiet.ui.tabs.working_tab.QgsProject.instance") - @patch("sn_verfahrensgebiet.ui.tabs.working_tab.verfahrensgebiet_alkis") + @patch("sn_verfahrensgebiet.ui.tabs.working_tab.verfahrensgebiet_alkis_komplett") @patch("sn_verfahrensgebiet.ui.tabs.working_tab.iface") @patch("sn_verfahrensgebiet.ui.tabs.working_tab.set_verfahrensgebiet_layer_id") def test_handle_verf_alkis_success_path( self, mock_set_layer_id, mock_iface, - mock_alkis, + mock_alkis_komplett, mock_project_instance, mock_apply_style, mock_success, ): + """verfahrensgebiet_alkis_komplett → 3-Tupel (enriched, flst, FIRST).""" tab = self._make_tab() - layer = _DummyLayer(valid=True, layer_id="layer_ok") - mock_alkis.return_value = (layer, LoadStatus.FIRST) - mock_project_instance.return_value = _DummyProject() + enriched = _DummyLayer(valid=True, layer_id="enriched_ok") + flst = _DummyLayer(valid=True, layer_id="flst_ok") + mock_alkis_komplett.return_value = (enriched, flst, LoadStatus.FIRST) + proj = _DummyProject() + proj.mapLayersByName = lambda _n: [] # kein bestehender Flst-Layer + mock_project_instance.return_value = proj mock_iface.mapCanvas.return_value = _DummyCanvas() tab.handle_verf_alkis() - mock_apply_style.assert_called_once() + mock_apply_style.assert_called() mock_success.assert_called_once() - mock_set_layer_id.assert_called_with("layer_ok") + mock_set_layer_id.assert_called_with("enriched_ok") @patch("sn_verfahrensgebiet.ui.tabs.working_tab.set_verfahrensgebiet_layer_id") - @patch("sn_verfahrensgebiet.ui.tabs.working_tab.verfahrensgebiet_alkis", return_value=(None, LoadStatus.NONE)) + @patch( + "sn_verfahrensgebiet.ui.tabs.working_tab.verfahrensgebiet_alkis_komplett", + return_value=(None, None, LoadStatus.NONE), + ) def test_handle_verf_alkis_failure_clears_layer_var(self, _mock_alkis, mock_set_layer_id): + """Status NONE → Layer-Variable wird auf '' gesetzt.""" tab = self._make_tab() tab.handle_verf_alkis() mock_set_layer_id.assert_called_once_with("") + # ------------------------------------------------------------------ + # handle_grenz_alkis + # ------------------------------------------------------------------ + + @patch("sn_verfahrensgebiet.ui.tabs.working_tab.grenzpunkt_alkis") + def test_handle_grenz_alkis_success_sets_haken(self, mock_grenz): + """Grenzpunkte erfolgreich geladen → haken_grenz wird gesetzt.""" + tab = self._make_tab() + haken_calls = [] + tab.setze_haken = lambda lbl, v: haken_calls.append((lbl, v)) + + pta = _DummyLayer(valid=True) + gp = _DummyLayer(valid=True) + mock_grenz.return_value = (pta, gp) + + tab.handle_grenz_alkis() + + self.assertEqual(len(haken_calls), 1) + self.assertIs(haken_calls[0][0], tab.haken_grenz) + self.assertTrue(haken_calls[0][1]) + + @patch("sn_verfahrensgebiet.ui.tabs.working_tab.grenzpunkt_alkis") + def test_handle_grenz_alkis_none_result_sets_no_haken(self, mock_grenz): + """grenzpunkt_alkis liefert (None, None) → kein Haken.""" + tab = self._make_tab() + haken_calls = [] + tab.setze_haken = lambda lbl, v: haken_calls.append((lbl, v)) + + mock_grenz.return_value = (None, None) + + tab.handle_grenz_alkis() + + self.assertEqual(len(haken_calls), 0) + # Button wieder aktiviert (finally) + self.assertTrue(tab.btn_grenz_alkis.isEnabled()) + # ------------------------------------------------------------------ # handle_flurst_alkis # ------------------------------------------------------------------ diff --git a/ui/tabs/working_tab.py b/ui/tabs/working_tab.py index 1f47ee9..110f001 100644 --- a/ui/tabs/working_tab.py +++ b/ui/tabs/working_tab.py @@ -13,8 +13,12 @@ from sn_basis.modules.Pruefmanager import Pruefmanager from sn_basis.modules.pruef_ergebnis import pruef_ergebnis from qgis.utils import iface -from sn_verfahrensgebiet.functions.verfahrensgebiet_alkis import verfahrensgebiet_alkis, LoadStatus +from sn_verfahrensgebiet.functions.verfahrensgebiet_alkis import LoadStatus from sn_verfahrensgebiet.functions.flurstueck_alkis import flurstueck_alkis +from sn_verfahrensgebiet.functions.verfahrensgebiet_alkis_komplett import verfahrensgebiet_alkis_komplett +# TODO Grenzpunkt-Pipeline: Import reaktivieren, sobald grenzpunkt_alkis.py +# mit der neuen Pipeline (PunktortTA → istTeilVon → Einzel-URN) umgeschrieben wurde. +# from sn_verfahrensgebiet.functions.grenzpunkt_alkis import grenzpunkt_alkis from sn_verfahrensgebiet.functions.verfahrensgebiet_edit_service import ( pruefe_flurstueck_layer, pruefe_auswahl, @@ -87,6 +91,20 @@ class WorkingTab(QWidget): row_flurst_buttons.addWidget(self.btn_flurst_shape) layout.addLayout(row_flurst_buttons) + # Abschnitt 2b: Grenzpunkte + row_grenz_header = QHBoxLayout() + lbl_grenz = QLabel("Grenzpunkte") + self.haken_grenz = self._haken_label() + row_grenz_header.addWidget(lbl_grenz) + row_grenz_header.addWidget(self.haken_grenz) + row_grenz_header.addStretch() + layout.addLayout(row_grenz_header) + + row_grenz_buttons = QHBoxLayout() + self.btn_grenz_alkis = QPushButton("Grenzpunkte aus ALKIS laden") + row_grenz_buttons.addWidget(self.btn_grenz_alkis) + layout.addLayout(row_grenz_buttons) + # Abschnitt 3: Verfahrensgebiet erstellen row_verf_erstellen_header = QHBoxLayout() lbl_verf_erstellen = QLabel("Verfahrensgebiet erstellen") @@ -110,6 +128,7 @@ class WorkingTab(QWidget): # Signale verbinden self.btn_verf_alkis.clicked.connect(self.handle_verf_alkis) self.btn_flurst_alkis.clicked.connect(self.handle_flurst_alkis) + self.btn_grenz_alkis.clicked.connect(self.handle_grenz_alkis) self.btn_verf_aus_auswahl.clicked.connect(self.handle_verf_aus_auswahl) self.btn_verf_aus_liste.clicked.connect(self.handle_verf_aus_liste) # self.btn_verf_shape.clicked.connect(self.handle_verf_shape) @@ -118,36 +137,43 @@ class WorkingTab(QWidget): # Beim Init: aktuelle Verfahrensgebiet-Layer-ID laden und im Dropdown anzeigen self._restore_verf_layer_selection() - def handle_verf_alkis(self): + def handle_verf_alkis(self) -> None: + """Startet den vollständigen VG-Lade-Workflow (ALKIS → Gemarkungen → Flurstücke → Verschneidung). + + Delegiert an :func:`~sn_verfahrensgebiet.functions.verfahrensgebiet_alkis_komplett.verfahrensgebiet_alkis_komplett`. + Nach Erfolg werden enriched VG-Layer und Flurstücks-Layer ins Projekt eingefügt, + Styles angewendet und die UI-Elemente aktualisiert. + """ self._set_busy(self.btn_verf_alkis, True, "Aus ALKIS laden") try: - layer, status = verfahrensgebiet_alkis(self) - if not layer or not layer.isValid(): + enriched_layer, flst_layer, status = verfahrensgebiet_alkis_komplett(self) + + if status == LoadStatus.NONE or not enriched_layer: set_verfahrensgebiet_layer_id("") return - if status in (LoadStatus.FIRST, LoadStatus.RELOAD): - QgsProject.instance().addMapLayer(layer) - iface.mapCanvas().setExtent(layer.extent()) + project = QgsProject.instance() - apply_style(layer, "verfahrensgebiet.qml") - iface.mapCanvas().refresh() - self.setze_haken(self.haken_verf, True) + # Flurstücks-Layer ins Projekt (sofern neu geladen) + if flst_layer and flst_layer.isValid(): + existing_flst = project.mapLayersByName(flst_layer.name()) + if not existing_flst: + project.addMapLayer(flst_layer) + apply_style(flst_layer, "GIS_Flst_Beschriftung_ALKIS_NAS.qml") + self.setze_haken(self.haken_flurst, True) - self.verf_layer_combo.setLayer(layer) - self._on_verf_layer_changed(layer) + # Enriched VG-Layer ins Projekt + project.addMapLayer(enriched_layer) + apply_style(enriched_layer, "verfahrensgebiet.qml") + iface.mapCanvas().setExtent(enriched_layer.extent()) + iface.mapCanvas().refresh() + self.setze_haken(self.haken_verf, True) - msg = "erstmalig geladen" if status == LoadStatus.FIRST else "neu geladen" - success("Verfahrensgebiet", f"Layer wurde {msg}.", duration=5) + self.verf_layer_combo.setLayer(enriched_layer) + self._on_verf_layer_changed(enriched_layer) - elif status == LoadStatus.KEEP: - self.verf_layer_combo.setLayer(layer) - self._on_verf_layer_changed(layer) - info("Verfahrensgebiet", "Vorhandener Layer wurde behalten.", duration=4) - - elif status == LoadStatus.NONE: - self._on_verf_layer_changed(None) - warning("Verfahrensgebiet", "Layer konnte nicht geladen werden.", duration=None) + msg = "erstmalig geladen" if status == LoadStatus.FIRST else "neu geladen" + success("Verfahrensgebiet", f"Layer wurde {msg} (inkl. Flurstücksverschneidung).", duration=5) finally: self._set_busy(self.btn_verf_alkis, False, "Aus ALKIS laden") @@ -368,3 +394,39 @@ class WorkingTab(QWidget): "Die Funktion 'Aus Liste' steht in einer zukünftigen Version zur Verfügung.", parent=self, ) + + # ------------------------------------------------------------------ + # Grenzpunkt-Handler + # ------------------------------------------------------------------ + + def handle_grenz_alkis(self) -> None: + """Lädt Grenzpunkte (AX_PunktortTA + AX_Grenzpunkt) aus ALKIS. + + TODO Grenzpunkt-Pipeline: Implementierung ausstehend. + Neue Pipeline: VG-Geometrie → PunktortTA (spatial filter) → + istTeilVon-Attribute → Grenzpunkt per Einzel-URN laden. + Bis dahin ist dieser Handler deaktiviert. + """ + # TODO Grenzpunkt-Pipeline: Körper reaktivieren, sobald grenzpunkt_alkis.py + # mit der neuen Pipeline umgeschrieben wurde. + # + # self._set_busy(self.btn_grenz_alkis, True, "Grenzpunkte aus ALKIS laden") + # try: + # punktort_layer, grenzpunkt_layer = grenzpunkt_alkis(self) + # + # if grenzpunkt_layer and grenzpunkt_layer.isValid(): + # iface.mapCanvas().refresh() + # self.setze_haken(self.haken_grenz, True) + # success("Grenzpunkte", "Grenzpunkte erfolgreich geladen.", duration=5) + # elif punktort_layer and punktort_layer.isValid(): + # iface.mapCanvas().refresh() + # self.setze_haken(self.haken_grenz, True) + # info( + # "Grenzpunkte", + # "PunktortTA-Layer geladen. Grenzpunkt-IDs konnten nicht aufgelöst werden.", + # duration=6, + # ) + # finally: + # self._set_busy(self.btn_grenz_alkis, False, "Grenzpunkte aus ALKIS laden") + from sn_basis.functions.message_wrapper import warning as _warn + _warn("Grenzpunkte", "Grenzpunkt-Pipeline ist noch nicht implementiert.", duration=5)