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