242 lines
9.1 KiB
Python
242 lines
9.1 KiB
Python
"""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
|
||
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 __future__ import annotations
|
||
|
||
from typing import List, Optional, Tuple
|
||
|
||
from sn_basis.functions import QgsVectorLayer, QgsProject, QgsFeature
|
||
from sn_basis.functions import QMessageBox
|
||
from sn_basis.functions.dialog_wrapper import ask_text
|
||
from sn_basis.functions.message_wrapper import info as _info
|
||
|
||
# LoadStatus aus verfahrensgebiet_alkis wiederverwenden (identische Semantik)
|
||
from sn_verfahrensgebiet.functions.verfahrensgebiet_alkis import LoadStatus
|
||
|
||
ALKIS_NAS_URL = "https://geodienste.sachsen.de/aaa/public_alkis/nas/wfs"
|
||
FLST_TYPENAME = "adv:AX_Flurstueck"
|
||
FLST_LAYER_NAME = "Flurstücke"
|
||
|
||
|
||
def _check_existing_flurstueck_layer(tab_widget):
|
||
"""Prüft, ob ein Flurstück-Layer bereits im Projekt existiert.
|
||
|
||
Returns
|
||
-------
|
||
Tuple[Optional[QgsVectorLayer], LoadStatus]
|
||
Bei ``LoadStatus.KEEP`` enthält das Tuple den vorhandenen Layer,
|
||
bei ``LoadStatus.FIRST`` und ``LoadStatus.RELOAD`` ist es None.
|
||
"""
|
||
project = QgsProject.instance()
|
||
existing_layers = project.mapLayersByName(FLST_LAYER_NAME)
|
||
if not existing_layers:
|
||
return None, LoadStatus.FIRST
|
||
|
||
reply = QMessageBox.question(
|
||
tab_widget,
|
||
"Layer bereits vorhanden",
|
||
f"Der Layer '{FLST_LAYER_NAME}' existiert bereits.\n"
|
||
"Möchten Sie ihn löschen und neu laden?",
|
||
QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No,
|
||
QMessageBox.StandardButton.No,
|
||
)
|
||
if reply == QMessageBox.StandardButton.Yes:
|
||
for lyr in existing_layers:
|
||
project.removeMapLayer(lyr.id())
|
||
return None, LoadStatus.RELOAD
|
||
else:
|
||
return existing_layers[0], LoadStatus.KEEP
|
||
|
||
|
||
def _lade_flurstuecke_per_gemarkung(schluessel_liste: List[str]):
|
||
"""Lädt Flurstücke für mehrere Gemarkungen, eine WFS-Anfrage pro Schlüssel.
|
||
|
||
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]
|
||
Memory-Layer mit allen Flurstücken und Ladestatus.
|
||
Bei Fehlern ist der Layer ``None`` und der Status :attr:`LoadStatus.NONE`.
|
||
"""
|
||
# --- 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
|
||
|
||
# --- Schritt 2: Gemarkungsschlüssel bestimmen -------------------------
|
||
if ohne_dialog:
|
||
# Pipeline-Modus: vorgabe_schluessel direkt verwenden, kein Dialog
|
||
if not vorgabe_schluessel:
|
||
return None, LoadStatus.NONE
|
||
schluessel_liste = [s.strip() for s in vorgabe_schluessel if s.strip()]
|
||
if not schluessel_liste:
|
||
return None, LoadStatus.NONE
|
||
else:
|
||
# Interaktiver Modus: Dialog mit Vorschlag
|
||
default_text = ", ".join(vorgabe_schluessel) if vorgabe_schluessel else ""
|
||
hinweis = ""
|
||
if vorgabe_namen:
|
||
hinweis = " (Vorschlag: " + ", ".join(vorgabe_namen) + ")"
|
||
label = f"Gemarkungsschlüssel eingeben (mehrere mit Komma trennen):{hinweis}"
|
||
|
||
eingabe, ok = ask_text(
|
||
"Gemarkungsschlüssel",
|
||
label,
|
||
default_text=default_text,
|
||
parent=tab_widget,
|
||
)
|
||
if not ok or not eingabe.strip():
|
||
QMessageBox.warning(
|
||
tab_widget,
|
||
"Abgebrochen",
|
||
"Bitte mindestens einen gültigen Gemarkungsschlüssel eingeben.",
|
||
)
|
||
return None, LoadStatus.NONE
|
||
|
||
schluessel_liste = [s.strip() for s in eingabe.split(",") if s.strip()]
|
||
if not schluessel_liste:
|
||
QMessageBox.warning(tab_widget, "Fehler", "Keine gültigen Schlüssel gefunden.")
|
||
return None, LoadStatus.NONE
|
||
|
||
# --- Schritt 3: Flurstücke gemarkungsweise laden ----------------------
|
||
layer = _lade_flurstuecke_per_gemarkung(schluessel_liste)
|
||
|
||
if not layer or not layer.isValid() or layer.featureCount() == 0:
|
||
if not ohne_dialog:
|
||
QMessageBox.critical(
|
||
tab_widget,
|
||
"Fehler",
|
||
"Flurstück-Layer konnte nicht geladen werden.\n"
|
||
"Bitte Gemarkungsschlüssel und Netzwerkverbindung prüfen.",
|
||
)
|
||
return None, LoadStatus.NONE
|
||
|
||
return layer, status
|