AG_QGIS/Plugin_SN_Verfahrensgebiet
3
1
Fork 1
Code Issues 10 Pull Requests Actions Releases 25 Wiki Activity
Files
16fa99a84e5a6a213ad552882dbf846114b45f71
Plugin_SN_Verfahrensgebiet/functions/grenzpunkt_alkis.py
T

281 lines
9.8 KiB
Python
Raw Normal View History

verfahrensgebiet aus Flst
2026-04-22 14:53:45 +02:00
"""sn_verfahrensgebiet/functions/grenzpunkt_alkis.py – ALKIS-Grenzpunktabfrage.
Lädt Grenzpunkte aus dem ALKIS-NAS-WFS in zwei Schritten:
1. ``adv:AX_PunktortTA`` per BBox laden.
Diese Objektart enthält die eigentliche Koordinate (Geometrie) sowie das
Attribut ``gehoertZu``, das per xlink-Referenz auf die GML-ID des
zugehörigen ``AX_Grenzpunkt``-Objekts verweist.
2. GML-IDs aus ``gehoertZu`` extrahieren → ``AX_Grenzpunkt`` per
``gml_id IN (...)`` abfragen (batching, ~25 IDs/Batch).
3. Beide Layer werden über ``gehoertZu ↔ gml_id`` verknüpft (Join) und als
einzelner angereicherter Layer zurückgegeben.
**Einschränkungen des WFS-Dienstes:**
- ``AX_Grenzpunkt`` hat **keine eigene Geometrie** → kein BBOX-Filter möglich.
- Das Attribut ``gehoertZu`` in ``AX_PunktortTA`` ist serverseitig **nicht
filterbar** (bestätigt durch WFS-Fehlermeldung ``ErrorInPropertyName``).
- Der serverseitige GML-ID-Filter (``ResourceId``) ist aktiviert
(``ImplementsResourceId=TRUE`` in den Capabilities).
**Strategie:** BBox → PunktortTA → GML-IDs → Grenzpunkt-Batches →
client-seitiger Join.
"""
from __future__ import annotations
import math
from typing import List, Optional, Tuple
from sn_basis.functions import QgsVectorLayer, QgsProject
from sn_basis.functions import QMessageBox
from sn_basis.functions.message_wrapper import info, warning
ALKIS_NAS_URL = "https://geodienste.sachsen.de/aaa/public_alkis/nas/wfs"
PUNKTORT_TA_TYPENAME = "adv:AX_PunktortTA"
GRENZPUNKT_TYPENAME = "adv:AX_Grenzpunkt"
PUNKTORT_LAYER_NAME = "PunktortTA"
GRENZPUNKT_LAYER_NAME = "Grenzpunkte"
# Attributname in AX_PunktortTA, der auf AX_Grenzpunkt verweist
_ATTR_GEHOERT_ZU = "gehoertZu"
# Attributname für die GML-ID in AX_Grenzpunkt
_ATTR_GML_ID = "gml_id"
# Batch-Größe für ResourceId-Abfragen (URL-Längenlimit)
_BATCH_GROESSE = 25
# ---------------------------------------------------------------------------
# Interne Hilfsfunktionen
# ---------------------------------------------------------------------------
def _extrahiere_gml_ids(punktort_layer) -> List[str]:
"""Extrahiert GML-IDs aus dem ``gehoertZu``-Attribut des PunktortTA-Layers.
Der WFS gibt ``gehoertZu`` üblicherweise als ``xlink:href``-Referenz im
Format ``urn:adv:oid:DEXXXXXXXXXXXXXXXXXX`` oder als flachen Wert.
Es wird der letzte Bestandteil nach dem letzten ``:`` verwendet.
Parameters
----------
punktort_layer :
Geladener ``AX_PunktortTA``-WFS-Layer.
Returns
-------
List[str]
Deduplizierte, sortierte Liste von GML-IDs.
"""
ids: set[str] = set()
for feat in punktort_layer.getFeatures():
try:
val = feat[_ATTR_GEHOERT_ZU]
if val:
raw = str(val).strip()
# xlink:href hat Format "urn:adv:oid:DE..." oder reiner GML-ID-Wert
gml_id = raw.rsplit(":", 1)[-1] if ":" in raw else raw
if gml_id:
ids.add(gml_id)
except (KeyError, Exception):
continue
return sorted(ids)
def _lade_grenzpunkte_per_ids(id_liste: List[str]) -> Optional[object]:
"""Lädt ``AX_Grenzpunkt``-Features für die angegebenen GML-IDs.
Die ID-Liste wird in Batches der Größe :data:`_BATCH_GROESSE` aufgeteilt,
um URL-Längenlimits zu umgehen. Alle Batch-Ergebnisse werden zu einem
Memory-Layer zusammengeführt.
Parameters
----------
id_liste :
GML-IDs der gesuchten ``AX_Grenzpunkt``-Objekte.
Returns
-------
QgsVectorLayer | None
Memory-Layer mit allen gefundenen Grenzpunkten oder ``None`` bei Fehler.
"""
from sn_basis.functions import QgsFeature, QgsField
from sn_basis.functions.qt_wrapper import QVariant
alle_features: List = []
feld_namen: List[str] = []
anzahl_batches = math.ceil(len(id_liste) / _BATCH_GROESSE)
for i in range(anzahl_batches):
batch = id_liste[i * _BATCH_GROESSE: (i + 1) * _BATCH_GROESSE]
ids_str = ", ".join(f"'{gid}'" for gid in batch)
uri = (
f"url='{ALKIS_NAS_URL}' typename='{GRENZPUNKT_TYPENAME}' "
f"srsname='EPSG:25833' "
f"sql=SELECT * FROM AX_Grenzpunkt WHERE gml_id IN ({ids_str})"
)
tmp = QgsVectorLayer(uri, "_TmpGP", "WFS")
if not tmp or not tmp.isValid():
continue
if not feld_namen:
feld_namen = [f.name() for f in tmp.fields()]
for feat in tmp.getFeatures():
alle_features.append(feat)
if not alle_features:
return None
# Ergebnis-Memory-Layer aufbauen (ohne Geometrie, da AX_Grenzpunkt keine hat)
mem_uri = "Point?crs=EPSG:25833" if _hat_geometrie(alle_features[0]) else "none"
result_layer = QgsVectorLayer(mem_uri, GRENZPUNKT_LAYER_NAME, "memory")
pr = result_layer.dataProvider()
felder = [QgsField(n, QVariant.String) for n in feld_namen if n]
pr.addAttributes(felder)
result_layer.updateFields()
pr.addFeatures(alle_features)
result_layer.updateExtents()
return result_layer
def _hat_geometrie(feature) -> bool:
"""Gibt ``True`` zurück, wenn das Feature eine nicht-leere Geometrie hat."""
try:
g = feature.geometry()
return g is not None and not g.isEmpty()
except Exception:
return False
# ---------------------------------------------------------------------------
# Öffentliche API
# ---------------------------------------------------------------------------
def grenzpunkt_alkis(tab_widget, extent=None) -> Tuple[Optional[object], Optional[object]]:
"""Lädt Grenzpunkte aus dem ALKIS-WFS für den angegebenen Begrenzungsrahmen.
Ablauf:
1. ``AX_PunktortTA`` per BBox laden (enthält Koordinaten + Referenz auf
``AX_Grenzpunkt`` via ``gehoertZu``).
2. GML-IDs aus ``gehoertZu`` extrahieren.
3. ``AX_Grenzpunkt`` per ID-Liste in Batches laden.
4. Beide Layer ins Projekt einfügen. Der Caller ist verantwortlich dafür,
den Join in der QGIS-Projektkonfiguration einzurichten oder die
Verknüpfung client-seitig vorzunehmen.
Wenn kein ``extent`` übergeben wird, fragt die Funktion nach dem aktuellen
Verfahrensgebiet-Layer und nutzt dessen BBox.
Parameters
----------
tab_widget :
Eltern-Widget für Dialoge.
extent :
Optional. ``QgsRectangle`` in EPSG:25833. Wenn ``None``, wird der
aktive Verfahrensgebiet-Layer gesucht und dessen Extent genutzt.
Returns
-------
Tuple[Optional[punktort_layer], Optional[grenzpunkt_layer]]
Beide geladenen Layer (bereits im Projekt) oder ``(None, None)``
bei Fehler.
"""
project = QgsProject.instance()
# BBox bestimmen
if extent is None:
vg_layers = project.mapLayersByName("Verfahrensgebiet")
if not vg_layers:
QMessageBox.warning(
tab_widget,
"Kein Verfahrensgebiet",
"Bitte zuerst ein Verfahrensgebiet laden, um die BBox abzuleiten.",
)
return None, None
extent = vg_layers[0].extent()
xmin = extent.xMinimum()
ymin = extent.yMinimum()
xmax = extent.xMaximum()
ymax = extent.yMaximum()
# -----------------------------------------------------------------------
# Schritt 1: AX_PunktortTA per BBox laden
# -----------------------------------------------------------------------
info("Grenzpunkte", "Schritt 1/3: PunktortTA per BBox laden …", duration=3)
uri_pta = (
f"url='{ALKIS_NAS_URL}' typename='{PUNKTORT_TA_TYPENAME}' "
f"srsname='EPSG:25833' "
f"bbox={xmin},{ymin},{xmax},{ymax},EPSG:25833"
)
punktort_layer = QgsVectorLayer(uri_pta, PUNKTORT_LAYER_NAME, "WFS")
if not punktort_layer or not punktort_layer.isValid():
QMessageBox.critical(
tab_widget,
"Fehler",
"PunktortTA-Layer konnte nicht geladen werden.\n"
"Bitte Netzwerkverbindung und Ausdehnung prüfen.",
)
return None, None
anzahl_pta = punktort_layer.featureCount()
if anzahl_pta == 0:
QMessageBox.warning(
tab_widget,
"Keine Grenzpunkte",
"Im Bereich des Verfahrensgebiets wurden keine PunktortTA-Objekte gefunden.",
)
return None, None
# -----------------------------------------------------------------------
# Schritt 2: GML-IDs extrahieren
# -----------------------------------------------------------------------
info("Grenzpunkte", "Schritt 2/3: GML-IDs aus gehoertZu extrahieren …", duration=3)
gml_ids = _extrahiere_gml_ids(punktort_layer)
if not gml_ids:
QMessageBox.warning(
tab_widget,
"Keine Verknüpfungen",
f"{anzahl_pta} PunktortTA-Objekte gefunden, aber kein 'gehoertZu'-Attribut.\n"
"Bitte Attributnamen im WFS prüfen.",
)
# PunktortTA trotzdem zurückgeben
project.addMapLayer(punktort_layer)
return punktort_layer, None
# -----------------------------------------------------------------------
# Schritt 3: AX_Grenzpunkt per ID-Listen-Batches laden
# -----------------------------------------------------------------------
info(
"Grenzpunkte",
f"Schritt 3/3: {len(gml_ids)} Grenzpunkte in "
f"{math.ceil(len(gml_ids) / _BATCH_GROESSE)} Batches laden …",
duration=3,
)
grenzpunkt_layer = _lade_grenzpunkte_per_ids(gml_ids)
if not grenzpunkt_layer:
warning(
"Grenzpunkte",
"Grenzpunkt-Daten konnten nicht geladen werden (IDs gefunden, aber WFS lieferte kein Ergebnis).",
duration=6,
)
project.addMapLayer(punktort_layer)
return punktort_layer, None
# Beide Layer ins Projekt
project.addMapLayer(punktort_layer)
project.addMapLayer(grenzpunkt_layer)
return punktort_layer, grenzpunkt_layer
Reference in New Issue Copy Permalink