Michael Otto
148d4dfa15
- Fortschrittsdialog beim Laden aus ALKIS schließt jetzt zuverlässig bei jedem Abbruch (progress.close() zentral im finally-Block). - Fortschrittsanzeige in zwei Dialoge aufgeteilt: Suche (1 Schritt) läuft und schließt separat, bevor der eigentliche Workflow (Gemarkung/ Flurstücke/Vergleich, 6 Schritte) startet – auch während des BROR- Auswahldialogs läuft kein störender Hintergrund-Fortschrittsbalken mehr. - verfahrensgebiet_alkis() in Suche-/Auswahl-Phasen aufgeteilt, damit mehrere gefundene BauRaumOderBodenordnungsrecht-Objekte gemeinsam ausgewählt und verschmolzen werden können (öffentliche Signatur unverändert). - Neu: Landkreis-weite Suche als Fallback, wenn keine Verfahrensnummer gesetzt ist (Landkreiskennzahl aus Projektvariable), mit Mehrfach- auswahl/"Alle auswählen" und "Nur laden"-Option (überspringt den restlichen Workflow). Gruppierung der Auswahl nach VKZ, damit je Verfahren ein eigenes Feature entsteht statt eines Gesamt-Blobs. - Tests entsprechend angepasst/ergänzt (29 Tests in test_verfahrensgebiet_alkis_komplett.py, davon 3 neu für den Landkreis-Fallback).
354 lines
14 KiB
Python
354 lines
14 KiB
Python
"""sn_verfahrensgebiet/functions/verfahrensgebiet_alkis.py – ALKIS-Integration.
|
||
Lädt Verfahrensgebiete aus dem ALKIS-WFS-Dienst.
|
||
"""
|
||
|
||
from enum import Enum
|
||
from sn_basis.functions import (
|
||
QgsVectorLayer, QgsProject, QgsFeature, QgsField, QgsGeometry
|
||
)
|
||
from sn_basis.functions import QMessageBox
|
||
from sn_basis.functions.qt_wrapper import QVariant
|
||
from sn_basis.functions.variable_wrapper import get_variable
|
||
from sn_basis.functions.dialog_wrapper import ask_bror_kombinationen, ask_landkreis_objekte
|
||
|
||
alkis_sf_url = "https://geodienste.sachsen.de/aaa/public_alkis/sf/wfs"
|
||
typename = "adv:AX_BauRaumOderBodenordnungsrecht"
|
||
|
||
#: Attribute, die aus dem BROR-WFS-Dienst geladen und im Memory-Layer gespeichert werden.
|
||
BROR_ATTRIBUTE_FELDER = [
|
||
"artderfestlegung",
|
||
"ausfuehrendestelle_ax_dienststelle_schluessel_stelle",
|
||
"adv_name",
|
||
"bezeichnung",
|
||
"datumanordnung",
|
||
"datumbesitzeinweisung",
|
||
"datumrechtskraft",
|
||
"datumabgabe",
|
||
]
|
||
|
||
class LoadStatus(Enum):
|
||
"""Status eines Layer-Ladevorgangs.
|
||
|
||
Attributes:
|
||
NONE: Nichts geladen
|
||
FIRST: Erstmalig geladen
|
||
RELOAD: Neu geladen
|
||
KEEP: Vorhandener Layer behalten
|
||
"""
|
||
NONE = "none" # nichts geladen
|
||
FIRST = "first" # erstmalig geladen
|
||
RELOAD = "reload" # neu geladen
|
||
KEEP = "keep" # vorhandener Layer behalten
|
||
APPEND = "append" # Features an vorhandenen Layer angehängt
|
||
|
||
|
||
def _check_existing_layer(tab_widget):
|
||
"""Prüft, ob ein Verfahrensgebiet-Layer bereits existiert.
|
||
|
||
Args:
|
||
tab_widget: Parent-Widget für Dialoge
|
||
|
||
Returns:
|
||
Tuple[Optional[QgsVectorLayer], LoadStatus]: Layer und Status
|
||
"""
|
||
project = QgsProject.instance()
|
||
existing_layers = project.mapLayersByName("Verfahrensgebiet")
|
||
if not existing_layers:
|
||
return None, LoadStatus.FIRST
|
||
|
||
reply = QMessageBox.question(
|
||
tab_widget,
|
||
"Layer bereits vorhanden",
|
||
"Der Layer 'Verfahrensgebiet' 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 _suche_verfahrensgebiet_features(tab_widget, skip_existing_check: bool = False):
|
||
"""Phase 1 ("Suchen"): vorhandenen Layer prüfen, WFS abfragen, nach
|
||
(adv_name, bezeichnung) gruppieren.
|
||
|
||
Args:
|
||
tab_widget: Parent-Widget für Dialoge und Meldungen
|
||
skip_existing_check: Überspringt die Nachfrage bei bereits vorhandenem
|
||
Layer (immer neu laden).
|
||
|
||
Returns:
|
||
Tuple[Optional[QgsVectorLayer], LoadStatus, dict, Optional[str]]:
|
||
``(fertig_layer, status, kombinationen, crs)``.
|
||
Bei ``KEEP`` oder Fehler/Abbruch ist ``fertig_layer`` bereits das
|
||
Endergebnis (ggf. ``None``) und ``kombinationen`` leer – Phase 2
|
||
entfällt dann. Bei Erfolg mit mindestens einer Kombination ist
|
||
``fertig_layer`` ``None`` und ``kombinationen``/``crs`` sind für
|
||
Phase 2 (:func:`_baue_layer_aus_kombinationen`) befüllt.
|
||
"""
|
||
verfahrensnummer = get_variable("verfahrensnummer")
|
||
if not verfahrensnummer:
|
||
QMessageBox.critical(tab_widget, "Fehler",
|
||
"In den Projekteigenschaften ist noch keine Verfahrensnummer eingetragen!")
|
||
return None, LoadStatus.NONE, {}, None
|
||
|
||
# Vorhandenen Layer prüfen (überspringbar für die Shape-Pipeline)
|
||
status = LoadStatus.FIRST
|
||
if not skip_existing_check:
|
||
existing, status = _check_existing_layer(tab_widget)
|
||
if status == LoadStatus.KEEP:
|
||
return existing, status, {}, None
|
||
|
||
# WFS mit Filter laden – nur erste 5 Stellen der VKZ für den Vergleich verwenden,
|
||
# da die Daten im Dienst ggf. abweichende Suffixe führen.
|
||
vkz_prefix = verfahrensnummer[:5]
|
||
wfs_uri_find = (
|
||
f"url='{alkis_sf_url}' typename='{typename}' srsname='EPSG:25833' "
|
||
f"sql=SELECT * FROM AX_BauRaumOderBodenordnungsrecht "
|
||
f"WHERE bezeichnung LIKE '{vkz_prefix}%'"
|
||
)
|
||
temp_layer = QgsVectorLayer(wfs_uri_find, "TempVerfahrensgebiet", "WFS")
|
||
|
||
if not temp_layer.isValid():
|
||
QMessageBox.critical(tab_widget, "Fehler", "Layer konnte nicht geladen werden.")
|
||
return None, LoadStatus.NONE, {}, 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, {}, None
|
||
|
||
# Objekte nach (adv_name, bezeichnung) gruppieren
|
||
kombinationen: dict[tuple[str, str], list] = {}
|
||
for feat in features:
|
||
try:
|
||
adv_name_val = str(feat.attribute("adv_name") or "")
|
||
bezeichnung_val = str(feat.attribute("bezeichnung") or "")
|
||
except Exception:
|
||
adv_name_val = ""
|
||
bezeichnung_val = ""
|
||
key = (adv_name_val, bezeichnung_val)
|
||
if key not in kombinationen:
|
||
kombinationen[key] = []
|
||
kombinationen[key].append(feat)
|
||
|
||
return None, status, kombinationen, temp_layer.crs().authid()
|
||
|
||
|
||
def _baue_layer_aus_kombinationen(tab_widget, kombinationen: dict, crs: str, status: "LoadStatus"):
|
||
"""Phase 2 ("Auswählen + Laden"): bei mehreren Kombinationen Auswahldialog
|
||
zeigen, Geometrien der gewählten Kombination(en) verschmelzen, Memory-Layer bauen.
|
||
|
||
Args:
|
||
tab_widget: Parent-Widget für Dialoge und Meldungen
|
||
kombinationen: Ergebnis aus :func:`_suche_verfahrensgebiet_features`
|
||
crs: CRS-Authid aus :func:`_suche_verfahrensgebiet_features`
|
||
status: Lade-Status aus Phase 1 (FIRST/RELOAD), wird durchgereicht
|
||
|
||
Returns:
|
||
Tuple[Optional[QgsVectorLayer], LoadStatus, List[QgsFeature]]
|
||
"""
|
||
if len(kombinationen) == 1:
|
||
selected_feats = next(iter(kombinationen.values()))
|
||
else:
|
||
# Mehrere Kombinationen → Nutzerauswahl, ein oder mehrere Objekte
|
||
# können gemeinsam ausgewählt und geladen (verschmolzen) werden.
|
||
key_list = sorted(kombinationen.keys())
|
||
selected_keys = ask_bror_kombinationen(tab_widget, key_list)
|
||
if not selected_keys:
|
||
return None, LoadStatus.NONE, []
|
||
selected_feats = [feat for key in selected_keys for feat in kombinationen[key]]
|
||
|
||
union_geom = QgsGeometry.unaryUnion([f.geometry() for f in selected_feats])
|
||
if union_geom is None or union_geom.isEmpty():
|
||
QMessageBox.critical(tab_widget, "Fehler", "Keine Geometrien zum Verschmelzen gefunden.")
|
||
return None, LoadStatus.NONE, []
|
||
|
||
# Memory-Layer erzeugen
|
||
memory_layer = QgsVectorLayer(f"Polygon?crs={crs}", "Verfahrensgebiet", "memory")
|
||
pr = memory_layer.dataProvider()
|
||
pr.addAttributes([QgsField(attr, QVariant.String) for attr in BROR_ATTRIBUTE_FELDER])
|
||
memory_layer.updateFields()
|
||
|
||
# Attributwerte aus dem ersten Feature der gewählten Kombination übernehmen
|
||
first_feat = selected_feats[0]
|
||
attrs = []
|
||
for attr in BROR_ATTRIBUTE_FELDER:
|
||
try:
|
||
val = first_feat.attribute(attr)
|
||
if val is None:
|
||
attrs.append("")
|
||
elif type(val).__name__ == "QDate":
|
||
# QDate-Objekte als DD.MM.YYYY formatieren
|
||
attrs.append(val.toString("dd.MM.yyyy"))
|
||
else:
|
||
attrs.append(str(val))
|
||
except Exception:
|
||
attrs.append("")
|
||
|
||
feat_union = QgsFeature()
|
||
feat_union.setGeometry(union_geom)
|
||
feat_union.setAttributes(attrs)
|
||
pr.addFeatures([feat_union])
|
||
memory_layer.updateExtents()
|
||
|
||
# Status: FIRST oder RELOAD; selected_feats für Gemarkungsermittlung weitergeben
|
||
return memory_layer, status, selected_feats
|
||
|
||
|
||
def _suche_landkreis_kombinationen(tab_widget, landkreis_kennzahl: str, skip_existing_check: bool = False):
|
||
"""Fallback-Suche über einen ganzen Landkreis (statt einer Verfahrensnummer).
|
||
|
||
Wie :func:`_suche_verfahrensgebiet_features`, aber mit WFS-Filter über die
|
||
Landkreiskennzahl (``bezeichnung LIKE '{landkreis_kennzahl}%'``) statt
|
||
über den 5-stelligen VKZ-Präfix. Gruppierung der Rohfeatures identisch:
|
||
nach exaktem (adv_name, bezeichnung).
|
||
|
||
Returns:
|
||
Tuple[Optional[QgsVectorLayer], LoadStatus, dict, Optional[str]]:
|
||
Gleiche Bedeutung wie bei :func:`_suche_verfahrensgebiet_features`.
|
||
"""
|
||
status = LoadStatus.FIRST
|
||
if not skip_existing_check:
|
||
existing, status = _check_existing_layer(tab_widget)
|
||
if status == LoadStatus.KEEP:
|
||
return existing, status, {}, None
|
||
|
||
wfs_uri_find = (
|
||
f"url='{alkis_sf_url}' typename='{typename}' srsname='EPSG:25833' "
|
||
f"sql=SELECT * FROM AX_BauRaumOderBodenordnungsrecht "
|
||
f"WHERE bezeichnung LIKE '{landkreis_kennzahl}%'"
|
||
)
|
||
temp_layer = QgsVectorLayer(wfs_uri_find, "TempVerfahrensgebiet", "WFS")
|
||
|
||
if not temp_layer.isValid():
|
||
QMessageBox.critical(tab_widget, "Fehler", "Layer konnte nicht geladen werden.")
|
||
return None, LoadStatus.NONE, {}, None
|
||
|
||
features = list(temp_layer.getFeatures())
|
||
if not features:
|
||
QMessageBox.critical(tab_widget, "Fehler",
|
||
f"Für Landkreis {landkreis_kennzahl} wurden keine Objekte im WFS gefunden.")
|
||
return None, LoadStatus.NONE, {}, None
|
||
|
||
kombinationen: dict[tuple[str, str], list] = {}
|
||
for feat in features:
|
||
try:
|
||
adv_name_val = str(feat.attribute("adv_name") or "")
|
||
bezeichnung_val = str(feat.attribute("bezeichnung") or "")
|
||
except Exception:
|
||
adv_name_val = ""
|
||
bezeichnung_val = ""
|
||
key = (adv_name_val, bezeichnung_val)
|
||
if key not in kombinationen:
|
||
kombinationen[key] = []
|
||
kombinationen[key].append(feat)
|
||
|
||
return None, status, kombinationen, temp_layer.crs().authid()
|
||
|
||
|
||
def _baue_layer_aus_landkreis_auswahl(tab_widget, kombinationen: dict, crs: str, status: "LoadStatus"):
|
||
"""Auswahl + Bau für den Landkreis-Fallback.
|
||
|
||
Zeigt :func:`~sn_basis.functions.dialog_wrapper.ask_landkreis_objekte`
|
||
(Mehrfachauswahl inkl. "Alle auswählen" und "Nur laden"-Checkbox). Die
|
||
ausgewählten Kombinationen werden anhand der ersten 5 Zeichen von
|
||
``bezeichnung`` (= die eigentliche VKZ) gruppiert; je Gruppe wird EIN
|
||
Feature erzeugt (Geometrien der Gruppe verschmolzen). Damit entsteht bei
|
||
Auswahl mehrerer/aller Objekte eines Landkreises ein Layer mit mehreren
|
||
Features – einem je tatsächlichem Verfahren, nicht ein Gesamt-Blob.
|
||
|
||
Args:
|
||
tab_widget: Parent-Widget für Dialoge und Meldungen
|
||
kombinationen: Ergebnis aus :func:`_suche_landkreis_kombinationen`
|
||
crs: CRS-Authid aus :func:`_suche_landkreis_kombinationen`
|
||
status: Lade-Status aus Phase 1 (FIRST/RELOAD), wird durchgereicht
|
||
|
||
Returns:
|
||
Tuple[Optional[QgsVectorLayer], LoadStatus, bool]:
|
||
``(layer, status, nur_laden)``.
|
||
"""
|
||
key_list = sorted(kombinationen.keys())
|
||
selected_keys, nur_laden = ask_landkreis_objekte(tab_widget, key_list)
|
||
if not selected_keys:
|
||
return None, LoadStatus.NONE, False
|
||
|
||
# Nach VKZ (erste 5 Zeichen der bezeichnung) gruppieren
|
||
vkz_gruppen: dict[str, list] = {}
|
||
for key in selected_keys:
|
||
_, bezeichnung = key
|
||
vkz = bezeichnung[:5]
|
||
vkz_gruppen.setdefault(vkz, []).extend(kombinationen[key])
|
||
|
||
memory_layer = QgsVectorLayer(f"Polygon?crs={crs}", "Verfahrensgebiet", "memory")
|
||
pr = memory_layer.dataProvider()
|
||
pr.addAttributes([QgsField(attr, QVariant.String) for attr in BROR_ATTRIBUTE_FELDER])
|
||
memory_layer.updateFields()
|
||
|
||
neue_features = []
|
||
for vkz, feats in sorted(vkz_gruppen.items()):
|
||
union_geom = QgsGeometry.unaryUnion([f.geometry() for f in feats])
|
||
if union_geom is None or union_geom.isEmpty():
|
||
continue
|
||
|
||
first_feat = feats[0]
|
||
attrs = []
|
||
for attr in BROR_ATTRIBUTE_FELDER:
|
||
try:
|
||
val = first_feat.attribute(attr)
|
||
if val is None:
|
||
attrs.append("")
|
||
elif type(val).__name__ == "QDate":
|
||
attrs.append(val.toString("dd.MM.yyyy"))
|
||
else:
|
||
attrs.append(str(val))
|
||
except Exception:
|
||
attrs.append("")
|
||
|
||
feat_out = QgsFeature()
|
||
feat_out.setGeometry(union_geom)
|
||
feat_out.setAttributes(attrs)
|
||
neue_features.append(feat_out)
|
||
|
||
if not neue_features:
|
||
QMessageBox.critical(tab_widget, "Fehler", "Keine Geometrien zum Verschmelzen gefunden.")
|
||
return None, LoadStatus.NONE, False
|
||
|
||
pr.addFeatures(neue_features)
|
||
memory_layer.updateExtents()
|
||
|
||
return memory_layer, status, nur_laden
|
||
|
||
|
||
def verfahrensgebiet_alkis(tab_widget, skip_existing_check: bool = False):
|
||
"""Lädt Verfahrensgebiet aus ALKIS-WFS basierend auf Verfahrensnummer.
|
||
|
||
Args:
|
||
tab_widget: Parent-Widget für Dialoge und Meldungen
|
||
|
||
Returns:
|
||
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.
|
||
|
||
Ruft intern nacheinander :func:`_suche_verfahrensgebiet_features` (Phase 1)
|
||
und ggf. :func:`_baue_layer_aus_kombinationen` (Phase 2) auf – aufgeteilt,
|
||
damit ``verfahrensgebiet_alkis_komplett()`` dazwischen einen eigenen
|
||
Fortschrittsschritt anzeigen kann. Verhalten/Signatur dieser Funktion
|
||
selbst ist unverändert.
|
||
"""
|
||
layer, status, kombinationen, crs = _suche_verfahrensgebiet_features(
|
||
tab_widget, skip_existing_check
|
||
)
|
||
if layer is not None or not kombinationen:
|
||
return layer, status, []
|
||
return _baue_layer_aus_kombinationen(tab_widget, kombinationen, crs, status)
|