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

588 lines
22 KiB
Python
Raw Normal View History

shape Verfahrensgebiet ergänzt
2026-07-15 15:29:48 +02:00
"""sn_verfahrensgebiet/functions/verfahrensgebiet_shape_komplett.py – VG-Lade-Workflow aus Shape (ZIP).
Orchestriert den vollständigen Arbeitsablauf zum Laden eines Verfahrensgebiets
aus einem ZIP-Archiv mit Shapefile-Set:
1. Dateidialog öffnen (.zip-Filter); Nutzerabbruch beendet den Vorgang.
2. ZIP neben der ZIP-Datei entpacken und auf ein gültiges Shapefile-Set prüfen.
3. Shapefile als QGIS-Vektorlayer laden und validieren (Standard-QGIS-Routinen).
4. Alle Features per ``QgsGeometry.unaryUnion`` zu einem Single-Feature-
Memory-Layer „Verfahrensgebiet" verschmelzen.
5. Gemarkungsschlüssel per Geometrie aus ``AX_Gemarkung`` vorschlagen.
6. Flurstücke per Dialog (mit Vorschlag) laden.
7. Flurstücke räumlich auf das Shape-VG einschränken (``selectByLocation``).
8. Temporären enriched Layer aus Auswahl für den Flächenvergleich erzeugen
(wird **nicht** ins Projekt eingefügt und **nicht** zurückgegeben).
9. Flächenvergleich zwischen Shape-VG und temporärem enriched Layer.
- **Übereinstimmung** (≤ Toleranz): modale OK-Meldung, Ablauf beendet.
- **Abweichung**: 2-Optionen-Dialog (Detailprüfung / Abbrechen).
10. Bei Wahl „Detailprüfung" (knickpunktbasiert):
- Grenzpunkte (AX_PunktortTA) per BBox laden.
- Knickpunkte des Shape-VG mit PunktortTA vergleichen (grün/rot/orange).
- Korrigiertes Polygon aus identischen + gefundenen PunktortTA ableiten.
- Ergebnislayer (Punkte), Verbindungslinien und Polygon ins Projekt einfügen.
- Abschlussmeldung mit Zählung.
Das Shape-VG (``Verfahrensgebiet``) ist das finale Rückgabe-Layer. Ein
``enriched Layer aus Flurstücken`` wird nur temporär für den Flächenvergleich
erzeugt, aber weder ins Projekt eingefügt noch zurückgegeben.
**Rückgabe:**
``Tuple[shape_vg_layer, flst_layer, None, LoadStatus]``
**Abbruchverhalten:** Jeder Schritt, der :attr:`LoadStatus.NONE` zurückgibt
oder kein Ergebnis liefert, bricht den Workflow ab.
"""
from __future__ import annotations
import os
import zipfile
from typing import Optional, Tuple
from sn_basis.functions import QgsProject, QgsVectorLayer, QgsFeature
from sn_basis.functions import QMessageBox
from sn_basis.functions.message_wrapper import warning
from sn_basis.functions.variable_wrapper import get_variable
from sn_basis.functions.qt_wrapper import QFileDialog
from sn_basis.functions.qgiscore_wrapper import QgsGeometry
from sn_basis.functions.dialog_wrapper import (
create_progress_dialog,
show_info_dialog,
ask_detailpruefung_oder_abbrechen,
)
from sn_verfahrensgebiet.functions.verfahrensgebiet_alkis import LoadStatus, verfahrensgebiet_alkis
from sn_verfahrensgebiet.functions.gemarkung_alkis import (
gemarkung_schluessel_und_namen_von_vg_geometrie,
)
from sn_verfahrensgebiet.functions.flurstueck_alkis import flurstueck_alkis
from sn_verfahrensgebiet.functions.verfahrensgebiet_edit_service import (
erstelle_verfahrensgebiet_aus_auswahl,
VERF_LAYER_NAME,
)
from sn_verfahrensgebiet.functions.grenzpunkt_alkis import lade_punktort_ta
from sn_verfahrensgebiet.functions.knickpunkt_pruefung import (
knickpunkt_pruefung,
_extrahiere_knickpunkte,
erstelle_korrigiertes_polygon,
)
from sn_verfahrensgebiet.functions.verfahrensgebiet_alkis_komplett import (
_select_by_location,
_vergleiche_flaechen,
_lese_flaechen_toleranz,
_entferne_hilfs_layer,
MINDEST_FLAECHEN_TOLERANZ_M2,
BROR_LAYER_NAME,
)
# ---------------------------------------------------------------------------
# Interne Hilfsfunktionen
# ---------------------------------------------------------------------------
def _finde_shp(verzeichnis: str) -> Optional[str]:
"""Sucht rekursiv nach der ersten ``.shp``-Datei im Verzeichnis.
Parameters
----------
verzeichnis :
Startverzeichnis für die Suche.
Returns
-------
Optional[str]
Absoluter Pfad zur ersten gefundenen ``.shp``-Datei, oder ``None``.
"""
for root, _dirs, files in os.walk(verzeichnis):
for fname in files:
if fname.lower().endswith(".shp"):
return os.path.join(root, fname)
return None
def _erstelle_memory_layer_aus_shp(shp_layer) -> Optional[object]:
"""Vereint alle Features des Shapefile-Layers zu einem Single-Feature-
Memory-Layer „Verfahrensgebiet" via ``QgsGeometry.unaryUnion``.
Alle Felder des Shapefiles werden in den Memory-Layer übernommen.
Die Attributwerte des ersten Features mit gültiger Geometrie werden
auf das Union-Feature gesetzt.
Parameters
----------
shp_layer :
Geladener ``QgsVectorLayer`` aus dem Shapefile.
Returns
-------
Optional[QgsVectorLayer]
Single-Feature-Memory-Layer mit Union-Geometrie und Shapefile-Attributen
oder ``None`` bei Fehler.
"""
geom_list = []
first_feat = None
for feat in shp_layer.getFeatures():
try:
geom = feat.geometry()
if geom and not geom.isEmpty():
geom_list.append(geom)
if first_feat is None:
first_feat = feat
except Exception:
pass
if not geom_list:
return None
union_geom = QgsGeometry.unaryUnion(geom_list)
if not union_geom or union_geom.isEmpty():
return None
crs_authid = shp_layer.crs().authid()
mem_layer = QgsVectorLayer(
f"MultiPolygon?crs={crs_authid}", VERF_LAYER_NAME, "memory"
)
pr = mem_layer.dataProvider()
# Felder des Shapefiles übernehmen
try:
pr.addAttributes(list(shp_layer.fields()))
mem_layer.updateFields()
except Exception:
pass
feat_out = QgsFeature(mem_layer.fields())
feat_out.setGeometry(union_geom)
# Attributwerte vom ersten Feature übernehmen
if first_feat is not None:
try:
feat_out.setAttributes(first_feat.attributes())
except Exception:
pass
pr.addFeatures([feat_out])
mem_layer.updateExtents()
if not mem_layer.isValid() or mem_layer.featureCount() == 0:
return None
return mem_layer
# ---------------------------------------------------------------------------
# Öffentliche Lade-Hilfsfunktion
# ---------------------------------------------------------------------------
def lade_vg_aus_zip(parent_widget) -> Optional[object]:
"""Öffnet Dateidialog, entpackt ZIP und gibt Union-Memory-Layer zurück.
Schrittfolge:
1. ``QFileDialog.getOpenFileName`` mit Filter ``ZIP-Dateien (*.zip)``.
2. ZIP neben der Quelldatei in ein Unterverzeichnis entpacken.
3. Erste ``.shp``-Datei im Verzeichnis finden.
4. Als ``QgsVectorLayer`` laden und ``isValid()`` prüfen.
5. Alle Features per ``QgsGeometry.unaryUnion`` verschmelzen.
6. Single-Feature-Memory-Layer „Verfahrensgebiet" zurückgeben.
Parameters
----------
parent_widget :
Eltern-Widget für Dateidialog und Fehlermeldungen.
Returns
-------
Optional[QgsVectorLayer]
Valider Memory-Layer „Verfahrensgebiet" oder ``None`` bei Abbruch/Fehler.
"""
pfad, _ = QFileDialog.getOpenFileName(
parent_widget,
"ZIP-Datei mit Shapefile-Set auswählen",
"",
"ZIP-Dateien (*.zip)",
)
if not pfad:
return None # Nutzerabbruch
zip_dir = os.path.dirname(pfad)
zip_base = os.path.splitext(os.path.basename(pfad))[0]
extract_dir = os.path.join(zip_dir, zip_base)
try:
with zipfile.ZipFile(pfad, "r") as zf:
zf.extractall(extract_dir)
except Exception:
QMessageBox.warning(
parent_widget,
"Fehler",
"Die ZIP-Datei konnte nicht entpackt werden.\n"
"Bitte eine gültige ZIP-Datei auswählen.",
)
return None
shp_path = _finde_shp(extract_dir)
if not shp_path:
QMessageBox.warning(
parent_widget,
"Fehler",
"gültiges Shape-Dateiset angeben",
)
return None
raw_layer = QgsVectorLayer(shp_path, "tmp_shape_vg", "ogr")
if not raw_layer or not raw_layer.isValid():
QMessageBox.warning(
parent_widget,
"Fehler",
"gültiges Shape-Dateiset angeben",
)
return None
mem_layer = _erstelle_memory_layer_aus_shp(raw_layer)
if mem_layer is None:
QMessageBox.warning(
parent_widget,
"Fehler",
"gültiges Shape-Dateiset angeben",
)
return None
return mem_layer
# ---------------------------------------------------------------------------
# Öffentliche API
# ---------------------------------------------------------------------------
def verfahrensgebiet_shape_komplett(
tab_widget,
) -> Tuple[Optional[object], Optional[object], Optional[object], LoadStatus]:
"""Vollständiger VG-Lade-Workflow aus Shape (ZIP): ZIP → BROR → Gemarkungen →
Flurstücke → Verschneidung → Flächenvergleich → opt. Umringsprüfung.
Das aus der ZIP geladene Shape-VG ist das finale Verfahrensgebiet-Layer.
Ist die Projektvariable ``verfahrensnummer`` gesetzt, wird das zugehörige
BROR-Objekt (``AX_BauRaumOderBodenordnungsrecht``) aus ALKIS geladen und
als Vergleichsobjekt im Flächenvergleich verwendet. Andernfalls wird ein
temporärer enriched Layer aus Flurstücken für den Vergleich herangezogen.
Schrittfolge:
1. ``lade_vg_aus_zip(tab_widget)`` – ZIP öffnen, Shapefile validieren,
Union-Memory-Layer „Verfahrensgebiet" erzeugen.
2. ``verfahrensgebiet_alkis(tab_widget, skip_existing_check=True)`` –
BROR-Objekt als Memory-Layer (sofern VKZ gesetzt).
Der BROR wird **vor** dem Shape-VG ins Projekt eingefügt, um den
Namenskonflikt mit ``_check_existing_layer`` zu vermeiden.
3. Shape-VG ins Projekt einfügen; BROR umbenennen zu
``BauRaumOderBodenordnungsrecht (ALKIS)`` und zum Projekt hinzufügen.
4. ``gemarkung_schluessel_und_namen_von_vg_geometrie(geom)`` – Schlüssel.
5. ``flurstueck_alkis(...)`` – Flurstücke laden.
6. ``_select_by_location(...)`` – räumliche Selektion.
7. Flächenvergleich:
- Mit BROR: ``_vergleiche_flaechen(bror_layer, shape_vg_layer, ...)``.
- Ohne BROR: temp. enriched Layer erzeugen und mit Shape-VG vergleichen.
8. Ggf. Detailprüfung (knickpunktbasiert) auf BROR bzw. Shape-VG.
Parameters
----------
tab_widget :
Eltern-Widget für alle Dialoge.
Returns
-------
Tuple[Optional[shape_vg_layer], Optional[flst_layer], Optional[bror_layer], LoadStatus]
- ``shape_vg_layer``: Memory-Layer „Verfahrensgebiet" (aus Shape).
- ``flst_layer``: Flurstücke-Layer (bleibt im Projekt).
- ``bror_layer``: BROR-Layer ``BauRaumOderBodenordnungsrecht (ALKIS)``
(wenn geladen, bleibt im Projekt); ``None`` wenn VKZ nicht gesetzt.
- :class:`LoadStatus`: ``FIRST`` oder ``NONE``.
"""
project = QgsProject.instance()
shape_vg_layer = None
bror_layer = None
shape_was_added_by_us = False
bror_was_added_by_us = False
abgeschlossen = False
progress = create_progress_dialog(
7, "Verfahrensgebiet aus Shape laden", "Initialisierung …"
)
try:
# ------------------------------------------------------------------
# Schritt 1: Shape-VG aus ZIP laden (noch NICHT ins Projekt einfügen)
# ------------------------------------------------------------------
progress.set_label("Schritt 1/6: Verfahrensgebiet aus ZIP laden …")
progress.set_value(0)
if progress.is_canceled():
return None, None, None, LoadStatus.NONE
shape_vg_layer = lade_vg_aus_zip(tab_widget)
progress.raise_to_front()
if not shape_vg_layer:
return None, None, None, LoadStatus.NONE
vg_status = LoadStatus.FIRST
progress.set_value(1)
# ------------------------------------------------------------------
# Schritt 2: BROR aus ALKIS laden (vor shape_vg, um Namenskonflikt
# in _check_existing_layer zu vermeiden)
# ------------------------------------------------------------------
progress.set_label("Schritt 2/6: BROR aus ALKIS laden …")
if progress.is_canceled():
return None, None, None, LoadStatus.NONE
vkz = get_variable("verfahrensnummer") or ""
if str(vkz).strip():
bror_result, bror_status, _ = verfahrensgebiet_alkis(
tab_widget, skip_existing_check=True
)
progress.raise_to_front()
if bror_status != LoadStatus.NONE and bror_result:
bror_result.setName(BROR_LAYER_NAME)
project.addMapLayer(bror_result)
bror_layer = bror_result
bror_was_added_by_us = True
progress.set_value(2)
# Shape-VG erst jetzt ins Projekt einfügen (nach BROR-Ladung)
project.addMapLayer(shape_vg_layer)
shape_was_added_by_us = True
# ------------------------------------------------------------------
# Schritt 3: Gemarkungsschlüssel per Geometrie-Spatial-Join ermitteln
# ------------------------------------------------------------------
progress.set_label("Schritt 3/6: Gemarkungsschlüssel ermitteln …")
if progress.is_canceled():
return None, None, None, LoadStatus.NONE
shape_feats = list(shape_vg_layer.getFeatures())
vg_geom = shape_feats[0].geometry() if shape_feats else None
schluessel_namen = (
gemarkung_schluessel_und_namen_von_vg_geometrie(vg_geom)
if vg_geom else []
)
vorgabe_schluessel = [nr for nr, _ in schluessel_namen]
vorgabe_namen = [nm for _, nm in schluessel_namen if nm]
if not vorgabe_schluessel:
warning(
"VG laden",
"Keine Gemarkungen in der BBox gefunden. Bitte Schlüssel manuell eingeben.",
duration=5,
)
progress.set_value(3)
# ------------------------------------------------------------------
# Schritt 4: Flurstücke laden (mit Nutzer-Dialog, Vorschlag befüllt)
# ------------------------------------------------------------------
progress.set_label("Schritt 4/6: Flurstücke aus ALKIS laden …")
if progress.is_canceled():
return None, None, None, LoadStatus.NONE
flst_layer, flst_status = flurstueck_alkis(
tab_widget,
vorgabe_schluessel=vorgabe_schluessel,
vorgabe_namen=vorgabe_namen if vorgabe_namen else None,
ohne_dialog=True,
)
if flst_status == LoadStatus.NONE or not flst_layer:
return None, None, None, LoadStatus.NONE
progress.set_value(4)
# ------------------------------------------------------------------
# Schritt 5: Räumliche Selektion
# ------------------------------------------------------------------
progress.set_label("Schritt 5/6: Flurstücke räumlich selektieren …")
if progress.is_canceled():
return None, None, None, LoadStatus.NONE
anzahl = _select_by_location(flst_layer, shape_vg_layer)
if anzahl == 0:
QMessageBox.warning(
tab_widget,
"Keine Flurstücke im VG",
"Es wurden keine Flurstücke gefunden, die das Verfahrensgebiet schneiden.\n"
"Bitte Gemarkungsschlüssel und Ausdehnung prüfen.",
)
return None, flst_layer, None, LoadStatus.NONE
progress.set_value(5)
# ------------------------------------------------------------------
# Schritt 6: Flächenvergleich als Gate
# ------------------------------------------------------------------
progress.set_label("Schritt 6/6: Flächenvergleich …")
if progress.is_canceled():
return None, None, None, LoadStatus.NONE
toleranz = _lese_flaechen_toleranz()
if bror_layer and bror_layer.isValid():
# BROR verfügbar → als Referenz-Objekt verwenden
ref_flaeche, shape_flaeche, stimmt_ueberein = _vergleiche_flaechen(
bror_layer, shape_vg_layer, toleranz
)
dialog_ref_label = "BauRaumOderBodenordnungsrecht"
dialog_vg_label = "Shape-Verfahrensgebiet "
pruef_layer = bror_layer
else:
# Kein BROR → temp. enriched Layer aus Flurstücken erzeugen
temp_enriched, _, _ = erstelle_verfahrensgebiet_aus_auswahl(flst_layer)
ref_flaeche, shape_flaeche, stimmt_ueberein = _vergleiche_flaechen(
shape_vg_layer, temp_enriched, toleranz
)
dialog_ref_label = "Shape-Verfahrensgebiet "
dialog_vg_label = "Verfahrensgebiet (Flurstücke)"
pruef_layer = shape_vg_layer
progress.set_value(6)
if stimmt_ueberein:
progress.set_value(7)
progress.close()
show_info_dialog(
"Flächenvergleich",
"Verfahrensgebiet mit erwarteter Geometrie erzeugt",
parent=tab_widget,
)
abgeschlossen = True
return shape_vg_layer, flst_layer, bror_layer, vg_status
# ------------------------------------------------------------------
# Flächenabweichung → 2-Optionen-Dialog
# ------------------------------------------------------------------
progress.close()
diff = abs(ref_flaeche - shape_flaeche)
entscheidung = ask_detailpruefung_oder_abbrechen(
tab_widget,
"Abweichung erkannt",
f"Abweichung erkannt.\n\n"
f"{dialog_ref_label}: {ref_flaeche:.2f} m²\n"
f"{dialog_vg_label}: {shape_flaeche:.2f} m²\n"
f"Differenz: {diff:.2f} m²\n\n"
f"Wie möchten Sie fortfahren?",
)
if entscheidung == "abbrechen":
abgeschlossen = True
return shape_vg_layer, flst_layer, bror_layer, vg_status
# ------------------------------------------------------------------
# Option: Detailprüfung (knickpunktbasiert) auf pruef_layer
# (= BROR wenn vorhanden, sonst Shape-VG)
# ------------------------------------------------------------------
knickpunkte = _extrahiere_knickpunkte(pruef_layer)
n_knick = len(knickpunkte)
gesamt_schritte = n_knick + 5
detail_progress = create_progress_dialog(
gesamt_schritte,
"Detailprüfung (Knickpunkte)",
"Schritt 1: Grenzpunkte laden …",
)
detail_progress.set_value(0)
punktort_layer = lade_punktort_ta(
vorgabe_schluessel, vg_geom, progress=detail_progress
)
if detail_progress.is_canceled():
detail_progress.close()
abgeschlossen = True
return shape_vg_layer, flst_layer, bror_layer, vg_status
detail_progress.set_value(1)
if punktort_layer is None:
detail_progress.close()
abgeschlossen = True
if not detail_progress.is_canceled():
QMessageBox.critical(
tab_widget,
"WFS-Fehler",
"Die Grenzpunktdaten (PunktortTA) konnten nicht geladen werden.\n\n"
"Mögliche Ursache: Netzwerk- oder Proxy-Fehler.\n"
"Bitte prüfen Sie die QGIS-Proxy-Einstellungen "
"(Einstellungen -> Optionen -> Netzwerk) und den Log-Eintrag "
"unter Ansicht -> Bedienfelder -> Protokollnachrichten "
"(Kategorie 'sn_verfahrensgebiet').",
)
return shape_vg_layer, flst_layer, bror_layer, vg_status
ergebnis_layer, identisch, kein_pta, fehlender_knick = knickpunkt_pruefung(
pruef_layer,
punktort_layer,
flst_layer,
progress=detail_progress,
knickpunkte=knickpunkte,
)
if detail_progress.is_canceled():
detail_progress.close()
abgeschlossen = True
return shape_vg_layer, flst_layer, bror_layer, vg_status
detail_progress.set_value(gesamt_schritte)
detail_progress.close()
# ------------------------------------------------------------------
# Korrigiertes Polygon aus identischen + fehlenden Knickpunkten
# (nur für Visualisierung – shape_vg_layer bleibt das finale VG)
# ------------------------------------------------------------------
if ergebnis_layer and ergebnis_layer.isValid():
linien_layer_korr, polygon_layer_korr = erstelle_korrigiertes_polygon(
pruef_layer, ergebnis_layer, pruef_layer.crs().authid()
)
if polygon_layer_korr is not None and polygon_layer_korr.isValid():
if linien_layer_korr is not None and linien_layer_korr.isValid():
project.addMapLayer(linien_layer_korr)
project.addMapLayer(polygon_layer_korr)
if ergebnis_layer and ergebnis_layer.isValid():
project.addMapLayer(ergebnis_layer)
show_info_dialog(
"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,
)
abgeschlossen = True
return shape_vg_layer, flst_layer, bror_layer, vg_status
finally:
_entferne_hilfs_layer(project, "TempVerfahrensgebiet")
# Shape-VG und BROR bei Fehler/Abbruch entfernen
if not abgeschlossen:
if shape_was_added_by_us and shape_vg_layer is not None:
try:
project.removeMapLayer(shape_vg_layer.id())
except Exception:
pass
if bror_was_added_by_us and bror_layer is not None:
try:
project.removeMapLayer(bror_layer.id())
except Exception:
pass
Reference in New Issue Copy Permalink