diff --git a/functions/grenzpunkt_alkis.py b/functions/grenzpunkt_alkis.py index cad0dac..0885cd7 100644 --- a/functions/grenzpunkt_alkis.py +++ b/functions/grenzpunkt_alkis.py @@ -28,11 +28,12 @@ client-seitiger Join. from __future__ import annotations import math -from typing import List, Optional, Tuple +from typing import Callable, Dict, 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 +from sn_basis.functions.dialog_wrapper import create_progress_dialog ALKIS_SF_URL = "https://geodienste.sachsen.de/aaa/public_alkis/sf/wfs" PUNKTORT_TA_TYPENAME = "adv:AX_PunktortTA" @@ -40,24 +41,369 @@ GRENZPUNKT_TYPENAME = "adv:AX_Grenzpunkt" PUNKTORT_LAYER_NAME = "PunktortTA" GRENZPUNKT_LAYER_NAME = "Grenzpunkte" +JOINED_LAYER_NAME = "Grenzpunkte (ALKIS)" # 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 +# Batch-Größe für ResourceId-Abfragen per GET +# WAF auf geodienste.sachsen.de blockiert POST und GML-BBOX-XML in URLs. +# GET+ResourceId-Filter funktioniert, aber URL-Länge ist begrenzt (~8192 Bytes). +# Bei ~54 Bytes/ID (URL-kodiert) passen sicher 100 IDs pro Request. +_BATCH_GROESSE = 100 +# Maximale Features pro WFS-Abruf (Serverlimit) +_MAX_FEATURES = 20000 + +# Präfix für Grenzpunkt-Attribute im angereicherten Join-Layer +_GP_PREFIX = "gp_" + +# Bekannte Feldnamen für die xlink-Referenz auf AX_Grenzpunkt (ALKIS 6 + 7) +# AX_PunktortTA NAS 6.x: "istTeilVon" verweist auf den übergeordneten Grenzpunkt +_GEHOERT_ZU_KANDIDATEN = ["istTeilVon", "gehoertZu", "gehoertZuAX_Grenzpunkt"] + +# Vereinfachter ALKIS-WFS für Gemarkungs-BBox-Abfragen +_ALKIS_VEREINF_URL = "https://geodienste.sachsen.de/aaa/public_alkis/vereinf/wfs" +_KATASTERBEZIRK_TYPENAME = "ave:KatasterBezirk" +_KB_ATTR_SCHLUESSEL = "schluessel" # --------------------------------------------------------------------------- # Interne Hilfsfunktionen # --------------------------------------------------------------------------- -def _extrahiere_gml_ids(punktort_layer) -> List[str]: - """Extrahiert GML-IDs aus dem ``gehoertZu``-Attribut des PunktortTA-Layers. +def _erkenne_gehoert_zu_feld(layer) -> Optional[str]: + """Erkennt den Feldnamen im ``AX_PunktortTA``-Layer, der auf + ``AX_Grenzpunkt`` verweist – ALKIS-6/7-kompatibel. - Der WFS gibt ``gehoertZu`` üblicherweise als ``xlink:href``-Referenz im + Prüft Kandidaten in fester Reihenfolge: + + 1. Bekannte Feldnamen aus :data:`_GEHOERT_ZU_KANDIDATEN`. + 2. Fallback: erstes Feld, dessen Name ``gehoert`` (case-insensitive) + enthält. + + Parameters + ---------- + layer : + Geladener ``AX_PunktortTA``-Layer. + + Returns + ------- + str | None + Gefundener Feldname oder ``None`` wenn kein Kandidat vorhanden. + """ + feld_namen = [f.name() for f in layer.fields()] + for kandidat in _GEHOERT_ZU_KANDIDATEN: + if kandidat in feld_namen: + return kandidat + # Fallback: erstes Feld, das 'gehoert' oder 'istteil' oder 'besteht' enthält + for name in feld_namen: + low = name.lower() + if "gehoert" in low or "istteil" in low or "bestehtaus" in low: + return name + warning( + "Grenzpunkte (Diagnose)", + "Kein Referenzfeld in AX_PunktortTA gefunden. " + "Vorhandene Felder: " + ", ".join(feld_namen), + duration=15, + ) + return None + + +def _punktort_ta_bbox_param(bbox) -> str: + """Gibt den ``BBOX``-KVP-Wert für eine PunktortTA-Abfrage zurück. + + **Warum BBOX-Parameter statt FILTER-XML?** + Die WAF auf geodienste.sachsen.de blockiert URLs, die GML-Geometrie-XML + im ``FILTER``-Parameter enthalten (``xmlns:gml``, ``gml:Envelope`` usw.). + Der native WFS-2.0-KVP-Parameter ``BBOX=minx,miny,maxx,maxy,crs`` wird + dagegen akzeptiert. + + Parameters + ---------- + bbox : + ``QgsRectangle`` in EPSG:25833. + + Returns + ------- + str + Wert für den ``BBOX``-URL-Parameter. + """ + return ( + f"{bbox.xMinimum()},{bbox.yMinimum()}" + f",{bbox.xMaximum()},{bbox.yMaximum()}" + ",urn:ogc:def:crs:EPSG::25833" + ) + + +def _wfs_get_gml(url: str, typename: str, extra_params: dict) -> Optional[bytes]: + """Sendet einen WFS 2.0 GetFeature GET-Request und gibt die rohe GML-Antwort zurück. + + **Warum GET?** + Die WAF auf geodienste.sachsen.de blockiert POST-Requests und GET-Requests + mit GML-Geometrie-XML im ``FILTER``-Parameter (``xmlns:gml``, ``gml:Envelope``). + GET-Requests mit ``BBOX=``-KVP-Parameter oder ResourceId-``FILTER`` (nur + ``xmlns:fes``, kein GML) werden dagegen akzeptiert. + + Parameters + ---------- + url : + WFS-Endpoint-URL. + typename : + Vollqualifizierter Feature-Typ, z.B. ``adv:AX_PunktortTA``. + extra_params : + Zusätzliche KVP-Parameter, z.B. ``{'BBOX': 'minx,miny,maxx,maxy,crs'}`` + oder ``{'FILTER': ''}``. + + Returns + ------- + bytes | None + Rohe GML-Antwort oder ``None`` bei Fehler. + """ + import urllib.request + import urllib.parse + import urllib.error + from qgis.core import QgsMessageLog, Qgis + + params = { + 'SERVICE': 'WFS', + 'VERSION': '2.0.0', + 'REQUEST': 'GetFeature', + 'TYPENAMES': typename, + 'COUNT': str(_MAX_FEATURES), + } + params.update(extra_params) + + full_url = url + '?' + urllib.parse.urlencode(params) + + try: + with urllib.request.urlopen(full_url, timeout=120) as resp: + gml_bytes = resp.read() + except urllib.error.HTTPError as exc: + body_preview = exc.read(200).decode('utf-8', errors='replace') + QgsMessageLog.logMessage( + f"WFS GET HTTP {exc.code} ({typename}): {exc.reason}\n" + f"URL: {full_url[:300]}\nAntwort: {body_preview}", + "sn_verfahrensgebiet", + Qgis.MessageLevel.Critical, + ) + warning("WFS GET", f"HTTP {exc.code}: {exc.reason}", duration=10) + return None + except Exception as exc: + QgsMessageLog.logMessage( + f"WFS GET fehlgeschlagen ({typename}): {exc}\nURL: {full_url[:300]}", + "sn_verfahrensgebiet", + Qgis.MessageLevel.Critical, + ) + warning("WFS GET", f"Netzwerkfehler: {exc}", duration=10) + return None + + if not gml_bytes: + return None + + # WFS-Fehlermeldung erkennen + if b"ExceptionReport" in gml_bytes[:800] or b"ExceptionText" in gml_bytes[:800]: + import xml.etree.ElementTree as _ET + try: + _root = _ET.fromstring(gml_bytes) + _exc = _root.find(".//{http://www.opengis.net/ows/1.1}ExceptionText") + msg = _exc.text.strip() if _exc is not None and _exc.text else "Unbekannter WFS-Fehler" + except Exception: + msg = gml_bytes[:500].decode("utf-8", errors="replace") + QgsMessageLog.logMessage( + f"WFS ExceptionReport ({typename}):\n{msg}", + "sn_verfahrensgebiet", + Qgis.MessageLevel.Critical, + ) + warning("WFS GET", f"WFS-Fehler: {msg}", duration=10) + return None + + return gml_bytes + + +def _parse_alkis_gml_memory_layer( + gml_bytes: bytes, + feature_local_name: str, + layer_name: str, + geom_prop: Optional[str] = "position", + extract_qualitaet: bool = False, +) -> Optional[object]: + """Parst ALKIS-GML-Bytes und erstellt einen QGIS Memory-Layer. + + Verarbeitet die GML-Antwort eines WFS-POST-Requests und extrahiert + Features des angegebenen Typs als Memory-Layer. + + Parameters + ---------- + gml_bytes : + Rohe GML-Bytes der WFS-Antwort. + feature_local_name : + Lokaler Name des Feature-Typs, z.B. ``AX_PunktortTA``. + layer_name : + Anzeigename des resultierenden Memory-Layers. + geom_prop : + Lokaler Name des Geometrie-Attributs (``None`` = kein Geometrie-Layer). + extract_qualitaet : + Wenn ``True``, wird das verschachtelte Attribut + ``qualitaetsangaben_ax_dqpunktort_genauigkeitsstufe`` aus + ``adv:qualitaetsangaben/adv:genauigkeitsstufe`` extrahiert. + Nur für ``AX_PunktortTA`` sinnvoll. + + Returns + ------- + QgsVectorLayer | None + Memory-Layer oder ``None`` wenn keine Features gefunden. + """ + import xml.etree.ElementTree as ET + from sn_basis.functions import QgsFeature, QgsField + from sn_basis.functions.qt_wrapper import QVariant + from qgis.core import QgsGeometry, QgsPointXY + + _ADV = "http://www.adv-online.de/namespaces/adv/gid/6.0" + _GML = "http://www.opengis.net/gml/3.2" + _XLINK = "http://www.w3.org/1999/xlink" + + try: + root = ET.fromstring(gml_bytes) + except ET.ParseError as exc: + warning("GML-Parser", f"GML konnte nicht geparst werden: {exc}", duration=8) + return None + + feat_tag = f"{{{_ADV}}}{feature_local_name}" + feat_elems = root.findall(f".//{feat_tag}") + if not feat_elems: + return None + + # Felder aus ALLEN Features ermitteln (2-Pass), damit optionale Felder + # wie "istTeilVon" nicht verloren gehen, wenn sie im ersten Feature fehlen. + _QUALITAET_CONTAINER = "qualitaetsangaben" + _QUALITAET_FELD = "qualitaetsangaben_ax_dqpunktort_genauigkeitsstufe" + feld_namen: List[str] = ["gml_id"] + for feat_el in feat_elems: + for child in feat_el: + local = child.tag.split("}")[-1] if "}" in child.tag else child.tag + if extract_qualitaet and local == _QUALITAET_CONTAINER: + continue # Container wird durch verschachteltes Einzelfeld ersetzt + if local not in feld_namen and local != geom_prop: + feld_namen.append(local) + if extract_qualitaet and _QUALITAET_FELD not in feld_namen: + feld_namen.append(_QUALITAET_FELD) + + # Memory-Layer anlegen + mem_uri = f"Point?crs=EPSG:25833" if geom_prop else "none" + mem_layer = QgsVectorLayer(mem_uri, layer_name, "memory") + pr = mem_layer.dataProvider() + pr.addAttributes([QgsField(n, QVariant.String) for n in feld_namen if n]) + mem_layer.updateFields() + + neue_features: List = [] + for feat_elem in feat_elems: + gml_id = feat_elem.get(f"{{{_GML}}}id", "") + + # Geometrie extrahieren + geom = None + if geom_prop: + pos_el = feat_elem.find(f"{{{_ADV}}}{geom_prop}") + if pos_el is not None: + pe = pos_el.find(f".//{{{_GML}}}pos") + if pe is not None and pe.text: + coords = pe.text.strip().split() + if len(coords) >= 2: + try: + geom = QgsGeometry.fromPointXY( + QgsPointXY(float(coords[0]), float(coords[1])) + ) + except (ValueError, Exception): + pass + + qgs_feat = QgsFeature(mem_layer.fields()) + if geom: + qgs_feat.setGeometry(geom) + qgs_feat["gml_id"] = gml_id + + for child in feat_elem: + local = child.tag.split("}")[-1] if "}" in child.tag else child.tag + if extract_qualitaet and local == _QUALITAET_CONTAINER: + # Verschachtelten Genauigkeitswert extrahieren + qual_el = child.find(f".//{{{_ADV}}}genauigkeitsstufe") + if qual_el is not None and qual_el.text: + try: + qgs_feat[_QUALITAET_FELD] = qual_el.text.strip() + except Exception: + pass + continue + if local == geom_prop or local not in feld_namen: + continue + href = child.get(f"{{{_XLINK}}}href", "") + val = href if href else (child.text.strip() if child.text else "") + try: + qgs_feat[local] = val + except Exception: + pass + + neue_features.append(qgs_feat) + + pr.addFeatures(neue_features) + mem_layer.updateExtents() + return mem_layer + + +def _extrahiere_feature_gml_id(feat, feld_namen: List[str]) -> str: + """Extrahiert die GML-ID aus einem WFS-Feature robust. + + Versucht mehrere Strategien in dieser Reihenfolge: + + 1. Bekannte Feldnamen (``gml_id``, ``identifier``). + 2. Scan aller String-Felder auf GML-ID-Muster (``DE`` + 14-20 alphanumerische Zeichen). + + Parameters + ---------- + feat : + WFS-Feature. + feld_namen : + Liste der Feldnamen des Layers. + + Returns + ------- + str + GML-ID oder leerer String. + """ + import re as _re + _GML_ID_PATTERN = _re.compile(r'^DE[0-9A-Za-z_]{10,24}$') + + for candidate in ("gml_id", "identifier", "gmlId", "featureId", "fid"): + if candidate in feld_namen: + try: + val = feat[candidate] + if val: + s = str(val).strip() + if s: + return s + except Exception: + pass + + # Scan aller Felder auf GML-ID-Muster + for name in feld_namen: + try: + val = feat[name] + if val: + s = str(val).strip() + if _GML_ID_PATTERN.match(s): + return s + except Exception: + pass + + return "" + + +def _extrahiere_gml_ids( + punktort_layer, + gehoert_zu_feld: Optional[str] = None, +) -> List[str]: + """Extrahiert GML-IDs aus dem Referenzfeld des PunktortTA-Layers. + + Der WFS gibt das Referenzfeld üblicherweise als ``xlink:href``-Referenz im Format ``urn:adv:oid:DEXXXXXXXXXXXXXXXXXX`` oder als flachen Wert. Es wird der letzte Bestandteil nach dem letzten ``:`` verwendet. @@ -65,16 +411,21 @@ def _extrahiere_gml_ids(punktort_layer) -> List[str]: ---------- punktort_layer : Geladener ``AX_PunktortTA``-WFS-Layer. + gehoert_zu_feld : + Feldname der Referenz auf ``AX_Grenzpunkt``. Wenn ``None``, wird + :data:`_ATTR_GEHOERT_ZU` als Fallback verwendet + (Rückwärtskompatibilität). Returns ------- List[str] Deduplizierte, sortierte Liste von GML-IDs. """ + feld = gehoert_zu_feld if gehoert_zu_feld is not None else _ATTR_GEHOERT_ZU ids: set[str] = set() for feat in punktort_layer.getFeatures(): try: - val = feat[_ATTR_GEHOERT_ZU] + val = feat[feld] if val: raw = str(val).strip() # xlink:href hat Format "urn:adv:oid:DE..." oder reiner GML-ID-Wert @@ -86,6 +437,89 @@ def _extrahiere_gml_ids(punktort_layer) -> List[str]: return sorted(ids) +def _lade_grenzpunkte_als_dict( + id_liste: List[str], + progress_callback: Optional[Callable[[int], None]] = None, +) -> Tuple[Dict[str, object], List[str]]: + """Lädt ``AX_Grenzpunkt``-Features als Dict für O(1)-Lookup. + + Parameters + ---------- + id_liste : + GML-IDs der gesuchten ``AX_Grenzpunkt``-Objekte. + progress_callback : + Optionaler Callback ``fn(batch_index)`` der nach jedem Batch aufgerufen + wird (0-basierter Batch-Index). Dient zur Fortschrittsanzeige. + + Returns + ------- + Tuple[Dict[str, QgsFeature], List[str]] + ``(gp_dict, feld_namen)`` – Dict ``{gml_id: QgsFeature}`` und + Liste der GP-Feldnamen (leer wenn keine Features geladen wurden). + """ + gp_dict: Dict[str, object] = {} + 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] + info( + "Grenzpunkte (Diagnose)", + f"AX_Grenzpunkt Batch {i + 1}/{anzahl_batches}: {len(batch)} IDs per ResourceId-Filter …", + duration=2, + ) + + rid_elemente = "".join(f'' for gid in batch) + filter_xml = ( + '' + + rid_elemente + + "" + ) + _gml_bytes = _wfs_get_gml(ALKIS_SF_URL, GRENZPUNKT_TYPENAME, {'FILTER': filter_xml}) + if not _gml_bytes: + warning( + "Grenzpunkte (Diagnose)", + f"Batch {i + 1}/{anzahl_batches}: GET-Anfrage fehlgeschlagen – Batch wird übersprungen.", + duration=4, + ) + else: + tmp = _parse_alkis_gml_memory_layer(_gml_bytes, "AX_Grenzpunkt", "_TmpGP", geom_prop=None) + if not tmp: + warning( + "Grenzpunkte (Diagnose)", + f"Batch {i + 1}/{anzahl_batches}: Keine Features in GML-Antwort – Batch wird übersprungen.", + duration=4, + ) + else: + batch_feld_namen = [f.name() for f in tmp.fields()] + if not feld_namen and batch_feld_namen: + feld_namen = batch_feld_namen + info( + "Grenzpunkte (Diagnose)", + f"AX_Grenzpunkt-Felder: {', '.join(feld_namen)}", + duration=8, + ) + + batch_count = 0 + for feat in tmp.getFeatures(): + gml_id = _extrahiere_feature_gml_id(feat, batch_feld_namen or feld_namen) + if gml_id: + gp_dict[gml_id] = feat + batch_count += 1 + + info( + "Grenzpunkte (Diagnose)", + f"Batch {i + 1}/{anzahl_batches}: {batch_count}/{len(batch)} Grenzpunkte empfangen " + f"(gesamt bisher: {len(gp_dict)}).", + duration=2, + ) + + if progress_callback is not None: + progress_callback(i) + + return gp_dict, feld_namen + + def _lade_grenzpunkte_per_ids(id_liste: List[str]) -> Optional[object]: """Lädt ``AX_Grenzpunkt``-Features für die angegebenen GML-IDs. @@ -93,6 +527,8 @@ def _lade_grenzpunkte_per_ids(id_liste: List[str]) -> Optional[object]: um URL-Längenlimits zu umgehen. Alle Batch-Ergebnisse werden zu einem Memory-Layer zusammengeführt. + Delegiert intern an :func:`_lade_grenzpunkte_als_dict`. + Parameters ---------- id_liste : @@ -103,34 +539,15 @@ def _lade_grenzpunkte_per_ids(id_liste: List[str]) -> Optional[object]: QgsVectorLayer | None Memory-Layer mit allen gefundenen Grenzpunkten oder ``None`` bei Fehler. """ - from sn_basis.functions import QgsFeature, QgsField + from sn_basis.functions import 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_SF_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: + gp_dict, feld_namen = _lade_grenzpunkte_als_dict(id_liste) + if not gp_dict: return None + alle_features = list(gp_dict.values()) + # 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") @@ -154,9 +571,521 @@ def _hat_geometrie(feature) -> bool: return False -# --------------------------------------------------------------------------- -# Öffentliche API -# --------------------------------------------------------------------------- +def _lade_gemarkung_extents(schluessel_liste: List[str]) -> Dict[str, object]: + """Lädt BBox-Rechtecke der Gemarkungen aus dem vereinfachten ALKIS-WFS. + + Da der vereinfachte WFS keine serverseitigen Filter unterstützt, werden + alle Katasterbezirke geladen und client-seitig nach den angegebenen + Gemarkungsnummern gefiltert. Der vollständige Katasterbezirksschlüssel hat + ≥ 4 Stellen; die letzten 4 Stellen entsprechen der Gemarkungsnummer. + + Parameters + ---------- + schluessel_liste : + Liste der 4-stelligen Gemarkungsnummern. + + Returns + ------- + Dict[str, QgsRectangle] + ``{gemarkungsnummer: QgsRectangle}`` – BBox je Gemarkung. + Leer bei Fehler oder keinen Treffern. + """ + from qgis.core import QgsRectangle + + uri = ( + f"url='{_ALKIS_VEREINF_URL}' typename='{_KATASTERBEZIRK_TYPENAME}' " + f"srsname='EPSG:25833'" + ) + layer = QgsVectorLayer(uri, "_TmpKB", "WFS") + if not layer or not layer.isValid(): + return {} + + schluessel_set = set(schluessel_liste) + extents: Dict[str, object] = {} + + for feat in layer.getFeatures(): + try: + val = feat[_KB_ATTR_SCHLUESSEL] + schluessel = str(val).strip() if val else "" + except Exception: + continue + + nummer = schluessel[-4:] if len(schluessel) >= 4 else schluessel + if nummer not in schluessel_set: + continue + + if not feat.hasGeometry() or feat.geometry().isEmpty(): + continue + + bbox = feat.geometry().boundingBox() + if nummer in extents: + extents[nummer].combineExtentWith(bbox) + else: + extents[nummer] = QgsRectangle(bbox) + + return extents + + +def _lade_gemarkung_extents_mit_puffer( + schluessel_liste: List[str], + vg_geom, + puffer_m: float = 200.0, +) -> Dict[str, object]: + """Lädt Gemarkungsgeometrien, verschneidet sie mit einem Puffer um das + Verfahrensgebiet und liefert BBoxen der Schnittflächen. + + Statt der vollen Gemarkungs-BBox wird nur die BBox des Anteils verwendet, + der innerhalb des gepufferten Verfahrensgebiets liegt. Das reduziert die + Anzahl der vom WFS gelieferten PunktortTA-Objekte erheblich. + + Parameters + ---------- + schluessel_liste : + Liste der 4-stelligen Gemarkungsnummern. + vg_geom : + ``QgsGeometry`` des Verfahrensgebiets (EPSG:25833). + puffer_m : + Pufferdistanz in Metern (Standard: 200 m). + + Returns + ------- + Dict[str, QgsRectangle] + ``{gemarkungsnummer: QgsRectangle}`` – BBox der Schnittfläche. + Gemarkungen ohne Schnitt werden ausgelassen. + """ + from qgis.core import QgsRectangle + + uri = ( + f"url='{_ALKIS_VEREINF_URL}' typename='{_KATASTERBEZIRK_TYPENAME}' " + f"srsname='EPSG:25833'" + ) + layer = QgsVectorLayer(uri, "_TmpKB", "WFS") + if not layer or not layer.isValid(): + return {} + + puffer_geom = vg_geom.buffer(puffer_m, 16) if vg_geom and not vg_geom.isEmpty() else None + + schluessel_set = set(schluessel_liste) + extents: Dict[str, object] = {} + + for feat in layer.getFeatures(): + try: + val = feat[_KB_ATTR_SCHLUESSEL] + schluessel = str(val).strip() if val else "" + except Exception: + continue + + nummer = schluessel[-4:] if len(schluessel) >= 4 else schluessel + if nummer not in schluessel_set: + continue + + if not feat.hasGeometry() or feat.geometry().isEmpty(): + continue + + geom = feat.geometry() + + # Mit gepuffertem VG verschneiden + if puffer_geom and not puffer_geom.isEmpty(): + schnitt = geom.intersection(puffer_geom) + if schnitt is None or schnitt.isEmpty(): + continue + bbox = schnitt.boundingBox() + else: + bbox = geom.boundingBox() + + if bbox.isEmpty(): + continue + + if nummer in extents: + extents[nummer].combineExtentWith(bbox) + else: + extents[nummer] = QgsRectangle(bbox) + + return extents + + +def _wfs_hits_count(url: str, typename: str, extra_params: dict) -> int: + """Fragt die Anzahl der Features per ``resultType=hits`` ab. + + Parameters + ---------- + url : + WFS-Endpoint-URL. + typename : + Vollqualifizierter Feature-Typ. + extra_params : + Zus\u00e4tzliche KVP-Parameter (z.B. ``{'BBOX': '...'}``). + + Returns + ------- + int + Anzahl der Features oder ``-1`` wenn nicht ermittelbar. + """ + import urllib.request + import urllib.parse + import urllib.error + import xml.etree.ElementTree as ET + + params = { + 'SERVICE': 'WFS', + 'VERSION': '2.0.0', + 'REQUEST': 'GetFeature', + 'TYPENAMES': typename, + 'RESULTTYPE': 'hits', + } + params.update(extra_params) + + full_url = url + '?' + urllib.parse.urlencode(params) + try: + with urllib.request.urlopen(full_url, timeout=30) as resp: + gml_bytes = resp.read() + except Exception: + return -1 + + try: + root = ET.fromstring(gml_bytes) + matched = root.get('numberMatched', '') + if matched and matched != 'unknown': + return int(matched) + returned = root.get('numberReturned', '') + if returned and returned != 'unknown': + return int(returned) + except Exception: + pass + return -1 + + +def _lade_punktort_ta_batches_per_gemarkung( + schluessel_liste: List[str], +) -> Optional[object]: + """Lädt ``AX_PunktortTA`` für jede Gemarkung separat per BBox. + + Pro Gemarkung wird eine eigene WFS-Abfrage durchgeführt, was präzisere + Ergebnisse als eine kombinierte BBox liefert. Doppelte Features an + Gemarkungsgrenzen werden anhand der ``gml_id`` dedupliziert. + + Parameters + ---------- + schluessel_liste : + Liste der 4-stelligen Gemarkungsnummern. + + Returns + ------- + QgsVectorLayer | None + Memory-Layer mit allen PunktortTA-Features oder ``None`` bei Fehler. + """ + from sn_basis.functions import QgsField + from sn_basis.functions.qt_wrapper import QVariant + + extents = _lade_gemarkung_extents(schluessel_liste) + if not extents: + return None + + alle_features: List = [] + gesehene_ids: set[str] = set() + feld_namen: List[str] = [] + + n_gemarkungen_gesamt = len(extents) + for g_idx, (_gemarkungsnummer, bbox) in enumerate(extents.items(), 1): + info( + "Grenzpunkte (Diagnose)", + f"PunktortTA Gemarkung {_gemarkungsnummer} ({g_idx}/{n_gemarkungen_gesamt}): " + f"BBox {bbox.xMinimum():.0f},{bbox.yMinimum():.0f} – " + f"{bbox.xMaximum():.0f},{bbox.yMaximum():.0f} anfragen …", + duration=2, + ) + _gml_bytes = _wfs_get_gml(ALKIS_SF_URL, PUNKTORT_TA_TYPENAME, {'BBOX': _punktort_ta_bbox_param(bbox)}) + if not _gml_bytes: + warning( + "Grenzpunkte (Diagnose)", + f"Gemarkung {_gemarkungsnummer}: GET-Anfrage fehlgeschlagen – wird übersprungen.", + duration=4, + ) + continue + tmp = _parse_alkis_gml_memory_layer(_gml_bytes, "AX_PunktortTA", "_TmpPTA", geom_prop="position", extract_qualitaet=True) + if not tmp: + warning( + "Grenzpunkte (Diagnose)", + f"Gemarkung {_gemarkungsnummer}: Keine Features in GML-Antwort – wird übersprungen.", + duration=4, + ) + continue + + if not feld_namen: + feld_namen = [f.name() for f in tmp.fields()] + + cur_feld_namen = [f.name() for f in tmp.fields()] + + neu = 0 + duplikat = 0 + for feat in tmp.getFeatures(): + gml_id = _extrahiere_feature_gml_id(feat, cur_feld_namen) + if gml_id and gml_id in gesehene_ids: + duplikat += 1 + continue + if gml_id: + gesehene_ids.add(gml_id) + alle_features.append(feat) + neu += 1 + + info( + "Grenzpunkte (Diagnose)", + f"Gemarkung {_gemarkungsnummer}: {neu} neue PunktortTA-Objekte " + f"({duplikat} Duplikate übersprungen, gesamt bisher: {len(alle_features)}).", + duration=2, + ) + + if not alle_features: + return None + + result_layer = QgsVectorLayer("Point?crs=EPSG:25833", PUNKTORT_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 _erstelle_joined_layer( + punktort_layer, + gp_dict: Dict[str, object], + gp_feld_namen: List[str], + gehoert_zu_feld: str, +) -> object: + """Erstellt einen angereicherten Memory-Layer aus PunktortTA + GP-Attributen. + + Geometrie und Attribute von ``AX_PunktortTA`` werden mit den Attributen + des zugehörigen ``AX_Grenzpunkt`` zusammengeführt. GP-Felder erhalten den + Präfix :data:`_GP_PREFIX` (``gp_``), um Namenskollisionen zu vermeiden. + Features ohne GP-Match werden mit leeren ``gp_``-Feldern aufgenommen. + + Parameters + ---------- + punktort_layer : + Memory-Layer mit ``AX_PunktortTA``-Features. + gp_dict : + ``{gml_id: QgsFeature}`` der geladenen ``AX_Grenzpunkt``-Features. + gp_feld_namen : + Feldnamen aus dem ``AX_Grenzpunkt``-WFS. + gehoert_zu_feld : + Feldname in ``AX_PunktortTA``, der die xlink-Referenz enthält. + + Returns + ------- + QgsVectorLayer + Angereicherter Memory-Layer ``Point?crs=EPSG:25833``. + """ + from sn_basis.functions import QgsFeature, QgsField + from sn_basis.functions.qt_wrapper import QVariant + + pta_feld_namen = [f.name() for f in punktort_layer.fields()] + + result_layer = QgsVectorLayer("Point?crs=EPSG:25833", JOINED_LAYER_NAME, "memory") + pr = result_layer.dataProvider() + + felder: List = [] + for n in pta_feld_namen: + if n: + felder.append(QgsField(n, QVariant.String)) + for n in gp_feld_namen: + if n: + felder.append(QgsField(f"{_GP_PREFIX}{n}", QVariant.String)) + + pr.addAttributes(felder) + result_layer.updateFields() + + neue_features: List = [] + for pta_feat in punktort_layer.getFeatures(): + try: + raw = pta_feat[gehoert_zu_feld] + raw = str(raw).strip() if raw else "" + gml_id = raw.rsplit(":", 1)[-1] if ":" in raw else raw + except Exception: + gml_id = "" + + gp_feat = gp_dict.get(gml_id) + + new_feat = QgsFeature(result_layer.fields()) + new_feat.setGeometry(pta_feat.geometry()) + + # PunktortTA-Attribute + for n in pta_feld_namen: + if n: + try: + new_feat[n] = pta_feat[n] + except Exception: + pass + + # Grenzpunkt-Attribute mit gp_-Präfix + if gp_feat is not None: + for n in gp_feld_namen: + if n: + try: + new_feat[f"{_GP_PREFIX}{n}"] = gp_feat[n] + except Exception: + pass + + neue_features.append(new_feat) + + pr.addFeatures(neue_features) + result_layer.updateExtents() + return result_layer + + +def lade_punktort_ta( + schluessel_liste: List[str], + vg_geom, + progress=None, +) -> Optional[object]: + """Lädt ``AX_PunktortTA`` gemarkungsweise mit STARTINDEX-Paginierung. + + Verwendet dieselbe Strategie wie :func:`grenzpunkte_gemarkungsweise`: + + 1. Gemarkungsextents laden und mit einem 200-m-VG-Puffer verschneiden. + 2. Hits-Abfrage pro Gemarkung zur Batch-Planung. + 3. PunktortTA je Gemarkung in 5000-Feature-Batches laden (STARTINDEX-Paging). + 4. gml_id-Deduplizierung über Gemarkungsgrenzen hinweg. + + Es werden keine Grenzpunkt-Batches geladen – die Funktion gibt ausschließlich + den PunktortTA-Layer zurück. + + Parameters + ---------- + schluessel_liste : + 4-stellige Gemarkungsnummern (aus dem aktuellen Workflow bereits bekannt). + vg_geom : + ``QgsGeometry`` des BROR-Objekts (EPSG:25833). Wird für den 200-m-Puffer + zur Einschränkung der Gemarkungsextents verwendet. + progress : + Optionales Fortschrittsdialog-Objekt mit ``set_label()``, ``set_value()`` + und ``is_canceled()``. + + Returns + ------- + QgsVectorLayer | None + Memory-Layer mit allen PunktortTA-Features oder ``None`` bei Fehler. + """ + from sn_basis.functions import QgsField + from sn_basis.functions import QgsFeature as _QgsFeature + from sn_basis.functions.qt_wrapper import QVariant + + _BBOX_BATCH = 5000 + + if not schluessel_liste: + return None + + if vg_geom and not vg_geom.isEmpty(): + extents = _lade_gemarkung_extents_mit_puffer(schluessel_liste, vg_geom, 200.0) + else: + extents = _lade_gemarkung_extents(schluessel_liste) + + if not extents: + return None + + n_extents = len(extents) + alle_records: List[Dict] = [] + gesehene_ids: set = set() + feld_namen: List[str] = [] + + for g_idx, (gemarkungsnummer, bbox) in enumerate(extents.items(), 1): + if progress is not None and progress.is_canceled(): + return None + + bbox_param = _punktort_ta_bbox_param(bbox) + n_hits = _wfs_hits_count(ALKIS_SF_URL, PUNKTORT_TA_TYPENAME, {'BBOX': bbox_param}) + n_batches = max(1, math.ceil(n_hits / _BBOX_BATCH)) if n_hits > 0 else 1 + + info( + "Grenzpunkte (Diagnose)", + f"Gemarkung {gemarkungsnummer} ({g_idx}/{n_extents}): " + f"~{n_hits} PunktortTA → {n_batches} Batch(es).", + duration=2, + ) + + for b_idx in range(n_batches): + if progress is not None: + progress.set_label( + f"PunktortTA: Gemarkung {gemarkungsnummer} " + f"({g_idx}/{n_extents}), Batch {b_idx + 1}/{n_batches} …" + ) + if progress.is_canceled(): + return None + + extra: dict = {'BBOX': bbox_param} + if n_batches > 1: + extra['COUNT'] = str(_BBOX_BATCH) + extra['STARTINDEX'] = str(b_idx * _BBOX_BATCH) + + _gml_bytes = _wfs_get_gml(ALKIS_SF_URL, PUNKTORT_TA_TYPENAME, extra) + if not _gml_bytes: + warning( + "Grenzpunkte (Diagnose)", + f"Gemarkung {gemarkungsnummer} Batch {b_idx + 1}: " + f"GET-Anfrage fehlgeschlagen – übersprungen.", + duration=4, + ) + continue + + tmp = _parse_alkis_gml_memory_layer( + _gml_bytes, "AX_PunktortTA", "_TmpPTA", geom_prop="position", extract_qualitaet=True + ) + if not tmp: + continue + + cur_felder = [f.name() for f in tmp.fields()] + for n in cur_felder: + if n not in feld_namen: + feld_namen.append(n) + + for feat in tmp.getFeatures(): + gml_id = _extrahiere_feature_gml_id(feat, cur_felder) + if gml_id and gml_id in gesehene_ids: + continue + if gml_id: + gesehene_ids.add(gml_id) + attrs: Dict = {} + for n in cur_felder: + if n: + try: + attrs[n] = feat[n] + except Exception: + pass + alle_records.append({'geom': feat.geometry(), 'attrs': attrs}) + + info( + "Grenzpunkte (Diagnose)", + f"Gemarkung {gemarkungsnummer}: {len(gesehene_ids)} PunktortTA bisher.", + duration=2, + ) + + if not alle_records: + return None + + result = QgsVectorLayer("Point?crs=EPSG:25833", PUNKTORT_LAYER_NAME, "memory") + pr = result.dataProvider() + pr.addAttributes([QgsField(n, QVariant.String) for n in feld_namen if n]) + result.updateFields() + + neue_features = [] + for record in alle_records: + f = _QgsFeature(result.fields()) + f.setGeometry(record['geom']) + for n, val in record['attrs'].items(): + if n in feld_namen: + try: + f[n] = val + except Exception: + pass + neue_features.append(f) + + pr.addFeatures(neue_features) + result.updateExtents() + return result + def grenzpunkt_alkis(tab_widget, extent=None) -> Tuple[Optional[object], Optional[object]]: """Lädt Grenzpunkte aus dem ALKIS-WFS für den angegebenen Begrenzungsrahmen. @@ -237,17 +1166,31 @@ def grenzpunkt_alkis(tab_widget, extent=None) -> Tuple[Optional[object], Optiona return None, None # ----------------------------------------------------------------------- - # Schritt 2: GML-IDs extrahieren + # Schritt 2: Referenzfeld erkennen + GML-IDs extrahieren (ALKIS-6/7) # ----------------------------------------------------------------------- - info("Grenzpunkte", "Schritt 2/3: GML-IDs aus gehoertZu extrahieren …", duration=3) - gml_ids = _extrahiere_gml_ids(punktort_layer) + info("Grenzpunkte", "Schritt 2/3: Referenzfeld erkennen und GML-IDs extrahieren …", duration=3) + gehoert_zu_feld = _erkenne_gehoert_zu_feld(punktort_layer) + if not gehoert_zu_feld: + vorhandene_felder = ", ".join(f.name() for f in punktort_layer.fields()) + QMessageBox.warning( + tab_widget, + "Kein Referenzfeld", + f"{anzahl_pta} PunktortTA-Objekte geladen, aber kein Referenzfeld auf " + f"AX_Grenzpunkt gefunden.\n\n" + f"Gesuchte Feldnamen: {', '.join(_GEHOERT_ZU_KANDIDATEN)}\n" + f"Vorhandene Felder: {vorhandene_felder}", + ) + project.addMapLayer(punktort_layer) + return punktort_layer, None + + gml_ids = _extrahiere_gml_ids(punktort_layer, gehoert_zu_feld) 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.", + f"{anzahl_pta} PunktortTA-Objekte gefunden, aber Feld '{gehoert_zu_feld}' " + "enthält keine gültigen IDs.\nBitte Attributwerte im WFS prüfen.", ) # PunktortTA trotzdem zurückgeben project.addMapLayer(punktort_layer) @@ -278,3 +1221,441 @@ def grenzpunkt_alkis(tab_widget, extent=None) -> Tuple[Optional[object], Optiona project.addMapLayer(grenzpunkt_layer) return punktort_layer, grenzpunkt_layer + + +# --------------------------------------------------------------------------- +# Gemarkungsweise Abfrage +# --------------------------------------------------------------------------- + +def grenzpunkte_gemarkungsweise( + tab_widget, + schluessel_liste: List[str], + vg_geom=None, + progress: Optional[Callable[[int, int], None]] = None, +) -> Optional[object]: + """Lädt Grenzpunkte aus dem ALKIS-WFS gemarkungsweise mit per-Gemarkung-Tracking. + + Ablauf: + + 1. Gemarkungsgeometrien laden, mit 200-m-Puffer um das Verfahrensgebiet + verschneiden, BBoxen der Schnittflächen bestimmen. + 2. Pro Gemarkung: Hits-Abfrage für ``AX_PunktortTA``, dann ggf. in + Batches à 5000 Features laden, Referenzfeld erkennen, GP-IDs + extrahieren, PunktortTA deduplizieren. + 3. ``AX_Grenzpunkt`` für alle eindeutigen IDs per ResourceId-Batches laden. + 4. Per-Gemarkung-Statistik ausgeben. + 5. Angereicherten Join-Layer erzeugen und ins Projekt einfügen. + + Parameters + ---------- + tab_widget : + Eltern-Widget für Fehlerdialoge. + schluessel_liste : + Liste der 4-stelligen Gemarkungsnummern (z.B. ``['1234', '5678']``). + vg_geom : + ``QgsGeometry`` des Verfahrensgebiets (EPSG:25833). Wird für den + 200-m-Puffer zur Verschneidung der Gemarkungen verwendet. + ``None`` = keine Einschränkung (volle Gemarkungs-BBox). + progress : + Reserviert (wird nicht ausgewertet; interner ``ProgressDialog`` wird + verwendet). + + Returns + ------- + QgsVectorLayer | None + Angereicherter Memory-Layer mit Geometrie und allen GP-Attributen + (Prefix ``gp_``) oder ``None`` bei Fehler. + """ + import urllib.parse + from sn_basis.functions import QgsField + from sn_basis.functions.qt_wrapper import QVariant + + # Maximale Features pro einzelnem PunktortTA-GET-Aufruf + _BBOX_BATCH = 5000 + + project = QgsProject.instance() + + if not schluessel_liste: + QMessageBox.warning( + tab_widget, + "Keine Gemarkungen", + "Bitte mindestens einen Gemarkungsschlüssel angeben.", + ) + return None + + n_gemarkungen = len(schluessel_liste) + + # Fortschrittsschritte werden nach den Hits-Abfragen präzisiert. + # Vorab-Schätzung: 1 (Extents) + N*2 (Hits+Laden) + 1 (GP) + 1 (Join) + prog_vorab = n_gemarkungen * 2 + 3 + prog = create_progress_dialog( + prog_vorab, + "Grenzpunkte laden", + f"Lade Gemarkungs-Geometrien für {n_gemarkungen} Gemarkung(en) …", + ) + abgeschlossen = False + + try: + prog.set_value(0) + if prog.is_canceled(): + return None + + # ------------------------------------------------------------------ + # Schritt 1: Gemarkungsgeometrien + Verschneidung mit VG-Puffer + # ------------------------------------------------------------------ + if vg_geom and not vg_geom.isEmpty(): + extents = _lade_gemarkung_extents_mit_puffer(schluessel_liste, vg_geom, 200.0) + else: + extents = _lade_gemarkung_extents(schluessel_liste) + + if not extents: + prog.close() + QMessageBox.critical( + tab_widget, + "Fehler", + "Gemarkungs-BBoxen konnten nicht geladen werden.\n" + "Bitte Netzwerkverbindung und Gemarkungsschlüssel prüfen.\n" + f"Gesuchte Schlüssel: {', '.join(schluessel_liste)}", + ) + return None + + n_extents = len(extents) + if n_extents < n_gemarkungen: + warning( + "Grenzpunkte", + f"Nur {n_extents}/{n_gemarkungen} Gemarkungen im WFS gefunden. " + f"Nicht gefunden: {set(schluessel_liste) - set(extents.keys())}", + duration=8, + ) + + # ------------------------------------------------------------------ + # Schritt 2a: Hits-Abfragen für alle BBoxen + # ------------------------------------------------------------------ + prog.set_label(f"Anzahl PunktortTA-Objekte abfragen ({n_extents} Gemarkungen) …") + prog.set_value(1) + if prog.is_canceled(): + return None + + # {gemarkungsnummer: anzahl_features} – -1 = unbekannt + hits_pro_gemarkung: Dict[str, int] = {} + n_pta_batches_gesamt = 0 + + for gemarkungsnummer, bbox in extents.items(): + bbox_param = _punktort_ta_bbox_param(bbox) + n_hits = _wfs_hits_count(ALKIS_SF_URL, PUNKTORT_TA_TYPENAME, {'BBOX': bbox_param}) + hits_pro_gemarkung[gemarkungsnummer] = n_hits + batches = max(1, math.ceil(n_hits / _BBOX_BATCH)) if n_hits > 0 else 1 + n_pta_batches_gesamt += batches + info( + "Grenzpunkte (Diagnose)", + f"Gemarkung {gemarkungsnummer}: ~{n_hits} PunktortTA-Objekte " + f"→ {batches} Batch(es).", + duration=2, + ) + + # Gesamtschritte jetzt präzise: + # 1 (Extents/Hits) + n_pta_batches_gesamt (PunktortTA) + n_gp_batches (GP) + 1 (Join) + # GP-Batches kennen wir noch nicht – vorläufig 0, wird unten korrigiert. + prog_gesamt_pta = 1 + n_pta_batches_gesamt + prog.set_total(prog_gesamt_pta + 10) # Platzhalter für GP + Join + + # ------------------------------------------------------------------ + # Schritt 2b: PunktortTA pro Gemarkung + Batching + # ------------------------------------------------------------------ + # Statt rohe QgsFeature-Objekte (positionsbasierte Attribute!) zu sammeln, + # speichern wir (geom, attrs-dict) – vollständig namensbasiert. + # Das verhindert Feldwert-Verschiebungen, wenn verschiedene Batches + # die Felder in unterschiedlicher Reihenfolge liefern. + alle_pta_records: List[Dict] = [] # {'geom': QgsGeometry, 'attrs': {name: val}} + gesehene_pta_ids: set[str] = set() + pta_feld_namen: List[str] = [] + gehoert_zu_feld: Optional[str] = None + + per_gemarkung_gp_ids: Dict[str, set] = {} + prog_schritt = 1 # nach Hits-Phase + + for g_idx, (gemarkungsnummer, bbox) in enumerate(extents.items(), 1): + n_hits = hits_pro_gemarkung.get(gemarkungsnummer, -1) + n_batches_diese_gemarkung = max(1, math.ceil(n_hits / _BBOX_BATCH)) if n_hits > 0 else 1 + + if prog.is_canceled(): + return None + + gp_ids_dieser_gemarkung: set[str] = set() + neu_gesamt = 0 + duplikat_gesamt = 0 + bbox_param = _punktort_ta_bbox_param(bbox) + + for b_idx in range(n_batches_diese_gemarkung): + start_index = b_idx * _BBOX_BATCH + prog.set_label( + f"Gemarkung {gemarkungsnummer} ({g_idx}/{n_extents}), " + f"Batch {b_idx + 1}/{n_batches_diese_gemarkung} " + f"(~{n_hits if n_hits >= 0 else '?'} Objekte) …" + ) + prog.set_value(prog_schritt) + prog_schritt += 1 + if prog.is_canceled(): + return None + + extra: dict = {'BBOX': bbox_param} + if n_batches_diese_gemarkung > 1: + extra['COUNT'] = str(_BBOX_BATCH) + extra['STARTINDEX'] = str(start_index) + + info( + "Grenzpunkte (Diagnose)", + f"Gemarkung {gemarkungsnummer} ({g_idx}/{n_extents}), " + f"Batch {b_idx + 1}/{n_batches_diese_gemarkung}: " + f"STARTINDEX={start_index}, COUNT={_BBOX_BATCH} …", + duration=2, + ) + + _gml_bytes = _wfs_get_gml(ALKIS_SF_URL, PUNKTORT_TA_TYPENAME, extra) + if not _gml_bytes: + warning( + "Grenzpunkte (Diagnose)", + f"Gemarkung {gemarkungsnummer} Batch {b_idx + 1}: " + f"GET-Anfrage fehlgeschlagen – Batch wird übersprungen.", + duration=6, + ) + continue + + tmp = _parse_alkis_gml_memory_layer( + _gml_bytes, "AX_PunktortTA", "_TmpPTA", geom_prop="position", extract_qualitaet=True + ) + if not tmp: + continue + + cur_feld_namen = [f.name() for f in tmp.fields()] + + # Globale Feldliste inkrementell um neue Felder erweitern + # (Union über alle Batches/Gemarkungen). + for n in cur_feld_namen: + if n not in pta_feld_namen: + pta_feld_namen.append(n) + + # Referenzfeld beim ersten Batch erkennen, der es enthält + if gehoert_zu_feld is None: + gehoert_zu_feld = _erkenne_gehoert_zu_feld(tmp) + if gehoert_zu_feld: + info( + "Grenzpunkte", + f"Referenzfeld erkannt: '{gehoert_zu_feld}'. " + f"PunktortTA-Felder bisher: {', '.join(pta_feld_namen)}", + duration=5, + ) + elif not pta_feld_namen: + warning( + "Grenzpunkte (Diagnose)", + "Kein Referenzfeld in AX_PunktortTA erkannt. " + "Vorhandene Felder: " + ", ".join(cur_feld_namen), + duration=15, + ) + + neu = 0 + duplikat = 0 + for feat in tmp.getFeatures(): + pta_id = _extrahiere_feature_gml_id(feat, cur_feld_namen) + if pta_id and pta_id in gesehene_pta_ids: + duplikat += 1 + continue + if pta_id: + gesehene_pta_ids.add(pta_id) + + # Attribute namensbasiert als Dict speichern – verhindert + # Positions-Verschiebung beim späteren Aufbau von punktort_layer. + _attrs: Dict[str, str] = {} + for n in cur_feld_namen: + if n: + try: + _attrs[n] = feat[n] + except Exception: + pass + alle_pta_records.append({'geom': feat.geometry(), 'attrs': _attrs}) + neu += 1 + + if gehoert_zu_feld: + try: + raw = feat[gehoert_zu_feld] + raw = str(raw).strip() if raw else "" + gp_id = raw.rsplit(":", 1)[-1] if ":" in raw else raw + if gp_id: + gp_ids_dieser_gemarkung.add(gp_id) + except Exception: + pass + + neu_gesamt += neu + duplikat_gesamt += duplikat + + per_gemarkung_gp_ids[gemarkungsnummer] = gp_ids_dieser_gemarkung + info( + "Grenzpunkte", + f"Gemarkung {gemarkungsnummer}: {neu_gesamt} PunktortTA-Objekte " + f"({duplikat_gesamt} Duplikate übersprungen), " + f"{len(gp_ids_dieser_gemarkung)} GP-Referenzen.", + duration=4, + ) + + # Abbruch wenn keine Features + if not alle_pta_records: + prog.close() + QMessageBox.warning( + tab_widget, + "Keine Grenzpunkte", + "Für die angegebenen Gemarkungen wurden keine PunktortTA-Objekte gefunden.", + ) + return None + + if not gehoert_zu_feld: + prog.close() + QMessageBox.warning( + tab_widget, + "Kein Referenzfeld", + f"{len(alle_pta_records)} PunktortTA-Objekte geladen, " + f"aber kein Referenzfeld auf AX_Grenzpunkt gefunden.\n\n" + f"Gesuchte Feldnamen: {', '.join(_GEHOERT_ZU_KANDIDATEN)}\n" + f"Vorhandene Felder: {', '.join(pta_feld_namen)}", + ) + return None + + punktort_layer = QgsVectorLayer("Point?crs=EPSG:25833", PUNKTORT_LAYER_NAME, "memory") + pr_pta = punktort_layer.dataProvider() + pr_pta.addAttributes([QgsField(n, QVariant.String) for n in pta_feld_namen if n]) + punktort_layer.updateFields() + + # Features mit vollständig namensbasierter Attributzuweisung aufbauen – + # vermeidet Positions-Verschiebungen, die entstehen würden, wenn + # QgsFeatures aus tmp-Layern mit unterschiedlicher Feldreihenfolge + # positionsbasiert in punktort_layer übertragen werden. + neue_pta_features = [] + from sn_basis.functions import QgsFeature as _QgsFeature + for record in alle_pta_records: + new_f = _QgsFeature(punktort_layer.fields()) + new_f.setGeometry(record['geom']) + for n, val in record['attrs'].items(): + if n in pta_feld_namen: + try: + new_f[n] = val + except Exception: + pass + neue_pta_features.append(new_f) + pr_pta.addFeatures(neue_pta_features) + punktort_layer.updateExtents() + + alle_gp_ids: set[str] = set() + for ids in per_gemarkung_gp_ids.values(): + alle_gp_ids.update(ids) + + n_pta_gesamt = len(alle_pta_records) + n_gp_ids = len(alle_gp_ids) + info( + "Grenzpunkte", + f"Gesamt: {n_pta_gesamt} PunktortTA-Objekte, {n_gp_ids} eindeutige GP-Referenzen " + f"aus {n_extents} Gemarkung(en).", + duration=5, + ) + + if not alle_gp_ids: + prog.close() + QMessageBox.warning( + tab_widget, + "Keine Verknüpfungen", + f"{n_pta_gesamt} PunktortTA-Objekte geladen, aber Referenzfeld " + f"'{gehoert_zu_feld}' enthält keine gültigen IDs.", + ) + return None + + # ------------------------------------------------------------------ + # Schritt 3: AX_Grenzpunkt per ResourceId-Batches + # ------------------------------------------------------------------ + anzahl_gp_batches = math.ceil(n_gp_ids / _BATCH_GROESSE) + # Gesamtschritte jetzt vollständig bekannt + prog_gesamt = prog_schritt + anzahl_gp_batches + 1 + prog.set_total(prog_gesamt) + prog.set_label( + f"Grenzpunkte laden: {n_gp_ids} IDs in {anzahl_gp_batches} Batch(es) …" + ) + prog.set_value(prog_schritt) + if prog.is_canceled(): + return None + + def _gp_progress(batch_idx: int) -> None: + prog.set_label( + f"AX_Grenzpunkt Batch {batch_idx + 1}/{anzahl_gp_batches} …" + ) + prog.set_value(prog_schritt + batch_idx + 1) + + gp_dict, gp_feld_namen = _lade_grenzpunkte_als_dict( + sorted(alle_gp_ids), progress_callback=_gp_progress + ) + + if not gp_dict: + prog.close() + warning( + "Grenzpunkte", + "Grenzpunkt-Daten konnten nicht geladen werden " + "(IDs vorhanden, WFS lieferte kein Ergebnis).", + duration=6, + ) + return None + + info( + "Grenzpunkte", + f"{len(gp_dict)}/{n_gp_ids} Grenzpunkte vom WFS empfangen " + f"({n_gp_ids - len(gp_dict)} IDs nicht aufgelöst).", + duration=4, + ) + + for gemarkungsnummer, gp_ids in per_gemarkung_gp_ids.items(): + if not gp_ids: + continue + n_gefunden = sum(1 for gid in gp_ids if gid in gp_dict) + info( + "Grenzpunkte", + f"Gemarkung {gemarkungsnummer}: {n_gefunden}/{len(gp_ids)} " + f"Grenzpunkte geladen.", + duration=6, + ) + + # ------------------------------------------------------------------ + # Schritt 4: Angereicherten Join-Layer erstellen + # ------------------------------------------------------------------ + prog.set_label("Angereicherten Layer erstellen …") + prog.set_value(prog_gesamt - 1) + if prog.is_canceled(): + return None + + joined_layer = _erstelle_joined_layer( + punktort_layer, gp_dict, gp_feld_namen, gehoert_zu_feld + ) + + if not joined_layer or not joined_layer.isValid(): + prog.close() + QMessageBox.critical( + tab_widget, + "Fehler", + "Angereicherter Grenzpunkt-Layer konnte nicht erstellt werden.", + ) + return None + + project.addMapLayer(joined_layer) + + prog.set_value(prog_gesamt) + prog.close() + abgeschlossen = True + + info( + "Grenzpunkte", + f"Fertig: {joined_layer.featureCount()} Grenzpunkte im Layer " + f"({len(gp_dict)} mit GP-Attributen, " + f"{joined_layer.featureCount() - len(gp_dict)} ohne GP-Match).", + duration=10, + ) + + return joined_layer + + finally: + if not abgeschlossen: + prog.close() + diff --git a/functions/knickpunkt_pruefung.py b/functions/knickpunkt_pruefung.py new file mode 100644 index 0000000..4044c74 --- /dev/null +++ b/functions/knickpunkt_pruefung.py @@ -0,0 +1,467 @@ +"""sn_verfahrensgebiet/functions/knickpunkt_pruefung.py – Knickpunktbasierte BROR-Prüfung. + +Vergleicht die Knickpunkte (Stützpunkte) des BauRaumOderBodenordnungsrecht- +Polygons mit den Punktorten der geladenen ``AX_PunktortTA``-Objekte und +identifiziert: + +- **Identische Punkte** (grün): An einem BROR-Knickpunkt liegt ein PunktortTA. +- **Fehlende PunktortTA** (rot): Ein BROR-Knickpunkt hat keinen zugehörigen PunktortTA. +- **Fehlende Knickpunkte** (orange): Ein PunktortTA liegt nahe des BROR-Umrings + auf einer Flurstücksgrenze, hat aber keinen entsprechenden Knickpunkt im BROR. + +**Algorithmus:** + +1. BROR-Polygon-Stützpunkte (Knickpunkte) extrahieren. +2. Spatial-Index auf PunktortTA aufbauen. +3. Für jeden Knickpunkt nächsten PunktortTA im Radius von + :data:`KOORDINATEN_TOLERANZ_M` suchen → grün oder rot. + Gematchte PunktortTA-Feature-IDs merken. +4. Gepufferter BROR-Umring + Flurstück-Vertex-Set ermitteln. +5. Nicht-gematchte PunktortTA nahe BROR-Umring, die auf einer Flurstücksgrenze + liegen → orange. + +**Rückgabe:** ``(result_layer, identisch, kein_pta, fehlender_knick)`` +""" +from __future__ import annotations + +from typing import Dict, List, Optional, Set, Tuple + +from sn_basis.functions.qgiscore_wrapper import ( + QgsVectorLayer, + QgsFeature, + QgsField, + QgsGeometry, +) +from sn_basis.functions.qt_wrapper import QVariant + +# --------------------------------------------------------------------------- +# Konstanten +# --------------------------------------------------------------------------- + +#: Koordinatentoleranz für die Knickpunkt ↔ PunktortTA-Übereinstimmung (m). +KOORDINATEN_TOLERANZ_M: float = 0.01 + +#: Puffer um den BROR-Umring für die Suche nach fehlenden Knickpunkten (m). +BOUNDARY_PUFFER_M: float = 0.5 + +#: Rundungsgenauigkeit für Vertex-Koordinaten im Flurstück-Set (Dezimalstellen). +_VERTEX_ROUND = 3 + +ERGEBNIS_LAYER_NAME = "Knickpunktprüfung" +FELD_TYP = "typ" +WERT_IDENTISCH = "identisch" +WERT_KEIN_PTA = "kein_pta" +WERT_FEHLENDER_KNICK = "fehlender_knick" + + +# --------------------------------------------------------------------------- +# Interne Hilfsfunktionen +# --------------------------------------------------------------------------- + +def _extrahiere_knickpunkte(bror_layer) -> List[object]: + """Extrahiert alle Stützpunkte (Knickpunkte) des BROR-Polygons. + + Doppelte Start/End-Punkte eines Rings werden dedupliziert. Gibt eine + flache Liste von ``QgsPointXY``-Objekten zurück. + + Parameters + ---------- + bror_layer : + BauRaumOderBodenordnungsrecht-Polygon-Layer. + + Returns + ------- + List[QgsPointXY] + Alle eindeutigen Stützpunkte; leer bei Fehler. + """ + try: + from qgis.core import QgsPointXY # type: ignore[import] + + bror_feats = [ + f for f in bror_layer.getFeatures() + if f.geometry() and not f.geometry().isEmpty() + ] + if not bror_feats: + return [] + + union_geom = QgsGeometry.unaryUnion([f.geometry() for f in bror_feats]) + if union_geom is None or union_geom.isEmpty(): + return [] + + punkte: List[object] = [] + seen: Set[Tuple[float, float]] = set() + + def _add_ring(vertices): + # Letzter Punkt = erster Punkt (Ring geschlossen) → überspringen + ring_pts = list(vertices) + n = len(ring_pts) + if n >= 2: + # Prüfen ob letzter Punkt identisch mit erstem + last = n - 1 + if ( + abs(ring_pts[last].x() - ring_pts[0].x()) < 1e-9 + and abs(ring_pts[last].y() - ring_pts[0].y()) < 1e-9 + ): + ring_pts = ring_pts[:-1] + for pt in ring_pts: + key = (round(pt.x(), _VERTEX_ROUND), round(pt.y(), _VERTEX_ROUND)) + if key not in seen: + seen.add(key) + punkte.append(pt) + + geom_type = union_geom.wkbType() + # Polygon oder MultiPolygon + if union_geom.isMultipart(): + for poly in union_geom.asMultiPolygon(): + for ring in poly: + _add_ring([QgsPointXY(x, y) for x, y in ring]) + else: + for ring in union_geom.asPolygon(): + _add_ring([QgsPointXY(x, y) for x, y in ring]) + + return punkte + + except Exception: + return [] + + +def _erstelle_ergebnis_layer(crs_authid: str) -> object: + """Erzeugt den leeren Punkt-Ergebnislayer mit Feld ``typ``. + + Parameters + ---------- + crs_authid : + CRS-Kennung, z.B. ``"EPSG:25833"``. + + Returns + ------- + QgsVectorLayer + Leerer Memory-Layer vom Typ Point. + """ + layer = QgsVectorLayer( + f"Point?crs={crs_authid}", ERGEBNIS_LAYER_NAME, "memory" + ) + pr = layer.dataProvider() + pr.addAttributes([QgsField(FELD_TYP, QVariant.String)]) + layer.updateFields() + return layer + + +def _setze_kategorien_symbologie(layer) -> None: + """Wendet Kategoriesymbologie an: grün/rot/orange Punkte. + + - ``identisch`` → grün + - ``kein_pta`` → rot + - ``fehlender_knick`` → orange + + Wird bei fehlendem QGIS-Kontext still ignoriert. + """ + try: + from qgis.core import ( # type: ignore[import] + QgsCategorizedSymbolRenderer, + QgsRendererCategory, + QgsMarkerSymbol, + ) + + def _marker(color: str, size: str = "2") -> object: + return QgsMarkerSymbol.createSimple( + {"color": color, "size": size, "outline_style": "no"} + ) + + categories = [ + QgsRendererCategory( + WERT_IDENTISCH, + _marker("0,180,0,255"), + "identisch (Knickpunkt mit PunktortTA)", + ), + QgsRendererCategory( + WERT_KEIN_PTA, + _marker("220,0,0,255"), + "kein PunktortTA an Knickpunkt", + ), + QgsRendererCategory( + WERT_FEHLENDER_KNICK, + _marker("255,128,0,255"), + "fehlender Knickpunkt (PunktortTA ohne BROR-Knickpunkt)", + ), + ] + renderer = QgsCategorizedSymbolRenderer(FELD_TYP, categories) + layer.setRenderer(renderer) + except Exception: + pass # Kein QGIS-Kontext (Tests) → Standarddarstellung behalten + + +def _baue_pta_index(punktort_layer) -> Tuple[object, Dict[int, object]]: + """Baut einen QgsSpatialIndex und eine ID→Geometrie-Map für PunktortTA. + + Parameters + ---------- + punktort_layer : + Geladener ``AX_PunktortTA``-Layer. + + Returns + ------- + Tuple[QgsSpatialIndex, Dict[int, QgsGeometry]] + Index und Feature-ID-zu-Geometrie-Dict. + """ + from qgis.core import QgsSpatialIndex # type: ignore[import] + + index = QgsSpatialIndex() + geom_map: Dict[int, object] = {} + for feat in punktort_layer.getFeatures(): + geom = feat.geometry() + if geom and not geom.isEmpty(): + index.insertFeature(feat) + geom_map[feat.id()] = geom + return index, geom_map + + +def _baue_flst_vertex_set( + flst_layer, + bror_buffer_geom, +) -> Set[Tuple[float, float]]: + """Erstellt ein Set gerundeter (x, y)-Koordinaten aller Flurstück-Vertices + innerhalb des gepufferten BROR-Umrings. + + Parameters + ---------- + flst_layer : + Flurstücke-Layer. + bror_buffer_geom : + Gepufferter BROR-Umring (Polygon-Geometrie). + + Returns + ------- + Set[Tuple[float, float]] + Gerundete Vertex-Koordinaten. + """ + vertex_set: Set[Tuple[float, float]] = set() + if flst_layer is None: + return vertex_set + + try: + for feat in flst_layer.getFeatures(): + geom = feat.geometry() + if geom is None or geom.isEmpty(): + continue + try: + # Nur Flurstücke berücksichtigen, die den Puffer berühren + if not geom.intersects(bror_buffer_geom): + continue + boundary = geom.convertToType(1) # 1 = Line + if boundary is None or boundary.isEmpty(): + continue + if boundary.isMultipart(): + parts = boundary.asMultiPolyline() + else: + parts = [boundary.asPolyline()] + for line in parts: + for pt in line: + vertex_set.add( + (round(pt.x(), _VERTEX_ROUND), round(pt.y(), _VERTEX_ROUND)) + ) + except Exception: + continue + except Exception: + pass + return vertex_set + + +# --------------------------------------------------------------------------- +# Öffentliche API +# --------------------------------------------------------------------------- + +def knickpunkt_pruefung( + bror_layer, + punktort_layer, + flst_layer, + progress=None, + knickpunkte: Optional[List] = None, +) -> Tuple[Optional[object], int, int, int]: + """Knickpunktbasierte BROR-Prüfung. + + Vergleicht BROR-Knickpunkte mit ``AX_PunktortTA``-Objekten und ermittelt + fehlende Knickpunkte anhand von Flurstücksgrenzen. + + Parameters + ---------- + bror_layer : + BauRaumOderBodenordnungsrecht-Polygon-Layer. + punktort_layer : + Geladener ``AX_PunktortTA``-Layer (Ergebnis aus ``grenzpunkt_alkis``). + flst_layer : + Flurstücke-Layer für die Orange-Prüfung. Kann ``None`` sein. + progress : + Optionaler ``ProgressDialog`` für Fortschrittsanzeige. + knickpunkte : + Vorberechnete Knickpunktliste (optional). Wenn ``None``, werden die + Knickpunkte aus ``bror_layer`` extrahiert. + + Returns + ------- + Tuple[Optional[QgsVectorLayer], int, int, int] + ``(result_layer, identisch_count, kein_pta_count, fehlender_knick_count)`` + + - ``result_layer``: Punkt-Layer mit grünen/roten/orangen Punkten, + oder ``None`` bei Abbruch oder fehlgeschlagener Verarbeitung. + - ``identisch_count``: Knickpunkte mit passendem PunktortTA (grün). + - ``kein_pta_count``: Knickpunkte ohne passenden PunktortTA (rot). + - ``fehlender_knick_count``: PunktortTA ohne BROR-Knickpunkt (orange). + """ + + def _canceled() -> bool: + return progress is not None and progress.is_canceled() + + def _schritt(label: str, wert: int) -> None: + if progress is not None: + progress.set_label(label) + progress.set_value(wert) + + # ------------------------------------------------------------------ + # Schritt 0: Knickpunkte ermitteln + # ------------------------------------------------------------------ + _schritt("Knickpunktprüfung: Knickpunkte extrahieren …", 0) + + if knickpunkte is None: + knickpunkte = _extrahiere_knickpunkte(bror_layer) + + if not knickpunkte: + return None, 0, 0, 0 + + n = len(knickpunkte) + crs_authid = bror_layer.crs().authid() + + if _canceled(): + return None, 0, 0, 0 + + # ------------------------------------------------------------------ + # Schritt 1: Spatial-Index für PunktortTA aufbauen + # ------------------------------------------------------------------ + _schritt("Knickpunktprüfung: PunktortTA-Index aufbauen …", 1) + + if punktort_layer is None: + return None, 0, 0, 0 + + try: + pta_index, geom_map = _baue_pta_index(punktort_layer) + except Exception: + return None, 0, 0, 0 + + if _canceled(): + return None, 0, 0, 0 + + # ------------------------------------------------------------------ + # Schritte 2…n+1: Jeden Knickpunkt mit PunktortTA vergleichen + # ------------------------------------------------------------------ + try: + from qgis.core import QgsRectangle # type: ignore[import] + except ImportError: + return None, 0, 0, 0 + + ergebnis_layer = _erstelle_ergebnis_layer(crs_authid) + pr = ergebnis_layer.dataProvider() + feats_to_add: List[object] = [] + + identisch_count = 0 + kein_pta_count = 0 + matched_pta_ids: Set[int] = set() + + for i, knick_pt in enumerate(knickpunkte): + if _canceled(): + return None, 0, 0, 0 + + _schritt(f"Knickpunktprüfung: Knickpunkt {i + 1}/{n} prüfen …", i + 2) + + x, y = knick_pt.x(), knick_pt.y() + t = KOORDINATEN_TOLERANZ_M + suchbox = QgsRectangle(x - t, y - t, x + t, y + t) + kandidaten = pta_index.intersects(suchbox) + + gefunden = False + for fid in kandidaten: + pta_geom = geom_map.get(fid) + if pta_geom is None: + continue + pta_pt = pta_geom.asPoint() + dist = ((pta_pt.x() - x) ** 2 + (pta_pt.y() - y) ** 2) ** 0.5 + if dist <= KOORDINATEN_TOLERANZ_M: + gefunden = True + matched_pta_ids.add(fid) + break + + feat = QgsFeature() + try: + from qgis.core import QgsPointXY # type: ignore[import] + feat.setGeometry(QgsGeometry.fromPointXY(QgsPointXY(x, y))) + except Exception: + continue + feat.setAttributes([WERT_IDENTISCH if gefunden else WERT_KEIN_PTA]) + feats_to_add.append(feat) + + if gefunden: + identisch_count += 1 + else: + kein_pta_count += 1 + + if _canceled(): + return None, 0, 0, 0 + + # ------------------------------------------------------------------ + # Schritt n+2: Fehlende Knickpunkte (orange) ermitteln + # ------------------------------------------------------------------ + _schritt("Knickpunktprüfung: Fehlende Knickpunkte ermitteln …", n + 2) + + fehlender_knick_count = 0 + + # BROR-Umring ermitteln und puffern + try: + bror_feats = [ + f for f in bror_layer.getFeatures() + if f.geometry() and not f.geometry().isEmpty() + ] + bror_union = QgsGeometry.unaryUnion([f.geometry() for f in bror_feats]) + bror_boundary = bror_union.convertToType(1) if bror_union else None + bror_buffer_geom = ( + bror_boundary.buffer(BOUNDARY_PUFFER_M, 8) if bror_boundary else None + ) + except Exception: + bror_buffer_geom = None + + if bror_buffer_geom is not None and flst_layer is not None: + flst_vertex_set = _baue_flst_vertex_set(flst_layer, bror_buffer_geom) + + for feat in punktort_layer.getFeatures(): + if _canceled(): + break + if feat.id() in matched_pta_ids: + continue + geom = feat.geometry() + if geom is None or geom.isEmpty(): + continue + try: + if not geom.within(bror_buffer_geom): + continue + pta_pt = geom.asPoint() + key = (round(pta_pt.x(), _VERTEX_ROUND), round(pta_pt.y(), _VERTEX_ROUND)) + if key in flst_vertex_set: + orange_feat = QgsFeature() + orange_feat.setGeometry(geom) + orange_feat.setAttributes([WERT_FEHLENDER_KNICK]) + feats_to_add.append(orange_feat) + fehlender_knick_count += 1 + except Exception: + continue + + if _canceled(): + return None, 0, 0, 0 + + # ------------------------------------------------------------------ + # Schritt n+3: Ergebnislayer fertigstellen + # ------------------------------------------------------------------ + _schritt("Knickpunktprüfung: Ergebnislayer erstellen …", n + 3) + + pr.addFeatures(feats_to_add) + ergebnis_layer.updateExtents() + _setze_kategorien_symbologie(ergebnis_layer) + + return ergebnis_layer, identisch_count, kein_pta_count, fehlender_knick_count diff --git a/functions/umringstatistik_grenzpunkte.py b/functions/umringstatistik_grenzpunkte.py new file mode 100644 index 0000000..3fe98e8 --- /dev/null +++ b/functions/umringstatistik_grenzpunkte.py @@ -0,0 +1,373 @@ +"""sn_verfahrensgebiet/functions/umringstatistik_grenzpunkte.py – Umringstatistik Grenzpunkte. + +Extrahiert aus einem vorhandenen ``Grenzpunkte (ALKIS)``-Layer alle Punkte, +die auf dem Umring des Verfahrensgebiets liegen, und klassifiziert sie in +drei Kategorien: + +1. **FGP** (grün): ``gp_festgestellterGrenzpunkt == "true"`` +2. **<3cm** (gelb): ``gp_festgestellterGrenzpunkt != "true"`` + UND ``qualitaetsangaben_ax_dqpunktort_genauigkeitsstufe == 2100`` +3. **>3cm** (rot): ``gp_festgestellterGrenzpunkt != "true"`` + UND ``qualitaetsangaben_ax_dqpunktort_genauigkeitsstufe > 2100`` +4. **unbekannt** (grau): keiner der obigen Fälle + +Der Umring-Test verwendet einen Abstandsschwellwert von :data:`UMRING_TOLERANZ_M`. + +**Rückgabe:** Memory-Layer ``Umringstatistik Grenzpunkte`` oder ``None`` bei Fehler. +""" +from __future__ import annotations + +from typing import Optional + +from sn_basis.functions import QgsProject +from sn_basis.functions.qt_wrapper import QMessageBox +from sn_basis.functions.dialog_wrapper import create_progress_dialog +from sn_basis.functions.message_wrapper import warning +from sn_basis.functions.qgiscore_wrapper import ( + QgsVectorLayer, + QgsFeature, + QgsField, + QgsGeometry, + polygon_zu_linie, +) +from sn_basis.functions.qt_wrapper import QVariant + +# --------------------------------------------------------------------------- +# Konstanten +# --------------------------------------------------------------------------- + +#: Anzeigename des Ergebnis-Layers. +ERGEBNIS_LAYER_NAME = "Umringstatistik Grenzpunkte" + +#: Feldname für die Klassifikationskategorie im Ergebnis-Layer. +FELD_KATEGORIE = "kategorie" + +#: Feldname für ``gp_festgestellterGrenzpunkt`` im Grenzpunkte(ALKIS)-Layer. +FELD_FESTGESTELLT = "gp_festgestellterGrenzpunkt" + +#: Feldname für das Genauigkeitsattribut im Grenzpunkte(ALKIS)-Layer. +FELD_GENAUIGKEIT = "qualitaetsangaben_ax_dqpunktort_genauigkeitsstufe" + +#: Feldname für die GML-ID im Grenzpunkte(ALKIS)-Layer. +FELD_GML_ID = "gml_id" + +#: Kategoriewert: festgestellter Grenzpunkt. +KATEGORIE_FGP = "FGP" +#: Kategoriewert: Lagegenauigkeit ≤ 3 cm (Gütestufe 2100). +KATEGORIE_KLEIN = "<3cm" +#: Kategoriewert: Lagegenauigkeit > 3 cm (schlechtere Gütestufe). +KATEGORIE_GROSS = ">3cm" +#: Kategoriewert: Kategorie konnte nicht bestimmt werden. +KATEGORIE_UNBEKANNT = "unbekannt" + +#: Maximaler Abstand (m) eines Punktes vom Umring, um als „auf dem Umring" zu gelten. +UMRING_TOLERANZ_M: float = 0.1 + +#: Grenzwert für ``qualitaetsangaben_ax_dqpunktort_genauigkeitsstufe`` (3-cm-Schwelle). +GENAUIGKEIT_GRENZWERT: int = 2100 + + +# --------------------------------------------------------------------------- +# Interne Hilfsfunktionen +# --------------------------------------------------------------------------- + +def _setze_kategorien_symbologie(layer) -> None: + """Wendet Kategoriesymbologie an: grün / gelb / rot / grau. + + Wird bei fehlendem QGIS-Kontext still ignoriert. + """ + try: + from qgis.core import ( # type: ignore[import] + QgsCategorizedSymbolRenderer, + QgsRendererCategory, + QgsMarkerSymbol, + ) + + def _marker(color: str, size: str = "2.5") -> object: + return QgsMarkerSymbol.createSimple( + {"color": color, "size": size, "outline_style": "no"} + ) + + categories = [ + QgsRendererCategory( + KATEGORIE_FGP, + _marker("0,180,0,255"), + f"{KATEGORIE_FGP} – festgestellter Grenzpunkt", + ), + QgsRendererCategory( + KATEGORIE_KLEIN, + _marker("255,220,0,255"), + f"{KATEGORIE_KLEIN} – Gütestufe {GENAUIGKEIT_GRENZWERT}", + ), + QgsRendererCategory( + KATEGORIE_GROSS, + _marker("220,0,0,255"), + f"{KATEGORIE_GROSS} – Gütestufe > {GENAUIGKEIT_GRENZWERT}", + ), + QgsRendererCategory( + KATEGORIE_UNBEKANNT, + _marker("128,128,128,255"), + f"{KATEGORIE_UNBEKANNT} – Kategorie nicht bestimmbar", + ), + ] + renderer = QgsCategorizedSymbolRenderer(FELD_KATEGORIE, categories) + layer.setRenderer(renderer) + except Exception: + pass # Kein QGIS-Kontext (Tests) → Standarddarstellung behalten + + +def _klassifiziere(feat) -> str: + """Bestimmt die Kategorie eines Grenzpunkts anhand seiner Attribute. + + Parameters + ---------- + feat : + Feature aus dem Grenzpunkte(ALKIS)-Layer. + + Returns + ------- + str + Einer der Kategorienwerte :data:`KATEGORIE_FGP`, :data:`KATEGORIE_KLEIN`, + :data:`KATEGORIE_GROSS` oder :data:`KATEGORIE_UNBEKANNT`. + """ + # festgestellterGrenzpunkt + try: + festgestellt_raw = feat[FELD_FESTGESTELLT] + festgestellt = str(festgestellt_raw).lower().strip() == "true" if festgestellt_raw else False + except Exception: + festgestellt = False + + if festgestellt: + return KATEGORIE_FGP + + # Genauigkeitsstufe auslesen + try: + gen_raw = feat[FELD_GENAUIGKEIT] + genauigkeit = int(str(gen_raw).strip()) if gen_raw else None + except (ValueError, TypeError, Exception): + genauigkeit = None + + if genauigkeit is None: + return KATEGORIE_UNBEKANNT + if genauigkeit == GENAUIGKEIT_GRENZWERT: + return KATEGORIE_KLEIN + if genauigkeit > GENAUIGKEIT_GRENZWERT: + return KATEGORIE_GROSS + # Genauigkeit < 2100 (besser als 3 cm, aber nicht festgestellt) + return KATEGORIE_UNBEKANNT + + +# --------------------------------------------------------------------------- +# Öffentliche API +# --------------------------------------------------------------------------- + +def umringstatistik_grenzpunkte( + tab_widget, + grenzpunkte_layer, + vg_layer, +) -> Optional[object]: + """Klassifiziert Grenzpunkte auf dem VG-Umring nach Lagegenauigkeit. + + Ablauf: + + 1. VG-Geometrie aus ``vg_layer`` berechnen → Umringslinie. + 2. Alle Features aus ``grenzpunkte_layer`` auf Abstand ≤ :data:`UMRING_TOLERANZ_M` + zum Umring prüfen. + 3. Treffer klassifizieren (FGP / <3cm / >3cm / unbekannt). + 4. Ergebnis-Memory-Layer aufbauen und Kategoriesymbologie setzen. + 5. Layer ins Projekt einfügen, Zusammenfassungs-MessageBox anzeigen. + + Parameters + ---------- + tab_widget : + Eltern-Widget für Dialoge. + grenzpunkte_layer : + Geladener ``Grenzpunkte (ALKIS)``-Layer mit Attributen ``gp_festgestellterGrenzpunkt`` + und ``qualitaetsangaben_ax_dqpunktort_genauigkeitsstufe``. + vg_layer : + Verfahrensgebiet-Layer (Polygon, EPSG:25833). + + Returns + ------- + QgsVectorLayer | None + Ergebnis-Memory-Layer oder ``None`` bei Fehler/Abbruch. + """ + project = QgsProject.instance() + prog = create_progress_dialog(5, "Umringstatistik Grenzpunkte", "Schritt 1/5: VG-Umring berechnen …") + abgeschlossen = False + + try: + prog.set_value(0) + if prog.is_canceled(): + return None + + # ------------------------------------------------------------------ + # Schritt 1: VG-Umring berechnen + # ------------------------------------------------------------------ + feats_vg = [f for f in vg_layer.getFeatures() if f.hasGeometry() and not f.geometry().isEmpty()] + if not feats_vg: + prog.close() + warning("Umringstatistik", "Verfahrensgebiet-Layer enthält keine Geometrien.", duration=6) + return None + + vg_geom = QgsGeometry.unaryUnion([f.geometry() for f in feats_vg]) + if vg_geom is None or vg_geom.isEmpty(): + prog.close() + warning("Umringstatistik", "Verfahrensgebiet-Geometrie konnte nicht vereint werden.", duration=6) + return None + + umring_geom = polygon_zu_linie(vg_geom) + if umring_geom is None or umring_geom.isEmpty(): + prog.close() + warning("Umringstatistik", "Umringslinie konnte aus der VG-Geometrie nicht extrahiert werden.", duration=6) + return None + + prog.set_value(1) + if prog.is_canceled(): + return None + + # ------------------------------------------------------------------ + # Schritt 2: Grenzpunkte auf dem Umring filtern + # ------------------------------------------------------------------ + prog.set_label("Schritt 2/5: Grenzpunkte auf Umring prüfen …") + + n_gesamt = grenzpunkte_layer.featureCount() + treffer: list = [] + + for i, feat in enumerate(grenzpunkte_layer.getFeatures()): + if i % 200 == 0: + prog.set_label( + f"Schritt 2/5: Grenzpunkte prüfen … ({i}/{n_gesamt})" + ) + if prog.is_canceled(): + return None + + if not feat.hasGeometry() or feat.geometry().isEmpty(): + continue + try: + dist = feat.geometry().distance(umring_geom) + except Exception: + continue + if dist <= UMRING_TOLERANZ_M: + treffer.append(feat) + + prog.set_value(2) + if prog.is_canceled(): + return None + + if not treffer: + prog.close() + QMessageBox.information( + tab_widget, + "Umringstatistik Grenzpunkte", + "Es wurden keine Grenzpunkte auf dem Umring des Verfahrensgebiets gefunden.\n\n" + f"Geprüft: {n_gesamt} Grenzpunkte\n" + f"Toleranz: {UMRING_TOLERANZ_M} m", + ) + return None + + # ------------------------------------------------------------------ + # Schritt 3: Klassifizieren + # ------------------------------------------------------------------ + prog.set_label(f"Schritt 3/5: {len(treffer)} Treffer klassifizieren …") + + n_fgp = 0 + n_klein = 0 + n_gross = 0 + n_unbekannt = 0 + + klassifiziert: list = [] # [(feat, kategorie), ...] + for feat in treffer: + kat = _klassifiziere(feat) + klassifiziert.append((feat, kat)) + if kat == KATEGORIE_FGP: + n_fgp += 1 + elif kat == KATEGORIE_KLEIN: + n_klein += 1 + elif kat == KATEGORIE_GROSS: + n_gross += 1 + else: + n_unbekannt += 1 + + prog.set_value(3) + if prog.is_canceled(): + return None + + # ------------------------------------------------------------------ + # Schritt 4: Ergebnis-Layer aufbauen + # ------------------------------------------------------------------ + prog.set_label("Schritt 4/5: Ergebnis-Layer aufbauen …") + + crs_authid = vg_layer.crs().authid() or "EPSG:25833" + result_layer = QgsVectorLayer( + f"Point?crs={crs_authid}", ERGEBNIS_LAYER_NAME, "memory" + ) + pr = result_layer.dataProvider() + + felder = [ + QgsField(FELD_KATEGORIE, QVariant.String), + QgsField(FELD_GML_ID, QVariant.String), + QgsField(FELD_FESTGESTELLT, QVariant.String), + QgsField(FELD_GENAUIGKEIT, QVariant.String), + ] + pr.addAttributes(felder) + result_layer.updateFields() + + neue_features = [] + for feat, kat in klassifiziert: + new_f = QgsFeature(result_layer.fields()) + new_f.setGeometry(feat.geometry()) + + new_f[FELD_KATEGORIE] = kat + + for src_feld in (FELD_GML_ID, FELD_FESTGESTELLT, FELD_GENAUIGKEIT): + try: + new_f[src_feld] = feat[src_feld] + except Exception: + pass + + neue_features.append(new_f) + + pr.addFeatures(neue_features) + result_layer.updateExtents() + _setze_kategorien_symbologie(result_layer) + + prog.set_value(4) + if prog.is_canceled(): + return None + + # ------------------------------------------------------------------ + # Schritt 5: Layer ins Projekt + Ergebnis-MessageBox + # ------------------------------------------------------------------ + prog.set_label("Schritt 5/5: Layer ins Projekt einfügen …") + + # Bestehenden Ergebnis-Layer ggf. ersetzen + vorhandene = project.mapLayersByName(ERGEBNIS_LAYER_NAME) + for lyr in vorhandene: + project.removeMapLayer(lyr.id()) + + project.addMapLayer(result_layer) + + prog.set_value(5) + abgeschlossen = True + prog.close() + + n_umring = len(treffer) + QMessageBox.information( + tab_widget, + "Umringstatistik Grenzpunkte", + f"Grenzpunkte auf dem Umring: {n_umring}\n" + f"\n" + f" FGP (festgestellt): {n_fgp}\n" + f" < 3 cm (Gütestufe {GENAUIGKEIT_GRENZWERT}): {n_klein}\n" + f" > 3 cm (schlechtere Güte): {n_gross}\n" + f" Unbekannt: {n_unbekannt}\n" + f"\n" + f"(Geprüft: {n_gesamt} Grenzpunkte, Toleranz: {UMRING_TOLERANZ_M} m)", + ) + + return result_layer + + finally: + if not abgeschlossen: + prog.close() diff --git a/functions/verfahrensgebiet_alkis_komplett.py b/functions/verfahrensgebiet_alkis_komplett.py index 25ee5ea..2d98739 100644 --- a/functions/verfahrensgebiet_alkis_komplett.py +++ b/functions/verfahrensgebiet_alkis_komplett.py @@ -12,13 +12,18 @@ optionaler Umringsprüfung: 6. Flächenvergleich zwischen BROR-Objekt und abgeleitetem Verfahrensgebiet. - **Übereinstimmung** (≤ Toleranz): modale OK-Meldung, Ablauf beendet. - - **Abweichung**: Ja/Nein-Abfrage für Detailprüfung. + - **Abweichung**: 3-Optionen-Dialog (Detailprüfung / Als VG laden / Abbrechen). -7. Optionale Umringsprüfung (Variante 1): +7. Bei Wahl „Als VG laden": - - BROR-Umring gegen Flurstücksgrenzen vergleichen. - - Ergebnislayer (grün/rot) ins Projekt einfügen. - - Abschlussmeldung mit Konfliktanzahl. + - BROR-Objekt direkt mit VG-Namen ins Projekt übernehmen. + +8. Bei Wahl „Detailprüfung" (knickpunktbasiert): + + - Grenzpunkte (AX_PunktortTA) per BBox laden. + - Knickpunkte des BROR mit PunktortTA vergleichen (grün/rot/orange). + - Ergebnislayer (Punkte) ins Projekt einfügen. + - Abschlussmeldung mit Zählung. Das BROR-Objekt (``AX_BauRaumOderBodenordnungsrecht``) bleibt bei Erfolg als temporärer Layer ``BauRaumOrdnungsrecht (ALKIS)`` im Projekt erhalten. @@ -42,7 +47,7 @@ from sn_basis.functions.variable_wrapper import get_variable from sn_basis.functions.dialog_wrapper import ( create_progress_dialog, show_info_dialog, - ask_yes_no, + ask_detailpruefung_oder_vg_laden, ) from sn_verfahrensgebiet.functions.verfahrensgebiet_alkis import ( @@ -57,7 +62,11 @@ from sn_verfahrensgebiet.functions.verfahrensgebiet_edit_service import ( erstelle_verfahrensgebiet_aus_auswahl, VERF_LAYER_NAME, ) -from sn_verfahrensgebiet.functions.umringspruefung import umringspruefung +from sn_verfahrensgebiet.functions.grenzpunkt_alkis import lade_punktort_ta +from sn_verfahrensgebiet.functions.knickpunkt_pruefung import ( + knickpunkt_pruefung, + _extrahiere_knickpunkte, +) # Feldname für das Verfahrenskennzeichen im enriched Layer FELD_VKZ = "verfahrensnummer" @@ -417,51 +426,96 @@ def verfahrensgebiet_alkis_komplett( return enriched_layer, flst_layer, alkis_vg_layer, vg_status # ------------------------------------------------------------------ - # Flächenabweichung → Nutzerabfrage für Detailprüfung + # Flächenabweichung → 3-Optionen-Dialog # ------------------------------------------------------------------ progress.close() diff = abs(bror_flaeche - vg_flaeche) - starten = ask_yes_no( + entscheidung = ask_detailpruefung_oder_vg_laden( + tab_widget, "Abweichung erkannt", f"Abweichung zwischen BauRaumOrdnungsrecht-Objekt und abgeleitetem " - f"Verfahrensgebiet erkannt. Detailprüfung starten?\n\n" + f"Verfahrensgebiet erkannt.\n\n" f"BauRaumOrdnungsrecht: {bror_flaeche:.2f} m²\n" f"Verfahrensgebiet: {vg_flaeche:.2f} m²\n" - f"Differenz: {diff:.2f} m²", - default=True, - parent=tab_widget, + f"Differenz: {diff:.2f} m²\n\n" + f"Wie möchten Sie fortfahren?", ) - if not starten: + if entscheidung == "abbrechen": abgeschlossen = True return enriched_layer, flst_layer, alkis_vg_layer, vg_status # ------------------------------------------------------------------ - # Umringsprüfung (Variante 1: linienbasierter Umringvergleich) + # Option: Bodenordnungsobjekt direkt als Verfahrensgebiet laden # ------------------------------------------------------------------ - umring_progress = create_progress_dialog( - 3, "Detailprüfung (Umringe)", "Starte Umringsprüfung …" + if entscheidung == "als_vg_laden": + # Layer bleibt im Projekt – nur umbenennen. working_tab überspringt + # addMapLayer wenn der Layer bereits registriert ist. + alkis_vg_layer.setName(VERF_LAYER_NAME) + bror_was_added_by_us = False + abgeschlossen = True + return alkis_vg_layer, None, None, vg_status + + # ------------------------------------------------------------------ + # Option: Detailprüfung (knickpunktbasiert) + # ------------------------------------------------------------------ + # Knickpunkte vorab extrahieren, um Schrittanzahl zu bestimmen + knickpunkte = _extrahiere_knickpunkte(alkis_vg_layer) + n_knick = len(knickpunkte) + # Gesamtschritte: 0=Start, 1=PTA-Index, 2…n+1=Knickpunkte, n+2=Orange, n+3=Fertig + # + 1 Schritt vorab für Grenzpunkte laden + gesamt_schritte = n_knick + 5 + + detail_progress = create_progress_dialog( + gesamt_schritte, + "Detailprüfung (Knickpunkte)", + "Schritt 1: Grenzpunkte laden …", + ) + detail_progress.set_value(0) + + # Schritt 1: PunktortTA gemarkungsweise laden (Hits + 5000er-Batches) + punktort_layer = lade_punktort_ta( + vorgabe_schluessel, vg_geom, progress=detail_progress ) - ergebnis_layer, konflikt_anzahl = umringspruefung( - alkis_vg_layer, flst_layer, progress=umring_progress - ) - - if umring_progress.is_canceled(): - umring_progress.close() + if detail_progress.is_canceled(): + detail_progress.close() abgeschlossen = True return enriched_layer, flst_layer, alkis_vg_layer, vg_status - umring_progress.set_value(3) - umring_progress.close() + detail_progress.set_value(1) + + if punktort_layer is None: + detail_progress.close() + abgeschlossen = True + return enriched_layer, flst_layer, alkis_vg_layer, vg_status + + # Schritte 2…Ende: Knickpunktprüfung + ergebnis_layer, identisch, kein_pta, fehlender_knick = knickpunkt_pruefung( + alkis_vg_layer, + punktort_layer, + flst_layer, + progress=detail_progress, + knickpunkte=knickpunkte, + ) + + if detail_progress.is_canceled(): + detail_progress.close() + abgeschlossen = True + return enriched_layer, flst_layer, alkis_vg_layer, vg_status + + detail_progress.set_value(gesamt_schritte) + detail_progress.close() if ergebnis_layer and ergebnis_layer.isValid(): project.addMapLayer(ergebnis_layer) show_info_dialog( - "Detailprüfung", - f"Detailprüfung abgeschlossen. " - f"Es wurden {konflikt_anzahl} Konfliktstellen identifiziert.", + "Detailprüfung abgeschlossen", + f"Knickpunkte geprüft: {n_knick}\n" + f" Identisch (grün): {identisch}\n" + f" Kein PunktortTA (rot): {kein_pta}\n" + f" Fehlende Knickpunkte (orange): {fehlender_knick}", parent=tab_widget, ) diff --git a/tests/test_knickpunkt_pruefung.py b/tests/test_knickpunkt_pruefung.py new file mode 100644 index 0000000..882208a --- /dev/null +++ b/tests/test_knickpunkt_pruefung.py @@ -0,0 +1,307 @@ +"""sn_verfahrensgebiet/tests/test_knickpunkt_pruefung.py + +Unit-Tests für :mod:`sn_verfahrensgebiet.functions.knickpunkt_pruefung`. + +Getestete Szenarien: + +_extrahiere_knickpunkte: +- Leerer Layer → leere Liste +- Polygon mit 4 Knickpunkten → korrekte Anzahl (ohne doppelten Schlusspunkt) +- Keine Geometrie → leere Liste + +knickpunkt_pruefung: +- Leere Knickpunktliste → (None, 0, 0, 0) +- Kein PunktortTA-Layer → (None, 0, 0, 0) +- Identischer Punkt → grün gezählt +- Kein passender PunktortTA → rot gezählt +- PunktortTA nahe Umring auf Flurstücksgrenze → orange gezählt +- Abbruch via progress.is_canceled() → (None, 0, 0, 0) +""" + +import sys +import types +import unittest +from unittest.mock import MagicMock, patch + +# --------------------------------------------------------------------------- +# Minimales QGIS-Mock-Setup (identisch mit test_verfahrensgebiet_alkis_komplett) +# --------------------------------------------------------------------------- +if "qgis" not in sys.modules: + qgis_mod = types.ModuleType("qgis") + qgis_core = types.ModuleType("qgis.core") + # Attribute vorab setzen, damit patch() sie überschreiben kann + qgis_core.QgsPointXY = None + qgis_core.QgsRectangle = None + qgis_core.QgsSpatialIndex = None + qgis_core.QgsMarkerSymbol = None + qgis_core.QgsCategorizedSymbolRenderer = None + qgis_core.QgsRendererCategory = None + qgis_utils = types.ModuleType("qgis.utils") + qgis_utils.iface = types.SimpleNamespace() + qgis_utils.plugins = {} + qgis_mod.utils = qgis_utils + qgis_mod.core = qgis_core + sys.modules["qgis"] = qgis_mod + sys.modules["qgis.core"] = qgis_core + sys.modules["qgis.utils"] = qgis_utils + +from sn_verfahrensgebiet.functions.knickpunkt_pruefung import ( + knickpunkt_pruefung, + _extrahiere_knickpunkte, + WERT_IDENTISCH, + WERT_KEIN_PTA, + WERT_FEHLENDER_KNICK, +) + +_MOD = "sn_verfahrensgebiet.functions.knickpunkt_pruefung" + + +# --------------------------------------------------------------------------- +# Hilfsfunktionen +# --------------------------------------------------------------------------- + +def _make_layer(valid=True, crs="EPSG:25833", lid="L1"): + layer = MagicMock() + layer.isValid.return_value = valid + layer.id.return_value = lid + mock_crs = MagicMock() + mock_crs.authid.return_value = crs + layer.crs.return_value = mock_crs + layer.getFeatures.return_value = [] + return layer + + +def _make_progress(canceled_after: int = 9999): + """Progress-Mock: gibt is_canceled() nach N Aufrufen True zurück.""" + call_count = [0] + + def _is_canceled(): + call_count[0] += 1 + return call_count[0] > canceled_after + + prog = MagicMock() + prog.is_canceled.side_effect = _is_canceled + return prog + + +# --------------------------------------------------------------------------- +# Tests: _extrahiere_knickpunkte +# --------------------------------------------------------------------------- + +class TestExtrahiereKnickpunkte(unittest.TestCase): + + def test_leerer_layer_gibt_leere_liste(self): + layer = _make_layer() + layer.getFeatures.return_value = [] + result = _extrahiere_knickpunkte(layer) + self.assertEqual(result, []) + + def test_keine_gueltige_geometrie_gibt_leere_liste(self): + feat = MagicMock() + feat.geometry.return_value = None + layer = _make_layer() + layer.getFeatures.return_value = [feat] + result = _extrahiere_knickpunkte(layer) + self.assertEqual(result, []) + + def test_feature_mit_leerer_geometrie_gibt_leere_liste(self): + feat = MagicMock() + geom = MagicMock() + geom.isEmpty.return_value = True + feat.geometry.return_value = geom + layer = _make_layer() + layer.getFeatures.return_value = [feat] + result = _extrahiere_knickpunkte(layer) + self.assertEqual(result, []) + + def test_qgis_nicht_verfuegbar_gibt_leere_liste(self): + """Wenn qgis.core kein QgsPointXY hat (ImportError in Funktion), leere Liste.""" + layer = _make_layer() + feat = MagicMock() + geom = MagicMock() + geom.isEmpty.return_value = False + feat.geometry.return_value = geom + layer.getFeatures.return_value = [feat] + + # QgsGeometry.unaryUnion wirft eine Exception → leere Liste erwartet + with patch(f"{_MOD}.QgsGeometry") as mock_geom_cls: + mock_geom_cls.unaryUnion.side_effect = ImportError("kein QGIS") + result = _extrahiere_knickpunkte(layer) + self.assertEqual(result, []) + + +# --------------------------------------------------------------------------- +# Tests: knickpunkt_pruefung – Eingabevalidierung +# --------------------------------------------------------------------------- + +class TestKnickpunktPruefungInputvalidierung(unittest.TestCase): + + def test_keine_knickpunkte_gibt_none_zurueck(self): + bror = _make_layer() + bror.getFeatures.return_value = [] + result = knickpunkt_pruefung(bror, MagicMock(), None) + layer, identisch, kein_pta, fehlend = result + self.assertIsNone(layer) + self.assertEqual(identisch, 0) + self.assertEqual(kein_pta, 0) + self.assertEqual(fehlend, 0) + + def test_kein_punktort_layer_gibt_none_zurueck(self): + bror = _make_layer() + result = knickpunkt_pruefung(bror, None, None, knickpunkte=[MagicMock()]) + layer, identisch, kein_pta, fehlend = result + self.assertIsNone(layer) + + def test_abbruch_via_progress_gibt_none_zurueck(self): + bror = _make_layer() + pta = _make_layer() + prog = _make_progress(canceled_after=0) # sofort abbrechen + result = knickpunkt_pruefung( + bror, pta, None, + progress=prog, + knickpunkte=[MagicMock()], + ) + self.assertIsNone(result[0]) + + +# --------------------------------------------------------------------------- +# Tests: knickpunkt_pruefung – Klassifizierungslogik (mit QGIS-Mocks) +# --------------------------------------------------------------------------- + +class TestKnickpunktPruefungKlassifizierung(unittest.TestCase): + """Testet die Grün/Rot/Orange-Klassifizierung mit gesteuertem QGIS-Mock.""" + + def _make_pta_layer_mit_punkt(self, x: float, y: float, fid: int = 1): + """Erstellt einen PunktortTA-Layer-Mock mit einem Punkt bei (x, y).""" + layer = _make_layer(lid=f"PTA_{fid}") + geom = MagicMock() + geom.isEmpty.return_value = False + pt = MagicMock() + pt.x.return_value = x + pt.y.return_value = y + geom.asPoint.return_value = pt + feat = MagicMock() + feat.id.return_value = fid + feat.geometry.return_value = geom + layer.getFeatures.return_value = [feat] + return layer + + def _make_bror_layer(self, x: float = 0.0, y: float = 0.0): + """BROR-Layer mit einer minimalen Polygon-Geometrie.""" + layer = _make_layer() + geom = MagicMock() + geom.isEmpty.return_value = False + geom.area.return_value = 10000.0 + feat = MagicMock() + feat.geometry.return_value = geom + layer.getFeatures.return_value = [feat] + return layer + + @patch(f"{_MOD}.QgsGeometry") + @patch(f"{_MOD}._baue_pta_index") + @patch(f"{_MOD}._erstelle_ergebnis_layer") + @patch(f"qgis.core.QgsRectangle", create=True) + @patch(f"qgis.core.QgsPointXY", create=True) + def test_identischer_punkt_wird_gruen_gezaehlt( + self, mock_ptxy, mock_rect, mock_erstelle, mock_index, mock_geom_cls + ): + """PunktortTA exakt am Knickpunkt → identisch_count == 1.""" + # Knickpunkt bei (100, 200) + knick = MagicMock() + knick.x.return_value = 100.0 + knick.y.return_value = 200.0 + + # PunktortTA ebenfalls bei (100, 200) – exakter Treffer + pta_geom = MagicMock() + pta_geom.isEmpty.return_value = False + pta_pt = MagicMock() + pta_pt.x.return_value = 100.0 + pta_pt.y.return_value = 200.0 + pta_geom.asPoint.return_value = pta_pt + + # Spatial-Index gibt Feature-ID 1 zurück + mock_idx = MagicMock() + mock_idx.intersects.return_value = [1] + mock_index.return_value = (mock_idx, {1: pta_geom}) + + # Ergebnis-Layer + pr = MagicMock() + pr.addFeatures.return_value = True + ergebnis = MagicMock() + ergebnis.dataProvider.return_value = pr + ergebnis.isValid.return_value = True + mock_erstelle.return_value = ergebnis + + bror = self._make_bror_layer() + pta = self._make_pta_layer_mit_punkt(100.0, 200.0) + + result = knickpunkt_pruefung( + bror, pta, None, knickpunkte=[knick] + ) + _, identisch, kein_pta, _ = result + self.assertEqual(identisch, 1) + self.assertEqual(kein_pta, 0) + + @patch(f"{_MOD}.QgsGeometry") + @patch(f"{_MOD}._baue_pta_index") + @patch(f"{_MOD}._erstelle_ergebnis_layer") + @patch(f"qgis.core.QgsRectangle", create=True) + @patch(f"qgis.core.QgsPointXY", create=True) + def test_kein_pta_an_knickpunkt_wird_rot_gezaehlt( + self, mock_ptxy, mock_rect, mock_erstelle, mock_index, mock_geom_cls + ): + """Kein PunktortTA am Knickpunkt → kein_pta_count == 1.""" + knick = MagicMock() + knick.x.return_value = 100.0 + knick.y.return_value = 200.0 + + # Spatial-Index gibt leere Liste zurück → kein Treffer + mock_idx = MagicMock() + mock_idx.intersects.return_value = [] + mock_index.return_value = (mock_idx, {}) + + pr = MagicMock() + pr.addFeatures.return_value = True + ergebnis = MagicMock() + ergebnis.dataProvider.return_value = pr + ergebnis.isValid.return_value = True + mock_erstelle.return_value = ergebnis + + bror = self._make_bror_layer() + pta = self._make_pta_layer_mit_punkt(999.0, 999.0) # weit weg + + result = knickpunkt_pruefung( + bror, pta, None, knickpunkte=[knick] + ) + _, identisch, kein_pta, _ = result + self.assertEqual(identisch, 0) + self.assertEqual(kein_pta, 1) + + +# --------------------------------------------------------------------------- +# Tests: _baue_flst_vertex_set +# --------------------------------------------------------------------------- + +class TestBaueFlstVertexSet(unittest.TestCase): + + def test_none_layer_gibt_leeres_set(self): + from sn_verfahrensgebiet.functions.knickpunkt_pruefung import _baue_flst_vertex_set + result = _baue_flst_vertex_set(None, MagicMock()) + self.assertEqual(result, set()) + + def test_layer_ohne_schnitt_mit_puffer_ignoriert(self): + from sn_verfahrensgebiet.functions.knickpunkt_pruefung import _baue_flst_vertex_set + layer = _make_layer() + geom = MagicMock() + geom.isEmpty.return_value = False + geom.intersects.return_value = False # kein Schnitt + feat = MagicMock() + feat.geometry.return_value = geom + layer.getFeatures.return_value = [feat] + buffer_geom = MagicMock() + result = _baue_flst_vertex_set(layer, buffer_geom) + self.assertEqual(result, set()) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_verfahrensgebiet_alkis_komplett.py b/tests/test_verfahrensgebiet_alkis_komplett.py index e19d134..3048b43 100644 --- a/tests/test_verfahrensgebiet_alkis_komplett.py +++ b/tests/test_verfahrensgebiet_alkis_komplett.py @@ -17,9 +17,10 @@ Abbruchpfade: - Progress-Abbruch nach Schritt 1 → sofortiger sauberer Rückgabestatus Flächenvergleich: -- Übereinstimmung → show_info_dialog, keine Umringsprüfung -- Abweichung + Nein → kein Umringsprüfungs-Aufruf -- Abweichung + Ja → umringspruefung aufgerufen, Ergebnislayer ins Projekt +- Übereinstimmung → show_info_dialog, keine Detailprüfung +- Abweichung + abbrechen → kein Detailprüfungs-Aufruf +- Abweichung + als_vg_laden → BROR umbenannt, als enriched_layer zurückgegeben +- Abweichung + detailpruefung → lade_punktort_ta + knickpunkt_pruefung aufgerufen Cleanup: - BROR-Layer wird bei Erfolg NICHT entfernt @@ -141,8 +142,10 @@ _PATCH_GET_VAR = f"{_MOD}.get_variable" _PATCH_PROGRESS = f"{_MOD}.create_progress_dialog" _PATCH_VERGLEICHE = f"{_MOD}._vergleiche_flaechen" _PATCH_SHOW_INFO = f"{_MOD}.show_info_dialog" -_PATCH_ASK_YES_NO = f"{_MOD}.ask_yes_no" -_PATCH_UMRING = f"{_MOD}.umringspruefung" +_PATCH_ASK_DIALOG = f"{_MOD}.ask_detailpruefung_oder_vg_laden" +_PATCH_GP_ALKIS = f"{_MOD}.lade_punktort_ta" +_PATCH_KNICK = f"{_MOD}.knickpunkt_pruefung" +_PATCH_EXTRAHIERE = f"{_MOD}._extrahiere_knickpunkte" # --------------------------------------------------------------------------- @@ -448,12 +451,12 @@ class TestVerfahrensgebietAlkisKomplett(unittest.TestCase): mock_vkz.assert_called_once_with(enriched, "") # ------------------------------------------------------------------ # - # Flächenvergleich – Abweichung + Nein + # Flächenvergleich – Abweichung + Abbrechen # ------------------------------------------------------------------ # @patch(_PATCH_PROGRESS) - @patch(_PATCH_UMRING) - @patch(_PATCH_ASK_YES_NO) + @patch(_PATCH_KNICK) + @patch(_PATCH_ASK_DIALOG) @patch(_PATCH_VERGLEICHE) @patch(_PATCH_VKZ) @patch(_PATCH_GET_VAR) @@ -463,11 +466,11 @@ class TestVerfahrensgebietAlkisKomplett(unittest.TestCase): @patch(_PATCH_GEM) @patch(_PATCH_FLST) @patch(_PATCH_VG_ALKIS) - def test_flaechen_abweichung_nein_kein_umring( + def test_flaechen_abweichung_abbrechen_kein_detailcheck( self, mock_vg, mock_flst, mock_gem, mock_proj, mock_select, - mock_erstelle, mock_get_var, mock_vkz, mock_vergl, mock_ask, mock_umring, mock_prog, + mock_erstelle, mock_get_var, mock_vkz, mock_vergl, mock_ask, mock_knick, mock_prog, ): - """Abweichung + Nein → umringspruefung wird NICHT aufgerufen.""" + """Abweichung + Abbrechen → knickpunkt_pruefung wird NICHT aufgerufen.""" mock_proj.instance.return_value = _DummyProject() vg_layer = _make_layer(name="BauRaumOrdnungsrecht (ALKIS)") flst_layer = _make_layer(name="Flurstücke") @@ -480,23 +483,22 @@ class TestVerfahrensgebietAlkisKomplett(unittest.TestCase): mock_erstelle.return_value = (enriched, 3, 0) mock_get_var.return_value = "" mock_vergl.return_value = (1000.0, 800.0, False) - mock_ask.return_value = False + mock_ask.return_value = "abbrechen" mock_prog.return_value = _make_progress_mock() result_enriched, _, _, status = verfahrensgebiet_alkis_komplett(None) - mock_umring.assert_not_called() + mock_knick.assert_not_called() self.assertIs(result_enriched, enriched) self.assertEqual(status, LoadStatus.FIRST) # ------------------------------------------------------------------ # - # Flächenvergleich – Abweichung + Ja → Umringsprüfung + # Flächenvergleich – Abweichung + Als VG laden # ------------------------------------------------------------------ # @patch(_PATCH_PROGRESS) - @patch(_PATCH_SHOW_INFO) - @patch(_PATCH_UMRING) - @patch(_PATCH_ASK_YES_NO) + @patch(_PATCH_KNICK) + @patch(_PATCH_ASK_DIALOG) @patch(_PATCH_VERGLEICHE) @patch(_PATCH_VKZ) @patch(_PATCH_GET_VAR) @@ -506,20 +508,18 @@ class TestVerfahrensgebietAlkisKomplett(unittest.TestCase): @patch(_PATCH_GEM) @patch(_PATCH_FLST) @patch(_PATCH_VG_ALKIS) - def test_flaechen_abweichung_ja_startet_umringspruefung( + def test_flaechen_abweichung_als_vg_laden( self, mock_vg, mock_flst, mock_gem, mock_proj, mock_select, - mock_erstelle, mock_get_var, mock_vkz, mock_vergl, mock_ask, - mock_umring, mock_info, mock_prog, + mock_erstelle, mock_get_var, mock_vkz, mock_vergl, mock_ask, mock_knick, mock_prog, ): - """Abweichung + Ja → umringspruefung aufgerufen, Ergebnislayer ins Projekt, - Abschlussmeldung mit Konfliktanzahl.""" + """Abweichung + Als VG laden → BROR-Layer wird umbenannt und als + enriched_layer zurückgegeben; flst_layer und bror_layer sind None.""" proj_inst = _DummyProject() mock_proj.instance.return_value = proj_inst - vg_layer = _make_layer(name="BauRaumOrdnungsrecht (ALKIS)") - flst_layer = _make_layer(name="Flurstücke") - enriched = _make_layer(name="Verfahrensgebiet", valid=True) - ergebnis_layer = _make_layer(name="Umringsprüfung", lid="UMRING_L") + vg_layer = _make_layer(name="BauRaumOrdnungsrecht (ALKIS)", lid="BROR_VG") + flst_layer = _make_layer(name="Flurstücke") + enriched = _make_layer(name="Verfahrensgebiet", valid=True) mock_vg.return_value = (vg_layer, LoadStatus.FIRST, []) mock_gem.return_value = [] @@ -528,17 +528,79 @@ class TestVerfahrensgebietAlkisKomplett(unittest.TestCase): mock_erstelle.return_value = (enriched, 3, 0) mock_get_var.return_value = "" mock_vergl.return_value = (1000.0, 800.0, False) - mock_ask.return_value = True - mock_umring.return_value = (ergebnis_layer, 4) + mock_ask.return_value = "als_vg_laden" mock_prog.return_value = _make_progress_mock() + result_enriched, result_flst, result_bror, status = verfahrensgebiet_alkis_komplett(None) + + mock_knick.assert_not_called() + # BROR-Layer wird als enriched_layer zurückgegeben + self.assertIs(result_enriched, vg_layer) + self.assertIsNone(result_flst) + self.assertIsNone(result_bror) + self.assertEqual(status, LoadStatus.FIRST) + # Name wurde auf VERF_LAYER_NAME geändert + vg_layer.setName.assert_called() + + # ------------------------------------------------------------------ # + # Flächenvergleich – Abweichung + Detailprüfung + # ------------------------------------------------------------------ # + + @patch(_PATCH_PROGRESS) + @patch(_PATCH_SHOW_INFO) + @patch(_PATCH_EXTRAHIERE) + @patch(_PATCH_KNICK) + @patch(_PATCH_GP_ALKIS) + @patch(_PATCH_ASK_DIALOG) + @patch(_PATCH_VERGLEICHE) + @patch(_PATCH_VKZ) + @patch(_PATCH_GET_VAR) + @patch(_PATCH_ERSTELLE) + @patch(_PATCH_SELECT) + @patch(_PATCH_PROJECT) + @patch(_PATCH_GEM) + @patch(_PATCH_FLST) + @patch(_PATCH_VG_ALKIS) + def test_flaechen_abweichung_detailpruefung_startet_knickpruefung( + self, mock_vg, mock_flst, mock_gem, mock_proj, mock_select, + mock_erstelle, mock_get_var, mock_vkz, mock_vergl, mock_ask, + mock_gp, mock_knick, mock_extrahiere, mock_info, mock_prog, + ): + """Abweichung + Detailprüfung → lade_punktort_ta + knickpunkt_pruefung + aufgerufen, Ergebnislayer ins Projekt, Abschlussmeldung mit Zählung.""" + proj_inst = _DummyProject() + mock_proj.instance.return_value = proj_inst + + vg_layer = _make_layer(name="BauRaumOrdnungsrecht (ALKIS)") + flst_layer = _make_layer(name="Flurstücke") + enriched = _make_layer(name="Verfahrensgebiet", valid=True) + punktort_layer = _make_layer(name="PunktortTA", lid="PTA_L") + ergebnis_layer = _make_layer(name="Knickpunktprüfung", lid="KNICK_L") + + mock_vg.return_value = (vg_layer, LoadStatus.FIRST, []) + mock_gem.return_value = [] + mock_flst.return_value = (flst_layer, LoadStatus.FIRST) + mock_select.return_value = 3 + mock_erstelle.return_value = (enriched, 3, 0) + mock_get_var.return_value = "" + mock_vergl.return_value = (1000.0, 800.0, False) + mock_ask.return_value = "detailpruefung" + mock_extrahiere.return_value = [MagicMock(), MagicMock(), MagicMock()] # 3 Knickpunkte + mock_gp.return_value = punktort_layer + mock_knick.return_value = (ergebnis_layer, 2, 1, 0) + mock_prog.return_value = _make_progress_mock() + result_enriched, _, _, status = verfahrensgebiet_alkis_komplett(None) - mock_umring.assert_called_once() + mock_gp.assert_called_once() + mock_knick.assert_called_once() self.assertIn(ergebnis_layer, proj_inst.added) mock_info.assert_called_once() - self.assertIn("4", mock_info.call_args[0][1]) - self.assertIn("Konfliktstellen", mock_info.call_args[0][1]) + # Abschlussmeldung muss Zahlen enthalten + info_text = mock_info.call_args[0][1] + self.assertIn("3", info_text) # Anzahl Knickpunkte + self.assertIn("2", info_text) # identisch + self.assertIn("1", info_text) # kein_pta self.assertEqual(status, LoadStatus.FIRST) # ------------------------------------------------------------------ # diff --git a/ui/tabs/working_tab.py b/ui/tabs/working_tab.py index 645f38f..46332c9 100644 --- a/ui/tabs/working_tab.py +++ b/ui/tabs/working_tab.py @@ -8,7 +8,7 @@ from sn_basis.functions import Qgis, QgsProject #QgsMessageLog from sn_basis.functions.qgisui_wrapper import QgsMapLayerComboBox from sn_basis.functions.qgiscore_wrapper import QgsMapLayerProxyModel from sn_basis.functions.verfahrensgebiet_manager import get_verfahrensgebiet_layer_id, set_verfahrensgebiet_layer_id -from sn_basis.functions.dialog_wrapper import show_info_dialog +from sn_basis.functions.dialog_wrapper import show_info_dialog, create_progress_dialog from sn_basis.modules.Pruefmanager import Pruefmanager from sn_basis.modules.pruef_ergebnis import pruef_ergebnis from qgis.utils import iface @@ -16,9 +16,9 @@ from qgis.utils import iface from sn_verfahrensgebiet.functions.verfahrensgebiet_alkis import LoadStatus from sn_verfahrensgebiet.functions.flurstueck_alkis import flurstueck_alkis from sn_verfahrensgebiet.functions.verfahrensgebiet_alkis_komplett import verfahrensgebiet_alkis_komplett -# TODO Grenzpunkt-Pipeline: Import reaktivieren, sobald grenzpunkt_alkis.py -# mit der neuen Pipeline (PunktortTA → istTeilVon → Einzel-URN) umgeschrieben wurde. -# from sn_verfahrensgebiet.functions.grenzpunkt_alkis import grenzpunkt_alkis +from sn_verfahrensgebiet.functions.grenzpunkt_alkis import grenzpunkte_gemarkungsweise, JOINED_LAYER_NAME +from sn_verfahrensgebiet.functions.gemarkung_alkis import gemarkung_schluessel_und_namen_von_vg_geometrie +from sn_verfahrensgebiet.functions.umringstatistik_grenzpunkte import umringstatistik_grenzpunkte from sn_verfahrensgebiet.functions.verfahrensgebiet_edit_service import ( pruefe_flurstueck_layer, pruefe_auswahl, @@ -105,6 +105,11 @@ class WorkingTab(QWidget): row_grenz_buttons.addWidget(self.btn_grenz_alkis) layout.addLayout(row_grenz_buttons) + row_umring_buttons = QHBoxLayout() + self.btn_umring_statistik = QPushButton("Umringstatistik Grenzpunkte") + row_umring_buttons.addWidget(self.btn_umring_statistik) + layout.addLayout(row_umring_buttons) + # Abschnitt 3: Verfahrensgebiet erstellen row_verf_erstellen_header = QHBoxLayout() lbl_verf_erstellen = QLabel("Verfahrensgebiet erstellen") @@ -129,6 +134,7 @@ class WorkingTab(QWidget): self.btn_verf_alkis.clicked.connect(self.handle_verf_alkis) self.btn_flurst_alkis.clicked.connect(self.handle_flurst_alkis) self.btn_grenz_alkis.clicked.connect(self.handle_grenz_alkis) + self.btn_umring_statistik.clicked.connect(self.handle_umringstatistik) self.btn_verf_aus_auswahl.clicked.connect(self.handle_verf_aus_auswahl) self.btn_verf_aus_liste.clicked.connect(self.handle_verf_aus_liste) # self.btn_verf_shape.clicked.connect(self.handle_verf_shape) @@ -167,8 +173,9 @@ class WorkingTab(QWidget): if bror_layer and bror_layer.isValid(): _setze_bror_stil(bror_layer) - # Enriched VG-Layer ins Projekt - project.addMapLayer(enriched_layer) + # Enriched VG-Layer ins Projekt (ggf. bereits registriert, z.B. bei "als VG laden") + if not project.mapLayer(enriched_layer.id()): + project.addMapLayer(enriched_layer) apply_style(enriched_layer, "verfahrensgebiet.qml") iface.mapCanvas().setExtent(enriched_layer.extent()) iface.mapCanvas().refresh() @@ -405,36 +412,126 @@ class WorkingTab(QWidget): # ------------------------------------------------------------------ def handle_grenz_alkis(self) -> None: - """Lädt Grenzpunkte (AX_PunktortTA + AX_Grenzpunkt) aus ALKIS. + """Lädt Grenzpunkte gemarkungsweise aus ALKIS. - TODO Grenzpunkt-Pipeline: Implementierung ausstehend. - Neue Pipeline: VG-Geometrie → PunktortTA (spatial filter) → - istTeilVon-Attribute → Grenzpunkt per Einzel-URN laden. - Bis dahin ist dieser Handler deaktiviert. + Verwendet denselben Gemarkungsfilter wie "Verfahrensgebiet aus ALKIS laden": + Die Gemarkungsschlüssel werden per Spatial-Join aus der Geometrie des + aktuell gewählten Verfahrensgebiet-Layers abgeleitet. """ - # TODO Grenzpunkt-Pipeline: Körper reaktivieren, sobald grenzpunkt_alkis.py - # mit der neuen Pipeline umgeschrieben wurde. - # - # self._set_busy(self.btn_grenz_alkis, True, "Grenzpunkte aus ALKIS laden") - # try: - # punktort_layer, grenzpunkt_layer = grenzpunkt_alkis(self) - # - # if grenzpunkt_layer and grenzpunkt_layer.isValid(): - # iface.mapCanvas().refresh() - # self.setze_haken(self.haken_grenz, True) - # success("Grenzpunkte", "Grenzpunkte erfolgreich geladen.", duration=5) - # elif punktort_layer and punktort_layer.isValid(): - # iface.mapCanvas().refresh() - # self.setze_haken(self.haken_grenz, True) - # info( - # "Grenzpunkte", - # "PunktortTA-Layer geladen. Grenzpunkt-IDs konnten nicht aufgelöst werden.", - # duration=6, - # ) - # finally: - # self._set_busy(self.btn_grenz_alkis, False, "Grenzpunkte aus ALKIS laden") - from sn_basis.functions.message_wrapper import warning as _warn - _warn("Grenzpunkte", "Grenzpunkt-Pipeline ist noch nicht implementiert.", duration=5) + # ------------------------------------------------------------------ # + # Schritt 0: Fortschrittsdialog sofort anzeigen # + # ------------------------------------------------------------------ # + # 4 Vorbereitungsschritte (Geometrie, Gemarkungen, Hits, Laden) – + # die genaue Schrittzahl wird in grenzpunkte_gemarkungsweise verfeinert. + prog = create_progress_dialog( + 4, "Grenzpunkte aus ALKIS laden", "Verfahrensgebiet prüfen …" + ) + prog.set_value(0) + + # Verfahrensgebiet-Layer aus dem Dropdown holen + vg_layer = self.verf_layer_combo.currentLayer() + if not vg_layer or not vg_layer.isValid(): + prog.close() + warning( + "Grenzpunkte", + "Kein Verfahrensgebiet-Layer gewählt. Bitte zuerst ein Verfahrensgebiet laden.", + duration=6, + ) + return + + # Geometrie des ersten Features als Suchgeometrie + feats = list(vg_layer.getFeatures()) + if not feats: + prog.close() + warning("Grenzpunkte", "Verfahrensgebiet-Layer enthält keine Features.", duration=5) + return + + # Geometrie über alle Features vereinen + from qgis.core import QgsGeometry # type: ignore[import] + vg_geom = QgsGeometry.unaryUnion([f.geometry() for f in feats if f.hasGeometry()]) + + if not vg_geom or vg_geom.isEmpty(): + prog.close() + warning("Grenzpunkte", "Verfahrensgebiet-Layer hat keine Geometrie.", duration=5) + return + + prog.set_label("Gemarkungsschlüssel per WFS ermitteln …") + prog.set_value(1) + if prog.is_canceled(): + return + + # Gemarkungsschlüssel per Spatial-Join ermitteln (identisch zu VG-Workflow) + schluessel_namen = gemarkung_schluessel_und_namen_von_vg_geometrie(vg_geom) + schluessel_liste = [nr for nr, _ in schluessel_namen] + + if not schluessel_liste: + prog.close() + warning( + "Grenzpunkte", + "Keine Gemarkungen für das Verfahrensgebiet gefunden. " + "Bitte Netzwerkverbindung prüfen.", + duration=6, + ) + return + + prog.set_label(f"{len(schluessel_liste)} Gemarkung(en) gefunden – Grenzpunkte laden …") + prog.set_value(2) + if prog.is_canceled(): + return + + self._set_busy(self.btn_grenz_alkis, True, "Grenzpunkte aus ALKIS laden") + prog.close() # Vorbereitungs-Dialog schließen; grenzpunkte_gemarkungsweise öffnet eigenen + try: + joined_layer = grenzpunkte_gemarkungsweise(self, schluessel_liste, vg_geom=vg_geom) + + if joined_layer and joined_layer.isValid(): + iface.mapCanvas().refresh() + self.setze_haken(self.haken_grenz, True) + success( + "Grenzpunkte", + f"Grenzpunkte erfolgreich geladen ({joined_layer.featureCount()} Objekte).", + duration=5, + ) + finally: + self._set_busy(self.btn_grenz_alkis, False, "Grenzpunkte aus ALKIS laden") + + def handle_umringstatistik(self) -> None: + """Klassifiziert Grenzpunkte auf dem VG-Umring nach Lagegenauigkeit. + + Sucht den vorhandenen ``Grenzpunkte (ALKIS)``-Layer im Projekt und + kombiniert ihn mit dem gewählten Verfahrensgebiet-Layer, um die + Umringstatistik zu berechnen. + """ + # 1. Grenzpunkte-Layer aus dem Projekt holen + project = QgsProject.instance() + gp_layers = project.mapLayersByName(JOINED_LAYER_NAME) + if not gp_layers: + warning( + "Umringstatistik", + f"Kein Layer '{JOINED_LAYER_NAME}' im Projekt gefunden. " + "Bitte zuerst 'Grenzpunkte aus ALKIS laden' ausführen.", + duration=8, + ) + return + grenzpunkte_layer = gp_layers[0] + + # 2. Verfahrensgebiet-Layer aus Dropdown + vg_layer = self.verf_layer_combo.currentLayer() + if not vg_layer or not vg_layer.isValid(): + warning( + "Umringstatistik", + "Kein Verfahrensgebiet-Layer gewählt. Bitte zuerst ein Verfahrensgebiet laden.", + duration=6, + ) + return + + self._set_busy(self.btn_umring_statistik, True, "Umringstatistik Grenzpunkte") + try: + result_layer = umringstatistik_grenzpunkte(self, grenzpunkte_layer, vg_layer) + if result_layer and result_layer.isValid(): + iface.mapCanvas().refresh() + finally: + self._set_busy(self.btn_umring_statistik, False, "Umringstatistik Grenzpunkte") # ---------------------------------------------------------------------------