208 lines
7.3 KiB
Python
208 lines
7.3 KiB
Python
"""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 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)]
|