"""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_SF_URL = "https://geodienste.sachsen.de/aaa/public_alkis/sf/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 msg = QMessageBox(tab_widget) msg.setWindowTitle("Layer bereits vorhanden") msg.setText( f"Der Layer '{FLST_LAYER_NAME}' existiert bereits.\nWas soll geschehen?" ) btn_ueberschreiben = msg.addButton("Überschreiben", QMessageBox.ButtonRole.DestructiveRole) btn_anhaengen = msg.addButton("Anhängen", QMessageBox.ButtonRole.AcceptRole) btn_abbrechen = msg.addButton("Abbrechen", QMessageBox.ButtonRole.RejectRole) msg.setDefaultButton(btn_abbrechen) msg.exec() clicked = msg.clickedButton() if clicked is btn_ueberschreiben: for lyr in existing_layers: project.removeMapLayer(lyr.id()) return None, LoadStatus.RELOAD elif clicked is btn_anhaengen: return existing_layers[0], LoadStatus.APPEND else: return None, LoadStatus.NONE 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_SF_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 ---------------------------- append_target = None 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 if status == LoadStatus.NONE: return None, LoadStatus.NONE if status == LoadStatus.APPEND: append_target = existing # --- 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 # --- Anhängen an vorhandenen Layer ------------------------------------ if status == LoadStatus.APPEND and append_target is not None: append_target.dataProvider().addFeatures(list(layer.getFeatures())) append_target.updateExtents() append_target.triggerRepaint() return append_target, LoadStatus.APPEND return layer, status