diff --git a/functions/flurstueck_alkis.py b/functions/flurstueck_alkis.py
index 40876b2..e20f83c 100644
--- a/functions/flurstueck_alkis.py
+++ b/functions/flurstueck_alkis.py
@@ -1,20 +1,40 @@
"""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
-ALKIS_NAS_URL = "https://geodienste.sachsen.de/aaa/public_alkis/nas/wfs"
+ALKIS_SF_URL = "https://geodienste.sachsen.de/aaa/public_alkis/sf/wfs"
FLST_TYPENAME = "adv:AX_Flurstueck"
FLST_LAYER_NAME = "Flurstücke"
@@ -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_SF_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..cad0dac
--- /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_SF_URL = "https://geodienste.sachsen.de/aaa/public_alkis/sf/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_SF_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_SF_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/umringspruefung.py b/functions/umringspruefung.py
new file mode 100644
index 0000000..e498909
--- /dev/null
+++ b/functions/umringspruefung.py
@@ -0,0 +1,311 @@
+"""sn_verfahrensgebiet/functions/umringspruefung.py – Umringsprüfung (Variante 1).
+
+Vergleicht den Umring (Boundary) des BauRaumOderBodenordnungsrecht-Polygons
+mit den Grenzen der angrenzenden Flurstücke und hebt Abweichungen hervor.
+
+**Algorithmus (Variante 1 – linienbasierter Umringvergleich):**
+
+1. BROR-Polygon → Umringslinie via :func:`sn_basis.functions.qgiscore_wrapper.polygon_zu_linie`
+ (QGIS-3/4-kompatibel).
+2. Flurstücke, die den BROR-Umring berühren oder schneiden, werden gefiltert.
+3. Gefilterte Flurstücke → Umringslinien.
+4. Union aller Flurstücksumringe.
+5. ``BROR-Umring.difference(Flurstücksumrings-Union)`` → Abweichungen (rot).
+6. ``BROR-Umring.difference(rote_Segmente)`` → Übereinstimmungen (grün).
+7. Ergebnislayer mit Kategoriesymbologie (grün/rot) und Legendenbeschriftung.
+
+**Rückgabe:** ``(result_layer, conflict_count)``
+
+- ``result_layer``: Temporärer Linienlayer mit Feld ``typ``
+ (``"identisch"`` = grün, ``"Abweichung"`` = rot).
+- ``conflict_count``: Anzahl der roten Konfliktsegmente (Parts).
+"""
+from __future__ import annotations
+
+from typing import List, Optional, Tuple
+
+from sn_basis.functions.qgiscore_wrapper import (
+ QgsVectorLayer,
+ QgsFeature,
+ QgsField,
+ QgsGeometry,
+ polygon_zu_linie,
+)
+from sn_basis.functions.qt_wrapper import QVariant
+
+# ---------------------------------------------------------------------------
+# Konstanten
+# ---------------------------------------------------------------------------
+
+#: Geometrische Toleranz für den Linienvergleich (EPSG:25833 = Meter).
+#: Der BROR-Umring wird gegen die um diesen Betrag gepufferten Flurstücksumringe
+#: verglichen, um numerische Rundungsfehler aus ALKIS-Koordinaten abzufangen.
+LINIE_TOLERANZ_M: float = 0.0001
+
+ERGEBNIS_LAYER_NAME = "Umringsprüfung"
+FELD_TYP = "typ"
+WERT_IDENTISCH = "identisch"
+WERT_ABWEICHUNG = "Abweichung"
+
+
+# ---------------------------------------------------------------------------
+# Interne Hilfsfunktionen
+# ---------------------------------------------------------------------------
+
+def _zerlege_in_teile(geom: object) -> List[object]:
+ """Zerlegt eine (Multi-)LineString-Geometrie in einzelne Teile.
+
+ Parameters
+ ----------
+ geom :
+ ``QgsGeometry`` vom Typ LineString oder MultiLineString.
+
+ Returns
+ -------
+ List[QgsGeometry]
+ Liste einzelner Geometrieteile; leer wenn ``geom`` None oder leer ist.
+ """
+ if geom is None:
+ return []
+ try:
+ if getattr(geom, "isEmpty", lambda: True)():
+ return []
+ # asGeometryCollection() gibt eigene Kopien zurück – kein Ownership-Problem.
+ # QgsGeometry(part) über geom.parts() ist in QGIS 4.x aus den Python-Bindings
+ # entfernt worden und führt zu Absturz beim Rendern.
+ parts = geom.asGeometryCollection()
+ if parts:
+ return [p for p in parts if p is not None and not p.isEmpty()]
+ except Exception:
+ pass
+ # Fallback: Geometrie nicht zerlegbar → als Ganzes zurückgeben
+ try:
+ if not geom.isEmpty():
+ return [geom]
+ except Exception:
+ pass
+ return []
+
+
+def _erstelle_ergebnis_layer(crs_authid: str) -> object:
+ """Erzeugt den leeren Ergebnislayer mit Feld ``typ``.
+
+ Parameters
+ ----------
+ crs_authid :
+ CRS-Kennung des Quell-Layers (z. B. ``"EPSG:25833"``).
+
+ Returns
+ -------
+ QgsVectorLayer
+ Leerer Memory-Layer vom Typ LineString.
+ """
+ layer = QgsVectorLayer(
+ f"LineString?crs={crs_authid}", ERGEBNIS_LAYER_NAME, "memory"
+ )
+ pr = layer.dataProvider()
+ pr.addAttributes([QgsField(FELD_TYP, QVariant.String)])
+ layer.updateFields()
+ return layer
+
+
+def _setze_kategorien_symbologie(layer) -> None:
+ """Wendet Kategoriesymbologie an: grün = identisch, rot = Abweichung.
+
+ Wird bei fehlendem QGIS-Kontext still ignoriert (Unit-Test-Robustheit).
+ """
+ try:
+ from qgis.core import ( # type: ignore[import]
+ QgsCategorizedSymbolRenderer,
+ QgsRendererCategory,
+ QgsLineSymbol,
+ )
+
+ green_sym = QgsLineSymbol.createSimple(
+ {"color": "0,180,0,255", "line_width": "0.5"}
+ )
+ red_sym = QgsLineSymbol.createSimple(
+ {"color": "220,0,0,255", "line_width": "0.8"}
+ )
+ categories = [
+ QgsRendererCategory(
+ WERT_IDENTISCH,
+ green_sym,
+ "identisch (übereinstimmend)",
+ ),
+ QgsRendererCategory(
+ WERT_ABWEICHUNG,
+ red_sym,
+ "Abweichung (nicht in Flurstücken)",
+ ),
+ ]
+ renderer = QgsCategorizedSymbolRenderer(FELD_TYP, categories)
+ layer.setRenderer(renderer)
+ except Exception:
+ pass # Kein QGIS-Kontext (Tests) → Standarddarstellung behalten
+
+
+# ---------------------------------------------------------------------------
+# Öffentliche API
+# ---------------------------------------------------------------------------
+
+def umringspruefung(
+ bror_layer,
+ flst_layer,
+ progress=None,
+ linie_toleranz_m: float = LINIE_TOLERANZ_M,
+) -> Tuple[Optional[object], int]:
+ """Variante 1: Linienbasierter Umringvergleich.
+
+ Ermittelt, welche Teile des BROR-Umrings mit den Flurstücksgrenzen
+ übereinstimmen (grün) und welche keine Entsprechung haben (rot).
+
+ Parameters
+ ----------
+ bror_layer :
+ BauRaumOderBodenordnungsrecht-Memory-Layer (Polygon).
+ flst_layer :
+ Flurstücke-Layer (Polygon, WFS oder Memory).
+ progress :
+ Optionaler ``ProgressDialog`` aus
+ :mod:`sn_basis.functions.dialog_wrapper` für Fortschrittsanzeige
+ und Abbrucherkennung.
+ linie_toleranz_m : float
+ Geometrische Puffertoleranz für den Linienvergleich in
+ CRS-Einheiten (Standardwert :data:`LINIE_TOLERANZ_M` = 0.001 m).
+
+ Returns
+ -------
+ Tuple[Optional[QgsVectorLayer], int]
+ ``(result_layer, conflict_count)``
+
+ - ``result_layer``: Linienlayer mit grünen/roten Segmenten,
+ oder ``None`` bei Abbruch oder fehlgeschlagener Verarbeitung.
+ - ``conflict_count``: Anzahl der roten Konfliktsegmente (BROR-Teile
+ ohne Entsprechung in den Flurstücksgrenzen).
+ """
+
+ def _canceled() -> bool:
+ return progress is not None and progress.is_canceled()
+
+ def _schritt(label: str, wert: int) -> None:
+ if progress is not None:
+ progress.set_label(label)
+ progress.set_value(wert)
+
+ # ------------------------------------------------------------------
+ # 1. BROR-Geometrie und Umringslinie ermitteln
+ # ------------------------------------------------------------------
+ _schritt("Umringsprüfung: BROR-Umring ermitteln …", 0)
+
+ bror_feats = [
+ f for f in bror_layer.getFeatures()
+ if f.geometry() and not f.geometry().isEmpty()
+ ]
+ if not bror_feats:
+ return None, 0
+
+ bror_union = QgsGeometry.unaryUnion([f.geometry() for f in bror_feats])
+ if bror_union is None or bror_union.isEmpty():
+ return None, 0
+
+ bror_boundary = polygon_zu_linie(bror_union)
+ if bror_boundary is None or bror_boundary.isEmpty():
+ return None, 0
+
+ crs_authid = bror_layer.crs().authid()
+
+ if _canceled():
+ return None, 0
+
+ # ------------------------------------------------------------------
+ # 2. Relevante Flurstücke: nur jene, deren Geometrie den BROR-Umring
+ # berührt oder schneidet
+ # ------------------------------------------------------------------
+ _schritt("Umringsprüfung: Relevante Flurstücke filtern …", 1)
+
+ relevante_boundaries: List[object] = []
+ for feat in flst_layer.getFeatures():
+ if _canceled():
+ return None, 0
+ geom = feat.geometry()
+ if geom is None or geom.isEmpty():
+ continue
+ try:
+ if geom.intersects(bror_boundary):
+ boundary = polygon_zu_linie(geom)
+ if boundary is not None and not boundary.isEmpty():
+ relevante_boundaries.append(boundary)
+ except Exception:
+ continue
+
+ if relevante_boundaries:
+ flst_boundary_union = QgsGeometry.unaryUnion(relevante_boundaries)
+ else:
+ flst_boundary_union = None
+
+ if _canceled():
+ return None, 0
+
+ # ------------------------------------------------------------------
+ # 3. Segmentvergleich: Differenz und Übereinstimmung berechnen
+ # ------------------------------------------------------------------
+ _schritt("Umringsprüfung: Segmentvergleich …", 2)
+
+ ergebnis_layer = _erstelle_ergebnis_layer(crs_authid)
+ pr = ergebnis_layer.dataProvider()
+ feats_to_add: List[object] = []
+ conflict_count = 0
+
+ if flst_boundary_union is None or flst_boundary_union.isEmpty():
+ # Kein Flurstück berührt den BROR-Umring → alles ist Abweichung
+ for part in _zerlege_in_teile(bror_boundary):
+ feat = QgsFeature()
+ feat.setGeometry(part)
+ feat.setAttributes([WERT_ABWEICHUNG])
+ feats_to_add.append(feat)
+ conflict_count += 1
+ else:
+ # Flurstücksumringe leicht puffern für Toleranz-Matching
+ if linie_toleranz_m > 0:
+ flst_fuer_diff = flst_boundary_union.buffer(linie_toleranz_m, 4)
+ else:
+ flst_fuer_diff = flst_boundary_union
+
+ # Rote Segmente: BROR-Teile ohne Entsprechung in Flurstücksgrenzen
+ try:
+ red_geom = bror_boundary.difference(flst_fuer_diff)
+ except Exception:
+ red_geom = None
+
+ # Grüne Segmente: BROR-Teile, die nicht rot sind
+ try:
+ if red_geom and not red_geom.isEmpty():
+ green_geom = bror_boundary.difference(red_geom)
+ else:
+ green_geom = bror_boundary
+ except Exception:
+ green_geom = None
+
+ for part in _zerlege_in_teile(green_geom):
+ feat = QgsFeature()
+ feat.setGeometry(part)
+ feat.setAttributes([WERT_IDENTISCH])
+ feats_to_add.append(feat)
+
+ for part in _zerlege_in_teile(red_geom):
+ feat = QgsFeature()
+ feat.setGeometry(part)
+ feat.setAttributes([WERT_ABWEICHUNG])
+ feats_to_add.append(feat)
+ conflict_count += 1
+
+ if _canceled():
+ return None, 0
+
+ pr.addFeatures(feats_to_add)
+ ergebnis_layer.updateExtents()
+ _setze_kategorien_symbologie(ergebnis_layer)
+
+ _schritt("Umringsprüfung abgeschlossen.", 3)
+ return ergebnis_layer, conflict_count
diff --git a/functions/verfahrensgebiet_alkis.py b/functions/verfahrensgebiet_alkis.py
index a7f4acf..00e05e2 100644
--- a/functions/verfahrensgebiet_alkis.py
+++ b/functions/verfahrensgebiet_alkis.py
@@ -10,7 +10,7 @@ from sn_basis.functions import QMessageBox
from sn_basis.functions.qt_wrapper import QVariant
from sn_basis.functions.variable_wrapper import get_variable
-alkis_NAS_url = "https://geodienste.sachsen.de/aaa/public_alkis/nas/wfs"
+alkis_sf_url = "https://geodienste.sachsen.de/aaa/public_alkis/sf/wfs"
typename = "adv:AX_BauRaumOderBodenordnungsrecht"
class LoadStatus(Enum):
@@ -60,27 +60,32 @@ 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 = (
- f"url='{alkis_NAS_url}' typename='{typename}' srsname='EPSG:25833' "
+ f"url='{alkis_sf_url}' typename='{typename}' srsname='EPSG:25833' "
f"sql=SELECT * FROM AX_BauRaumOderBodenordnungsrecht "
f"WHERE bezeichnung LIKE '{verfahrensnummer}%'"
)
@@ -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..25ee5ea
--- /dev/null
+++ b/functions/verfahrensgebiet_alkis_komplett.py
@@ -0,0 +1,480 @@
+"""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, Flächenvergleich und
+optionaler Umringsprüfung:
+
+1. Verfahrensgebiet aus ``AX_BauRaumOderBodenordnungsrecht`` per VKZ laden.
+2. Gemarkungsschlüssel per Geometrie aus ``AX_Gemarkung`` vorschlagen.
+3. Flurstücke per Dialog (mit Vorschlag) laden.
+4. Flurstücke räumlich auf das VG einschränken (``selectByLocation``).
+5. Aus der Auswahl den enriched Verfahrensgebiet-Memory-Layer erzeugen.
+6. Flächenvergleich zwischen BROR-Objekt und abgeleitetem Verfahrensgebiet.
+
+ - **Übereinstimmung** (≤ Toleranz): modale OK-Meldung, Ablauf beendet.
+ - **Abweichung**: Ja/Nein-Abfrage für Detailprüfung.
+
+7. Optionale Umringsprüfung (Variante 1):
+
+ - BROR-Umring gegen Flurstücksgrenzen vergleichen.
+ - Ergebnislayer (grün/rot) ins Projekt einfügen.
+ - Abschlussmeldung mit Konfliktanzahl.
+
+Das BROR-Objekt (``AX_BauRaumOderBodenordnungsrecht``) bleibt bei Erfolg als
+temporärer Layer ``BauRaumOrdnungsrecht (ALKIS)`` im Projekt erhalten.
+
+**Rückgabe:**
+``Tuple[enriched_layer, flst_layer, bror_layer, LoadStatus]``
+
+**Abbruchverhalten:** Jeder Schritt, der :attr:`LoadStatus.NONE` zurückgibt
+oder kein Ergebnis liefert, bricht den Workflow ab. Der BROR-Layer wird nur
+bei Fehler/Abbruch entfernt (wenn er in diesem Lauf hinzugefügt wurde).
+"""
+
+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 warning
+from sn_basis.functions.variable_wrapper import get_variable
+from sn_basis.functions.dialog_wrapper import (
+ create_progress_dialog,
+ show_info_dialog,
+ ask_yes_no,
+)
+
+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,
+ VERF_LAYER_NAME,
+)
+from sn_verfahrensgebiet.functions.umringspruefung import umringspruefung
+
+# Feldname für das Verfahrenskennzeichen im enriched Layer
+FELD_VKZ = "verfahrensnummer"
+#: Anzeigename des BROR-Layers im Projekt
+BROR_LAYER_NAME = "BauRaumOrdnungsrecht (ALKIS)"
+#: Projektvariable für die Flächentoleranz (in m²); Default 0
+FLAECHEN_TOLERANZ_VARIABLE = "flaechen_toleranz_m2"
+
+
+
+# ---------------------------------------------------------------------------
+# Interne Hilfsfunktionen
+# ---------------------------------------------------------------------------
+
+def _entferne_hilfs_layer(project, layer_name: str) -> None:
+ """Entfernt alle Layer mit dem angegebenen Namen aus dem Projekt.
+
+ Wird verwendet, um temporäre WFS-Hilfs-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 innerhalb des Verfahrensgebiets liegen.
+
+ Verwendet ``processing.run("native:selectbylocation")`` mit Prädikat 6
+ (within). Flurstücke, die das VG nur berühren, werden ausgeschlossen.
+
+ Parameters
+ ----------
+ flst_layer :
+ Flurstücks-Layer (Quelle der Selektion).
+ vg_layer :
+ Verfahrensgebiet-Layer (Schnittmaske).
+
+ Returns
+ -------
+ int
+ Anzahl der selektierten Features; 0 bei Fehler oder leerem Ergebnis.
+ """
+ try:
+ import processing # type: ignore[import]
+
+ processing.run(
+ "native:selectbylocation",
+ {
+ "INPUT": flst_layer,
+ "PREDICATE": [6], # 6 = within; schließt reines Berühren aus
+ "INTERSECT": vg_layer,
+ "METHOD": 0, # 0 = neue Selektion
+ },
+ )
+ return flst_layer.selectedFeatureCount()
+
+ except Exception:
+ 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.
+
+ 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
+
+
+def _lese_flaechen_toleranz() -> float:
+ """Liest die Flächentoleranz aus der Projektvariable (Default: 0.0 m²).
+
+ Die Projektvariable :data:`FLAECHEN_TOLERANZ_VARIABLE` kann in den
+ Projekteigenschaften gesetzt werden. Fehlt sie oder ist sie ungültig,
+ wird 0.0 zurückgegeben (exakter Vergleich).
+
+ Returns
+ -------
+ float
+ Toleranz in m² (bei EPSG:25833 identisch mit den Flächen-Einheiten).
+ """
+ try:
+ raw = get_variable(FLAECHEN_TOLERANZ_VARIABLE)
+ if raw is not None and str(raw).strip():
+ return float(str(raw).strip())
+ except Exception:
+ pass
+ return 0.0
+
+
+def _flaeche_des_layers(layer) -> float:
+ """Summiert die Flächen aller Features eines Layers (in CRS-Einheiten²).
+
+ Parameters
+ ----------
+ layer :
+ Beliebiger Polygon-Layer.
+
+ Returns
+ -------
+ float
+ Summe aller Feature-Flächen; 0.0 wenn keine Features vorhanden.
+ """
+ total = 0.0
+ for feat in layer.getFeatures():
+ try:
+ geom = feat.geometry()
+ if geom and not geom.isEmpty():
+ total += geom.area()
+ except Exception:
+ pass
+ return total
+
+
+def _vergleiche_flaechen(
+ bror_layer, vg_layer, toleranz_m2: float = 0.0
+) -> Tuple[float, float, bool]:
+ """Vergleicht Flächen von BROR- und Verfahrensgebiet-Layer.
+
+ Parameters
+ ----------
+ bror_layer :
+ BauRaumOrdnungsrecht-Layer.
+ vg_layer :
+ Enriched Verfahrensgebiet-Layer.
+ toleranz_m2 :
+ Maximale zulässige Flächendifferenz in m².
+
+ Returns
+ -------
+ Tuple[float, float, bool]
+ ``(bror_flaeche, vg_flaeche, stimmt_ueberein)``
+ """
+ bror_flaeche = _flaeche_des_layers(bror_layer)
+ vg_flaeche = _flaeche_des_layers(vg_layer)
+ stimmt_ueberein = abs(bror_flaeche - vg_flaeche) <= toleranz_m2
+ return bror_flaeche, vg_flaeche, stimmt_ueberein
+
+
+# ---------------------------------------------------------------------------
+# Öffentliche API
+# ---------------------------------------------------------------------------
+
+def verfahrensgebiet_alkis_komplett(
+ tab_widget,
+) -> Tuple[Optional[object], Optional[object], Optional[object], LoadStatus]:
+ """Vollständiger VG-Lade-Workflow: ALKIS-VG → Gemarkungen →
+ Flurstücke → Verschneidung → Flächenvergleich → opt. Umringsprüfung.
+
+ Schrittfolge:
+
+ 1. ``verfahrensgebiet_alkis(tab_widget)`` – BROR-Objekt als Memory-Layer.
+ 2. ``gemarkung_schluessel_und_namen_von_vg_geometrie(geom)`` – Schlüssel.
+ 3. ``flurstueck_alkis(...)`` – Flurstücke laden.
+ 4. ``_select_by_location(...)`` – räumliche Selektion.
+ 5. ``erstelle_verfahrensgebiet_aus_auswahl(...)`` – enriched Layer.
+ 6. ``_vergleiche_flaechen(...)`` – Flächenvergleich als Gate.
+ 7. Ggf. ``umringspruefung(...)`` nach Nutzerbestätigung.
+
+ Der BROR-Layer (``BauRaumOrdnungsrecht (ALKIS)``) bleibt bei Erfolg
+ im Projekt. Bei Abbruch/Fehler wird er entfernt, falls er in diesem
+ Lauf hinzugefügt wurde.
+
+ Parameters
+ ----------
+ tab_widget :
+ Eltern-Widget für alle Dialoge.
+
+ Returns
+ -------
+ Tuple[Optional[enriched_layer], Optional[flst_layer], Optional[bror_layer], LoadStatus]
+ - ``enriched_layer``: Memory-Layer „Verfahrensgebiet".
+ - ``flst_layer``: Flurstücke-Layer (bleibt im Projekt).
+ - ``bror_layer``: BROR-Layer ``BauRaumOrdnungsrecht (ALKIS)``
+ (bleibt bei Erfolg im Projekt).
+ - :class:`LoadStatus`: ``FIRST``, ``RELOAD`` oder ``NONE``.
+ """
+ project = QgsProject.instance()
+ alkis_vg_layer = None
+ bror_was_added_by_us = False
+ abgeschlossen = False
+
+ # Fortschrittsdialog für den Basisworkflow (7 Schritte)
+ progress = create_progress_dialog(
+ 7, "Verfahrensgebiet laden", "Initialisierung …"
+ )
+
+ try:
+ # ------------------------------------------------------------------
+ # Schritt 1: VG aus ALKIS laden (BROR-Objekt → Memory-Layer)
+ # ------------------------------------------------------------------
+ progress.set_label("Schritt 1/6: Verfahrensgebiet aus ALKIS laden …")
+ progress.set_value(0)
+ if progress.is_canceled():
+ return None, None, None, LoadStatus.NONE
+
+ alkis_vg_layer, vg_status, _ = verfahrensgebiet_alkis(tab_widget)
+
+ if vg_status == LoadStatus.NONE or not alkis_vg_layer:
+ return None, None, None, LoadStatus.NONE
+
+ # BROR-Layer umbenennen, um Namenskonflikt mit enriched VG-Layer zu vermeiden
+ alkis_vg_layer.setName(BROR_LAYER_NAME)
+
+ if vg_status in (LoadStatus.FIRST, LoadStatus.RELOAD):
+ project.addMapLayer(alkis_vg_layer)
+ bror_was_added_by_us = True
+
+ progress.set_value(1)
+
+ # ------------------------------------------------------------------
+ # Schritt 2: Gemarkungsschlüssel per Geometrie-Spatial-Join ermitteln
+ # ------------------------------------------------------------------
+ progress.set_label("Schritt 2/6: Gemarkungsschlüssel ermitteln …")
+ if progress.is_canceled():
+ return None, None, None, LoadStatus.NONE
+
+ bror_feats = list(alkis_vg_layer.getFeatures())
+ vg_geom = bror_feats[0].geometry() if bror_feats else None
+
+ schluessel_namen = (
+ gemarkung_schluessel_und_namen_von_vg_geometrie(vg_geom)
+ if vg_geom else []
+ )
+ 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,
+ )
+
+ progress.set_value(2)
+
+ # ------------------------------------------------------------------
+ # Schritt 3: Flurstücke laden (mit Nutzer-Dialog, Vorschlag befüllt)
+ # ------------------------------------------------------------------
+ progress.set_label("Schritt 3/6: Flurstücke aus ALKIS laden …")
+ if progress.is_canceled():
+ return None, None, None, LoadStatus.NONE
+
+ flst_layer, flst_status = flurstueck_alkis(
+ tab_widget,
+ vorgabe_schluessel=vorgabe_schluessel,
+ vorgabe_namen=vorgabe_namen if vorgabe_namen else None,
+ ohne_dialog=True,
+ )
+
+ if flst_status == LoadStatus.NONE or not flst_layer:
+ return None, None, None, LoadStatus.NONE
+
+ progress.set_value(3)
+
+ # ------------------------------------------------------------------
+ # Schritt 4: Räumliche Selektion
+ # ------------------------------------------------------------------
+ progress.set_label("Schritt 4/6: Flurstücke räumlich selektieren …")
+ if progress.is_canceled():
+ return None, None, None, LoadStatus.NONE
+
+ 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, None, LoadStatus.NONE
+
+ progress.set_value(4)
+
+ # ------------------------------------------------------------------
+ # Schritt 5: Enriched Memory-Layer aus Auswahl erzeugen
+ # ------------------------------------------------------------------
+ progress.set_label("Schritt 5/6: Verfahrensgebiet-Layer erzeugen …")
+ if progress.is_canceled():
+ return None, None, None, LoadStatus.NONE
+
+ 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, None, LoadStatus.NONE
+
+ vkz = get_variable("verfahrensnummer") or ""
+ _fuege_vkz_feld_hinzu(enriched_layer, vkz)
+
+ progress.set_value(5)
+
+ # ------------------------------------------------------------------
+ # Schritt 6: Flächenvergleich als Gate
+ # ------------------------------------------------------------------
+ progress.set_label("Schritt 6/6: Flächenvergleich …")
+ if progress.is_canceled():
+ return None, None, None, LoadStatus.NONE
+
+ toleranz = _lese_flaechen_toleranz()
+ bror_flaeche, vg_flaeche, stimmt_ueberein = _vergleiche_flaechen(
+ alkis_vg_layer, enriched_layer, toleranz
+ )
+
+ progress.set_value(6)
+
+ if stimmt_ueberein:
+ progress.set_value(7)
+ progress.close()
+ show_info_dialog(
+ "Flächenvergleich",
+ "Verfahrensgebiet mit erwarteter Geometrie erzeugt",
+ parent=tab_widget,
+ )
+ abgeschlossen = True
+ return enriched_layer, flst_layer, alkis_vg_layer, vg_status
+
+ # ------------------------------------------------------------------
+ # Flächenabweichung → Nutzerabfrage für Detailprüfung
+ # ------------------------------------------------------------------
+ progress.close()
+ diff = abs(bror_flaeche - vg_flaeche)
+ starten = ask_yes_no(
+ "Abweichung erkannt",
+ f"Abweichung zwischen BauRaumOrdnungsrecht-Objekt und abgeleitetem "
+ f"Verfahrensgebiet erkannt. Detailprüfung starten?\n\n"
+ f"BauRaumOrdnungsrecht: {bror_flaeche:.2f} m²\n"
+ f"Verfahrensgebiet: {vg_flaeche:.2f} m²\n"
+ f"Differenz: {diff:.2f} m²",
+ default=True,
+ parent=tab_widget,
+ )
+
+ if not starten:
+ abgeschlossen = True
+ return enriched_layer, flst_layer, alkis_vg_layer, vg_status
+
+ # ------------------------------------------------------------------
+ # Umringsprüfung (Variante 1: linienbasierter Umringvergleich)
+ # ------------------------------------------------------------------
+ umring_progress = create_progress_dialog(
+ 3, "Detailprüfung (Umringe)", "Starte Umringsprüfung …"
+ )
+
+ ergebnis_layer, konflikt_anzahl = umringspruefung(
+ alkis_vg_layer, flst_layer, progress=umring_progress
+ )
+
+ if umring_progress.is_canceled():
+ umring_progress.close()
+ abgeschlossen = True
+ return enriched_layer, flst_layer, alkis_vg_layer, vg_status
+
+ umring_progress.set_value(3)
+ umring_progress.close()
+
+ if ergebnis_layer and ergebnis_layer.isValid():
+ project.addMapLayer(ergebnis_layer)
+
+ show_info_dialog(
+ "Detailprüfung",
+ f"Detailprüfung abgeschlossen. "
+ f"Es wurden {konflikt_anzahl} Konfliktstellen identifiziert.",
+ parent=tab_widget,
+ )
+
+ abgeschlossen = True
+ return enriched_layer, flst_layer, alkis_vg_layer, vg_status
+
+ finally:
+ # WFS-Temp-Layer stets aus dem Projekt entfernen
+ _entferne_hilfs_layer(project, "TempVerfahrensgebiet")
+ # BROR-Layer bei Fehler/Abbruch entfernen (nur wenn in diesem Lauf hinzugefügt)
+ if not abgeschlossen and bror_was_added_by_us and alkis_vg_layer is not None:
+ 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..e19d134
--- /dev/null
+++ b/tests/test_verfahrensgebiet_alkis_komplett.py
@@ -0,0 +1,608 @@
+"""sn_verfahrensgebiet/tests/test_verfahrensgebiet_alkis_komplett.py
+
+Unit-Tests für :mod:`sn_verfahrensgebiet.functions.verfahrensgebiet_alkis_komplett`.
+
+Getestete Szenarien:
+
+Erfolgsfall (Flächenübereinstimmung):
+- Vollständiger Workflow → enriched Layer + flst_layer + bror_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, NONE)
+- flurstueck_alkis gibt LoadStatus.NONE → (None, None, None, NONE)
+- Keine Flurstücke selektiert → (None, flst_layer, None, NONE) + Warnung
+- erstelle_verfahrensgebiet_aus_auswahl liefert ungültigen Layer → (None, flst_layer, None, NONE)
+- Progress-Abbruch nach Schritt 1 → sofortiger sauberer Rückgabestatus
+
+Flächenvergleich:
+- Übereinstimmung → show_info_dialog, keine Umringsprüfung
+- Abweichung + Nein → kein Umringsprüfungs-Aufruf
+- Abweichung + Ja → umringspruefung aufgerufen, Ergebnislayer ins Projekt
+
+Cleanup:
+- BROR-Layer wird bei Erfolg NICHT entfernt
+- BROR-Layer wird bei Fehler entfernt (wenn in diesem Lauf hinzugefügt)
+- TempVerfahrensgebiet wird stets entfernt
+
+Hilfsfunktionen:
+- _select_by_location: processing-Fallback liefert 0 → keine Exception
+- _fuege_vkz_feld_hinzu: leeres VKZ wird ignoriert
+- _vergleiche_flaechen: korrekte Toleranzlogik
+- _lese_flaechen_toleranz: Default 0.0, parst Projektvariable
+"""
+
+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.verfahrensgebiet_alkis import LoadStatus
+from sn_verfahrensgebiet.functions.verfahrensgebiet_alkis_komplett import (
+ verfahrensgebiet_alkis_komplett,
+ _select_by_location,
+ _fuege_vkz_feld_hinzu,
+ _vergleiche_flaechen,
+ _lese_flaechen_toleranz,
+)
+
+
+# ---------------------------------------------------------------------------
+# 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, area=1000.0):
+ """Erzeugt einen konfigurierten Mock-Layer mit getFeatures-Support."""
+ 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
+
+ mock_geom = MagicMock()
+ mock_geom.isEmpty.return_value = False
+ mock_geom.area.return_value = area
+ mock_geom.boundary.return_value = MagicMock(isEmpty=lambda: True)
+ mock_feat = MagicMock()
+ mock_feat.geometry.return_value = mock_geom
+ layer.getFeatures.return_value = [mock_feat]
+
+ mock_crs = MagicMock()
+ mock_crs.authid.return_value = "EPSG:25833"
+ layer.crs.return_value = mock_crs
+ 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
+
+
+def _make_progress_mock(canceled=False):
+ prog = MagicMock()
+ prog.is_canceled.return_value = canceled
+ return prog
+
+
+# ---------------------------------------------------------------------------
+# Patch-Konstanten
+# ---------------------------------------------------------------------------
+
+_MOD = "sn_verfahrensgebiet.functions.verfahrensgebiet_alkis_komplett"
+_PATCH_VG_ALKIS = f"{_MOD}.verfahrensgebiet_alkis"
+_PATCH_GEM = f"{_MOD}.gemarkung_schluessel_und_namen_von_vg_geometrie"
+_PATCH_FLST = f"{_MOD}.flurstueck_alkis"
+_PATCH_SELECT = f"{_MOD}._select_by_location"
+_PATCH_ERSTELLE = f"{_MOD}.erstelle_verfahrensgebiet_aus_auswahl"
+_PATCH_VKZ = f"{_MOD}._fuege_vkz_feld_hinzu"
+_PATCH_PROJECT = f"{_MOD}.QgsProject"
+_PATCH_MSGBOX = f"{_MOD}.QMessageBox"
+_PATCH_WARNING = f"{_MOD}.warning"
+_PATCH_GET_VAR = f"{_MOD}.get_variable"
+_PATCH_PROGRESS = f"{_MOD}.create_progress_dialog"
+_PATCH_VERGLEICHE = f"{_MOD}._vergleiche_flaechen"
+_PATCH_SHOW_INFO = f"{_MOD}.show_info_dialog"
+_PATCH_ASK_YES_NO = f"{_MOD}.ask_yes_no"
+_PATCH_UMRING = f"{_MOD}.umringspruefung"
+
+
+# ---------------------------------------------------------------------------
+# Tests: _select_by_location
+# ---------------------------------------------------------------------------
+
+class TestSelectByLocation(unittest.TestCase):
+
+ def test_processing_not_available_returns_zero(self):
+ flst, vg = MagicMock(), 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):
+ flst = MagicMock()
+ flst.selectedFeatureCount.return_value = 7
+ vg = MagicMock()
+ mock_proc = MagicMock()
+ with patch.dict("sys.modules", {"processing": mock_proc}):
+ result = _select_by_location(flst, vg)
+ self.assertEqual(result, 7)
+
+ def test_processing_exception_returns_zero(self):
+ flst, vg = MagicMock(), MagicMock()
+ mock_proc = MagicMock()
+ mock_proc.run.side_effect = RuntimeError("Kein QGIS")
+ with patch.dict("sys.modules", {"processing": mock_proc}):
+ 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):
+ layer = MagicMock()
+ _fuege_vkz_feld_hinzu(layer, "")
+ layer.dataProvider.assert_not_called()
+
+ def test_none_vkz_does_nothing(self):
+ layer = MagicMock()
+ _fuege_vkz_feld_hinzu(layer, None)
+ layer.dataProvider.assert_not_called()
+
+ def test_vkz_field_added(self):
+ 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: _vergleiche_flaechen / _lese_flaechen_toleranz
+# ---------------------------------------------------------------------------
+
+class TestFlaechenvergleich(unittest.TestCase):
+
+ def _layer(self, area: float):
+ layer = MagicMock()
+ geom = MagicMock()
+ geom.isEmpty.return_value = False
+ geom.area.return_value = area
+ feat = MagicMock()
+ feat.geometry.return_value = geom
+ layer.getFeatures.return_value = [feat]
+ return layer
+
+ def test_gleiche_flaeche_stimmt_ueberein(self):
+ self.assertTrue(_vergleiche_flaechen(self._layer(1234.5), self._layer(1234.5), 0.0)[2])
+
+ def test_verschiedene_flaeche_stimmt_nicht_ueberein(self):
+ self.assertFalse(_vergleiche_flaechen(self._layer(1234.5), self._layer(1200.0), 0.0)[2])
+
+ def test_differenz_innerhalb_toleranz(self):
+ self.assertTrue(_vergleiche_flaechen(self._layer(1234.5), self._layer(1234.8), 1.0)[2])
+
+ def test_differenz_ausserhalb_toleranz(self):
+ self.assertFalse(_vergleiche_flaechen(self._layer(1234.5), self._layer(1236.0), 1.0)[2])
+
+ def test_flaechen_werte_korrekt(self):
+ bror_f, vg_f, _ = _vergleiche_flaechen(self._layer(800.0), self._layer(750.0))
+ self.assertAlmostEqual(bror_f, 800.0)
+ self.assertAlmostEqual(vg_f, 750.0)
+
+ @patch(f"{_MOD}.get_variable", return_value=None)
+ def test_lese_toleranz_default_ist_null(self, _):
+ self.assertEqual(_lese_flaechen_toleranz(), 0.0)
+
+ @patch(f"{_MOD}.get_variable", return_value="2.5")
+ def test_lese_toleranz_parst_string(self, _):
+ self.assertAlmostEqual(_lese_flaechen_toleranz(), 2.5)
+
+ @patch(f"{_MOD}.get_variable", return_value="ungueltig")
+ def test_lese_toleranz_ungueltig_fallback_null(self, _):
+ self.assertEqual(_lese_flaechen_toleranz(), 0.0)
+
+
+# ---------------------------------------------------------------------------
+# Tests: verfahrensgebiet_alkis_komplett (Hauptworkflow)
+# ---------------------------------------------------------------------------
+
+class TestVerfahrensgebietAlkisKomplett(unittest.TestCase):
+
+ # ------------------------------------------------------------------ #
+ # Abbruchpfade
+ # ------------------------------------------------------------------ #
+
+ @patch(_PATCH_PROGRESS)
+ @patch(_PATCH_PROJECT)
+ @patch(_PATCH_VG_ALKIS)
+ def test_vg_alkis_none_aborts(self, mock_vg, mock_proj, mock_prog):
+ mock_proj.instance.return_value = _DummyProject()
+ mock_vg.return_value = (None, LoadStatus.NONE, [])
+ mock_prog.return_value = _make_progress_mock()
+
+ enriched, flst, bror, status = verfahrensgebiet_alkis_komplett(None)
+
+ self.assertIsNone(enriched)
+ self.assertIsNone(flst)
+ self.assertIsNone(bror)
+ self.assertEqual(status, LoadStatus.NONE)
+
+ @patch(_PATCH_PROGRESS)
+ @patch(_PATCH_PROJECT)
+ @patch(_PATCH_GEM)
+ @patch(_PATCH_VG_ALKIS)
+ def test_flst_none_aborts(self, mock_vg, mock_gem, mock_proj, mock_prog):
+ mock_proj.instance.return_value = _DummyProject()
+ vg_layer = _make_layer(name="BauRaumOrdnungsrecht (ALKIS)")
+ mock_vg.return_value = (vg_layer, LoadStatus.FIRST, [])
+ mock_gem.return_value = [("1234", "Muster")]
+ mock_prog.return_value = _make_progress_mock()
+
+ with patch(_PATCH_FLST, return_value=(None, LoadStatus.NONE)):
+ enriched, flst, bror, status = verfahrensgebiet_alkis_komplett(None)
+
+ self.assertIsNone(enriched)
+ self.assertIsNone(flst)
+ self.assertEqual(status, LoadStatus.NONE)
+
+ @patch(_PATCH_PROGRESS)
+ @patch(_PATCH_MSGBOX)
+ @patch(_PATCH_SELECT)
+ @patch(_PATCH_PROJECT)
+ @patch(_PATCH_GEM)
+ @patch(_PATCH_FLST)
+ @patch(_PATCH_VG_ALKIS)
+ def test_no_selection_returns_none_enriched(
+ self, mock_vg, mock_flst, mock_gem, mock_proj, mock_select, mock_msgbox, mock_prog
+ ):
+ mock_proj.instance.return_value = _DummyProject()
+ vg_layer = _make_layer(name="BauRaumOrdnungsrecht (ALKIS)")
+ 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
+ mock_prog.return_value = _make_progress_mock()
+
+ enriched, flst, bror, status = verfahrensgebiet_alkis_komplett(None)
+
+ self.assertIsNone(enriched)
+ self.assertIs(flst, flst_layer)
+ self.assertIsNone(bror)
+ self.assertEqual(status, LoadStatus.NONE)
+ mock_msgbox.warning.assert_called_once()
+
+ @patch(_PATCH_PROGRESS)
+ @patch(_PATCH_MSGBOX)
+ @patch(_PATCH_ERSTELLE)
+ @patch(_PATCH_SELECT)
+ @patch(_PATCH_PROJECT)
+ @patch(_PATCH_GEM)
+ @patch(_PATCH_FLST)
+ @patch(_PATCH_VG_ALKIS)
+ def test_invalid_enriched_layer_returns_none(
+ self, mock_vg, mock_flst, mock_gem, mock_proj, mock_select, mock_erstelle, mock_msgbox, mock_prog
+ ):
+ mock_proj.instance.return_value = _DummyProject()
+ vg_layer = _make_layer(name="BauRaumOrdnungsrecht (ALKIS)")
+ 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)
+ mock_prog.return_value = _make_progress_mock()
+
+ enriched, flst, bror, status = verfahrensgebiet_alkis_komplett(None)
+
+ self.assertIsNone(enriched)
+ self.assertIs(flst, flst_layer)
+ self.assertIsNone(bror)
+ self.assertEqual(status, LoadStatus.NONE)
+ mock_msgbox.critical.assert_called_once()
+
+ @patch(_PATCH_PROGRESS)
+ @patch(_PATCH_PROJECT)
+ @patch(_PATCH_VG_ALKIS)
+ def test_progress_cancel_after_step1_returns_none(self, mock_vg, mock_proj, mock_prog):
+ """Abbruch nach Schritt 1 → sofortiger sauberer Rückgabestatus."""
+ mock_proj.instance.return_value = _DummyProject()
+ vg_layer = _make_layer(name="BauRaumOrdnungsrecht (ALKIS)")
+ mock_vg.return_value = (vg_layer, LoadStatus.FIRST, [])
+
+ prog = MagicMock()
+ prog.is_canceled.side_effect = [False, True] # 1. Aufruf False, 2. True
+ mock_prog.return_value = prog
+
+ enriched, flst, bror, status = verfahrensgebiet_alkis_komplett(None)
+
+ self.assertIsNone(enriched)
+ self.assertEqual(status, LoadStatus.NONE)
+
+ # ------------------------------------------------------------------ #
+ # Flächenvergleich – Übereinstimmung
+ # ------------------------------------------------------------------ #
+
+ @patch(_PATCH_PROGRESS)
+ @patch(_PATCH_SHOW_INFO)
+ @patch(_PATCH_VERGLEICHE)
+ @patch(_PATCH_VKZ)
+ @patch(_PATCH_GET_VAR)
+ @patch(_PATCH_ERSTELLE)
+ @patch(_PATCH_SELECT)
+ @patch(_PATCH_PROJECT)
+ @patch(_PATCH_GEM)
+ @patch(_PATCH_FLST)
+ @patch(_PATCH_VG_ALKIS)
+ def test_success_flaechen_match_zeigt_info(
+ self, mock_vg, mock_flst, mock_gem, mock_proj, mock_select,
+ mock_erstelle, mock_get_var, mock_vkz, mock_vergl, mock_info, mock_prog,
+ ):
+ """Flächenübereinstimmung → show_info_dialog mit erwartetem Text."""
+ proj_inst = _DummyProject()
+ mock_proj.instance.return_value = proj_inst
+
+ vg_layer = _make_layer(name="BauRaumOrdnungsrecht (ALKIS)")
+ 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"
+ mock_vergl.return_value = (1234.5, 1234.5, True)
+ mock_prog.return_value = _make_progress_mock()
+
+ result_enriched, result_flst, result_bror, status = verfahrensgebiet_alkis_komplett(None)
+
+ self.assertIs(result_enriched, enriched)
+ self.assertIs(result_flst, flst_layer)
+ self.assertIs(result_bror, vg_layer)
+ self.assertEqual(status, LoadStatus.FIRST)
+ mock_vkz.assert_called_once_with(enriched, "VF-2024-001")
+ mock_info.assert_called_once()
+ self.assertIn("erwarteter Geometrie", mock_info.call_args[0][1])
+
+ @patch(_PATCH_PROGRESS)
+ @patch(_PATCH_SHOW_INFO)
+ @patch(_PATCH_VERGLEICHE)
+ @patch(_PATCH_VKZ)
+ @patch(_PATCH_GET_VAR)
+ @patch(_PATCH_ERSTELLE)
+ @patch(_PATCH_SELECT)
+ @patch(_PATCH_PROJECT)
+ @patch(_PATCH_GEM)
+ @patch(_PATCH_FLST)
+ @patch(_PATCH_VG_ALKIS)
+ def test_no_vkz_variable_passes_empty_string(
+ self, mock_vg, mock_flst, mock_gem, mock_proj, mock_select,
+ mock_erstelle, mock_get_var, mock_vkz, mock_vergl, mock_info, mock_prog,
+ ):
+ mock_proj.instance.return_value = _DummyProject()
+ vg_layer = _make_layer(name="BauRaumOrdnungsrecht (ALKIS)")
+ 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 = ""
+ mock_vergl.return_value = (1000.0, 1000.0, True)
+ mock_prog.return_value = _make_progress_mock()
+
+ verfahrensgebiet_alkis_komplett(None)
+
+ mock_vkz.assert_called_once_with(enriched, "")
+
+ # ------------------------------------------------------------------ #
+ # Flächenvergleich – Abweichung + Nein
+ # ------------------------------------------------------------------ #
+
+ @patch(_PATCH_PROGRESS)
+ @patch(_PATCH_UMRING)
+ @patch(_PATCH_ASK_YES_NO)
+ @patch(_PATCH_VERGLEICHE)
+ @patch(_PATCH_VKZ)
+ @patch(_PATCH_GET_VAR)
+ @patch(_PATCH_ERSTELLE)
+ @patch(_PATCH_SELECT)
+ @patch(_PATCH_PROJECT)
+ @patch(_PATCH_GEM)
+ @patch(_PATCH_FLST)
+ @patch(_PATCH_VG_ALKIS)
+ def test_flaechen_abweichung_nein_kein_umring(
+ self, mock_vg, mock_flst, mock_gem, mock_proj, mock_select,
+ mock_erstelle, mock_get_var, mock_vkz, mock_vergl, mock_ask, mock_umring, mock_prog,
+ ):
+ """Abweichung + Nein → umringspruefung wird NICHT aufgerufen."""
+ mock_proj.instance.return_value = _DummyProject()
+ vg_layer = _make_layer(name="BauRaumOrdnungsrecht (ALKIS)")
+ 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 = 3
+ mock_erstelle.return_value = (enriched, 3, 0)
+ mock_get_var.return_value = ""
+ mock_vergl.return_value = (1000.0, 800.0, False)
+ mock_ask.return_value = False
+ mock_prog.return_value = _make_progress_mock()
+
+ result_enriched, _, _, status = verfahrensgebiet_alkis_komplett(None)
+
+ mock_umring.assert_not_called()
+ self.assertIs(result_enriched, enriched)
+ self.assertEqual(status, LoadStatus.FIRST)
+
+ # ------------------------------------------------------------------ #
+ # Flächenvergleich – Abweichung + Ja → Umringsprüfung
+ # ------------------------------------------------------------------ #
+
+ @patch(_PATCH_PROGRESS)
+ @patch(_PATCH_SHOW_INFO)
+ @patch(_PATCH_UMRING)
+ @patch(_PATCH_ASK_YES_NO)
+ @patch(_PATCH_VERGLEICHE)
+ @patch(_PATCH_VKZ)
+ @patch(_PATCH_GET_VAR)
+ @patch(_PATCH_ERSTELLE)
+ @patch(_PATCH_SELECT)
+ @patch(_PATCH_PROJECT)
+ @patch(_PATCH_GEM)
+ @patch(_PATCH_FLST)
+ @patch(_PATCH_VG_ALKIS)
+ def test_flaechen_abweichung_ja_startet_umringspruefung(
+ self, mock_vg, mock_flst, mock_gem, mock_proj, mock_select,
+ mock_erstelle, mock_get_var, mock_vkz, mock_vergl, mock_ask,
+ mock_umring, mock_info, mock_prog,
+ ):
+ """Abweichung + Ja → umringspruefung aufgerufen, Ergebnislayer ins Projekt,
+ Abschlussmeldung mit Konfliktanzahl."""
+ proj_inst = _DummyProject()
+ mock_proj.instance.return_value = proj_inst
+
+ vg_layer = _make_layer(name="BauRaumOrdnungsrecht (ALKIS)")
+ flst_layer = _make_layer(name="Flurstücke")
+ enriched = _make_layer(name="Verfahrensgebiet", valid=True)
+ ergebnis_layer = _make_layer(name="Umringsprüfung", lid="UMRING_L")
+
+ mock_vg.return_value = (vg_layer, LoadStatus.FIRST, [])
+ mock_gem.return_value = []
+ 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 = ""
+ mock_vergl.return_value = (1000.0, 800.0, False)
+ mock_ask.return_value = True
+ mock_umring.return_value = (ergebnis_layer, 4)
+ mock_prog.return_value = _make_progress_mock()
+
+ result_enriched, _, _, status = verfahrensgebiet_alkis_komplett(None)
+
+ mock_umring.assert_called_once()
+ self.assertIn(ergebnis_layer, proj_inst.added)
+ mock_info.assert_called_once()
+ self.assertIn("4", mock_info.call_args[0][1])
+ self.assertIn("Konfliktstellen", mock_info.call_args[0][1])
+ self.assertEqual(status, LoadStatus.FIRST)
+
+ # ------------------------------------------------------------------ #
+ # Cleanup-Verhalten
+ # ------------------------------------------------------------------ #
+
+ @patch(_PATCH_PROGRESS)
+ @patch(_PATCH_SHOW_INFO)
+ @patch(_PATCH_VERGLEICHE)
+ @patch(_PATCH_VKZ)
+ @patch(_PATCH_GET_VAR)
+ @patch(_PATCH_ERSTELLE)
+ @patch(_PATCH_SELECT)
+ @patch(_PATCH_PROJECT)
+ @patch(_PATCH_GEM)
+ @patch(_PATCH_FLST)
+ @patch(_PATCH_VG_ALKIS)
+ def test_bror_layer_bleibt_bei_erfolg_im_projekt(
+ self, mock_vg, mock_flst, mock_gem, mock_proj, mock_select,
+ mock_erstelle, mock_get_var, mock_vkz, mock_vergl, mock_info, mock_prog,
+ ):
+ """Bei Erfolg darf der BROR-Layer NICHT entfernt werden."""
+ proj_inst = _DummyProject()
+ mock_proj.instance.return_value = proj_inst
+
+ vg_layer = _make_layer(name="BauRaumOrdnungsrecht (ALKIS)", lid="BROR_ID")
+ 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 = 3
+ mock_erstelle.return_value = (enriched, 3, 0)
+ mock_get_var.return_value = ""
+ mock_vergl.return_value = (1000.0, 1000.0, True)
+ mock_prog.return_value = _make_progress_mock()
+
+ verfahrensgebiet_alkis_komplett(None)
+
+ self.assertNotIn("BROR_ID", proj_inst.removed)
+
+ @patch(_PATCH_PROGRESS)
+ @patch(_PATCH_PROJECT)
+ @patch(_PATCH_GEM)
+ @patch(_PATCH_VG_ALKIS)
+ def test_bror_layer_wird_bei_fehler_entfernt(
+ self, mock_vg, mock_gem, mock_proj, mock_prog
+ ):
+ """Bei Fehler (flst_alkis NONE) wird der BROR-Layer aus dem Projekt entfernt."""
+ proj_inst = _DummyProject()
+ mock_proj.instance.return_value = proj_inst
+
+ vg_layer = _make_layer(name="BauRaumOrdnungsrecht (ALKIS)", lid="BROR_ERR")
+ mock_vg.return_value = (vg_layer, LoadStatus.FIRST, [])
+ mock_gem.return_value = []
+ mock_prog.return_value = _make_progress_mock()
+
+ with patch(_PATCH_FLST, return_value=(None, LoadStatus.NONE)):
+ verfahrensgebiet_alkis_komplett(None)
+
+ self.assertIn("BROR_ERR", proj_inst.removed)
+
+
+if __name__ == "__main__":
+ unittest.main()
+
diff --git a/tests/test_working_tab.py b/tests/test_working_tab.py
index c2c23cc..f2fc707 100644
--- a/tests/test_working_tab.py
+++ b/tests/test_working_tab.py
@@ -18,9 +18,10 @@ from sn_verfahrensgebiet.ui.tabs.working_tab import WorkingTab, LoadStatus
class _DummyLayer:
- def __init__(self, valid=True, layer_id="L1"):
+ def __init__(self, valid=True, layer_id="L1", layer_name="DummyLayer"):
self._valid = valid
self._id = layer_id
+ self._name = layer_name
def isValid(self):
return self._valid
@@ -28,6 +29,9 @@ class _DummyLayer:
def id(self):
return self._id
+ def name(self):
+ return self._name
+
def extent(self):
return object()
@@ -83,11 +87,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 +132,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 → 4-Tupel (enriched, flst, bror, 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, None, 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, 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..645f38f 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,48 @@ 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 → Flächenvergleich → opt. Umringsprüfung).
+
+ Delegiert an :func:`~sn_verfahrensgebiet.functions.verfahrensgebiet_alkis_komplett.verfahrensgebiet_alkis_komplett`.
+ Nach Erfolg werden enriched VG-Layer, Flurstücks-Layer und BROR-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, bror_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)
+ # BROR-Layer: einfache Darstellung setzen (roter Umring, transparent)
+ if bror_layer and bror_layer.isValid():
+ _setze_bror_stil(bror_layer)
- msg = "erstmalig geladen" if status == LoadStatus.FIRST else "neu geladen"
- success("Verfahrensgebiet", f"Layer wurde {msg}.", duration=5)
+ # 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)
- elif status == LoadStatus.KEEP:
- self.verf_layer_combo.setLayer(layer)
- self._on_verf_layer_changed(layer)
- info("Verfahrensgebiet", "Vorhandener Layer wurde behalten.", duration=4)
+ self.verf_layer_combo.setLayer(enriched_layer)
+ self._on_verf_layer_changed(enriched_layer)
- 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 +399,73 @@ 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)
+
+
+# ---------------------------------------------------------------------------
+# Modul-Hilfsfunktionen
+# ---------------------------------------------------------------------------
+
+def _setze_bror_stil(layer) -> None:
+ """Setzt eine einfache Darstellung für den BauRaumOrdnungsrecht-Layer.
+
+ Roter Umring, halbtransparente Füllung; kein QGIS-Kontext → stilles Ignorieren.
+
+ Parameters
+ ----------
+ layer :
+ BauRaumOrdnungsrecht-Memory-Layer (Polygon).
+ """
+ try:
+ from qgis.core import ( # type: ignore[import]
+ QgsSimpleFillSymbolLayer,
+ QgsFillSymbol,
+ QgsSingleSymbolRenderer,
+ )
+ from qgis.PyQt.QtGui import QColor # type: ignore[import]
+ from qgis.PyQt.QtCore import Qt # type: ignore[import]
+
+ fill = QgsSimpleFillSymbolLayer()
+ fill.setColor(QColor(220, 50, 50, 40)) # halbtransparent rot
+ fill.setStrokeColor(QColor(220, 50, 50))
+ fill.setStrokeWidth(0.8)
+ symbol = QgsFillSymbol([fill])
+ layer.setRenderer(QgsSingleSymbolRenderer(symbol))
+ layer.triggerRepaint()
+ except Exception:
+ pass # Kein QGIS-Kontext (Tests, Mock-Modus) → Standarddarstellung