"""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_verfahrensgebiet.functions.verfahrensgebiet_dialoge import ask_bror_kombinationen, ask_landkreis_objekte from sn_basis.functions.landkreise import ALLE_LANDKREISE 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 # "Alle Landkreise" (Sentinel) -> kein Kennzahl-Präfix, LIKE '%' matcht alles. praefix = "" if landkreis_kennzahl == ALLE_LANDKREISE else landkreis_kennzahl landkreis_anzeige = "allen Landkreisen" if landkreis_kennzahl == ALLE_LANDKREISE else f"Landkreis {landkreis_kennzahl}" wfs_uri_find = ( f"url='{alkis_sf_url}' typename='{typename}' srsname='EPSG:25833' " f"sql=SELECT * FROM AX_BauRaumOderBodenordnungsrecht " f"WHERE bezeichnung LIKE '{praefix}%'" ) 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_anzeige} 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_verfahrensgebiet.functions.verfahrensgebiet_dialoge.ask_landkreis_objekte` (Mehrfachauswahl inkl. "Alle auswählen"). 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]: ``(layer, status)``. """ key_list = sorted(kombinationen.keys()) selected_keys = ask_landkreis_objekte(tab_widget, key_list) if not selected_keys: return None, LoadStatus.NONE # 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 pr.addFeatures(neue_features) memory_layer.updateExtents() return memory_layer, status 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)