"""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 lade_gemarkungen_layer(): """Lädt alle Katasterbezirk-Objekte aus dem vereinfachten ALKIS-WFS als benannten Layer „Gemarkungen". Dieser Layer dient als Referenzkontext im Bearbeitungs-Tab und wird mit dem Stil ``gemarkungen_beschriftet.qml`` versehen. Returns ------- QgsVectorLayer or None Geladener Layer (Name ``"Gemarkungen"``) oder ``None`` bei Ungültigkeit (z. B. Dienst nicht erreichbar). """ uri = ( f"url='{ALKIS_VEREINF_URL}' typename='{KATASTERBEZIRK_TYPENAME}' " f"srsname='EPSG:25833'" ) layer = QgsVectorLayer(uri, "Gemarkungen", "WFS") return layer if (layer and layer.isValid()) else None 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)]