verfahrensgebiet aus Flst
This commit is contained in:
+150
-22
@@ -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,33 +69,147 @@ 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
|
||||
# --- 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
|
||||
# --- 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",
|
||||
"Gemarkungsschlüssel eingeben (mehrere mit Komma trennen):",
|
||||
label,
|
||||
default_text=default_text,
|
||||
parent=tab_widget,
|
||||
)
|
||||
if not ok or not eingabe.strip():
|
||||
@@ -91,17 +225,11 @@ def flurstueck_alkis(tab_widget):
|
||||
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():
|
||||
if not layer or not layer.isValid() or layer.featureCount() == 0:
|
||||
if not ohne_dialog:
|
||||
QMessageBox.critical(
|
||||
tab_widget,
|
||||
"Fehler",
|
||||
|
||||
@@ -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)]
|
||||
@@ -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
|
||||
@@ -65,18 +65,23 @@ def verfahrensgebiet_alkis(tab_widget):
|
||||
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
|
||||
|
||||
@@ -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
|
||||
@@ -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()
|
||||
|
||||
@@ -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()
|
||||
@@ -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()
|
||||
@@ -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()
|
||||
+56
-10
@@ -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
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
+82
-20
@@ -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("<b>Grenzpunkte</b>")
|
||||
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("<b>Verfahrensgebiet erstellen</b>")
|
||||
@@ -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")
|
||||
# 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)
|
||||
|
||||
# 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)
|
||||
|
||||
self.verf_layer_combo.setLayer(layer)
|
||||
self._on_verf_layer_changed(layer)
|
||||
self.verf_layer_combo.setLayer(enriched_layer)
|
||||
self._on_verf_layer_changed(enriched_layer)
|
||||
|
||||
msg = "erstmalig geladen" if status == LoadStatus.FIRST else "neu geladen"
|
||||
success("Verfahrensgebiet", f"Layer wurde {msg}.", duration=5)
|
||||
|
||||
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)
|
||||
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)
|
||||
|
||||
Reference in New Issue
Block a user