Files
Plugin_SN_Verfahrensgebiet/functions/verfahrensgebiet_edit_service.py
T

359 lines
12 KiB
Python
Raw Normal View History

2026-04-21 21:50:26 +02:00
"""sn_verfahrensgebiet/functions/verfahrensgebiet_edit_service.py
Service-Modul für Verfahrensgebiet-Editoperationen aus Flurstück-Auswahl.
Der Verfahrensgebiet-Layer ist ein Single-Feature-Memory-Layer: alle
Flurstück-Geometrien werden via ``QgsGeometry.unaryUnion`` zu einem einzigen
(Multi-)Polygon verschmolzen und als ein Feature gespeichert. Dadurch
erscheint das Verfahrensgebiet im Kartenfenster als zusammenhängendes Objekt.
Die Flurstückskennzeichen werden als kommagetrennte Zeichenkette im
Attributfeld ``flurstueckskennzeichen`` des Single-Features gespeichert,
sodass Delta-Operationen (Ergänzen/Entfernen) ohne erneuten Zugriff auf den
Quell-Flurstück-Layer möglich bleiben und die ID-Menge via
``get_kennzeichen_menge()`` jederzeit abrufbar ist.
"""
from typing import Optional, Tuple, List
from sn_basis.functions import QgsVectorLayer, QgsFeature, QgsField
from sn_basis.functions.qgiscore_wrapper import QgsGeometry
from sn_basis.functions.qt_wrapper import QVariant
from sn_basis.modules.layerpruefer import Layerpruefer
from sn_basis.modules.pruef_ergebnis import pruef_ergebnis
VERF_LAYER_NAME = "Verfahrensgebiet"
FELD_KENNZEICHEN = "flurstueckskennzeichen"
FELD_GEMARKUNG = "gemarkung_ax_gemarkung_schluessel_gemarkungsnummer"
PFLICHTFELDER: List[str] = [FELD_KENNZEICHEN, FELD_GEMARKUNG]
# ---------------------------------------------------------------------------
# Vorbedingungsprüfungen
# ---------------------------------------------------------------------------
def pruefe_flurstueck_layer(layer) -> pruef_ergebnis:
"""Prüft, ob ein Layer als Flurstück-Quelle geeignet ist.
Prüfreihenfolge via Layerpruefer:
existiert → Vektorlayer → Polygon-Geometrie → nicht leer → Pflichtfelder vorhanden.
Returns
-------
pruef_ergebnis
ok=True, wenn alle Prüfungen bestanden wurden.
"""
pruefer = Layerpruefer(
layer=layer,
erwarteter_layertyp="vector",
erwarteter_geotyp="Polygon",
erforderliche_felder=PFLICHTFELDER,
)
return pruefer.pruefe()
def pruefe_auswahl(layer) -> pruef_ergebnis:
"""Prüft, ob auf dem Layer mindestens ein Feature ausgewählt ist.
Returns
-------
pruef_ergebnis
ok=True bei mindestens einer Auswahl, sonst ok=False mit aktion ``layer_leer``.
"""
try:
count = int(layer.selectedFeatureCount())
except Exception:
count = 0
if count == 0:
return pruef_ergebnis(
ok=False,
meldung="Es sind keine Flurstücke ausgewählt. Bitte zuerst eine Auswahl treffen.",
aktion="layer_leer",
kontext=None,
)
return pruef_ergebnis(
ok=True,
meldung=f"{count} Flurstücke ausgewählt.",
aktion="ok",
kontext=None,
)
# ---------------------------------------------------------------------------
# Hilfsfunktionen
# ---------------------------------------------------------------------------
def get_kennzeichen_menge(layer) -> set:
"""Gibt die Menge aller ``flurstueckskennzeichen`` aus dem Layer zurück.
Im Single-Feature-Modell sind die Kennzeichen als kommagetrennte
Zeichenkette im einzigen Feature gespeichert.
"""
result = set()
for feat in layer.getFeatures():
val = feat[FELD_KENNZEICHEN]
if val:
for kz in str(val).split(","):
kz = kz.strip()
if kz:
result.add(kz)
return result
def _erstelle_memory_layer(crs_authid: str) -> QgsVectorLayer:
"""Erzeugt einen leeren Multi-Feature-Memory-Layer für das Verfahrensgebiet."""
layer = QgsVectorLayer(
f"MultiPolygon?crs={crs_authid}", VERF_LAYER_NAME, "memory"
)
pr = layer.dataProvider()
pr.addAttributes([
QgsField(FELD_KENNZEICHEN, QVariant.String),
QgsField(FELD_GEMARKUNG, QVariant.String),
])
layer.updateFields()
return layer
# ---------------------------------------------------------------------------
# Hauptoperationen
# ---------------------------------------------------------------------------
def erstelle_verfahrensgebiet_aus_auswahl(
source_layer,
) -> Tuple[Optional[object], int, int]:
"""Erstellt neuen Verfahrensgebiet-Memory-Layer aus der aktuellen Auswahl.
Alle Geometrien werden via ``QgsGeometry.unaryUnion`` zu einem einzigen
(Multi-)Polygon verschmolzen. Der Layer enthält danach genau ein Feature.
Die Flurstückskennzeichen werden kommagetrennt im Attribut gespeichert.
Parameters
----------
source_layer :
Flurstück-Layer mit aktiver Auswahl.
Returns
-------
Tuple[layer, erfolgreich, fehlgeschlagen]
"""
crs = source_layer.crs().authid()
mem_layer = _erstelle_memory_layer(crs)
pr = mem_layer.dataProvider()
erfolgreich = 0
fehlgeschlagen = 0
geom_list = []
kz_list = []
gemarkung_set = set()
for src_feat in source_layer.selectedFeatures():
try:
geom = src_feat.geometry()
if not geom or geom.isEmpty():
fehlgeschlagen += 1
continue
geom_list.append(geom)
kz = str(src_feat[FELD_KENNZEICHEN] or "")
gm = str(src_feat[FELD_GEMARKUNG] or "")
if kz:
kz_list.append(kz)
if gm:
gemarkung_set.add(gm)
erfolgreich += 1
except Exception:
fehlgeschlagen += 1
if geom_list:
union_geom = QgsGeometry.unaryUnion(geom_list)
if union_geom and not union_geom.isEmpty():
feat = QgsFeature()
feat.setGeometry(union_geom)
feat.setAttributes([
",".join(kz_list),
",".join(sorted(gemarkung_set)),
])
pr.addFeatures([feat])
mem_layer.updateExtents()
return mem_layer, erfolgreich, fehlgeschlagen
def ergaenze_verfahrensgebiet(
verf_layer,
source_layer,
) -> Tuple[object, int, int, int]:
"""Fügt nur neue (noch nicht enthaltene) Flurstücke zum Verfahrensgebiet hinzu.
Die Geometrien der neuen Flurstücke werden mit der bestehenden Union-Geometrie
verschmolzen. Das Single-Feature wird an Ort und Stelle aktualisiert.
Parameters
----------
verf_layer :
Bestehender Verfahrensgebiet-Memory-Layer (Single-Feature).
source_layer :
Flurstück-Layer mit aktiver Auswahl.
Returns
-------
Tuple[verf_layer, hinzugefuegt, bereits_vorhanden, fehlgeschlagen]
"""
vorhandene_ids = get_kennzeichen_menge(verf_layer)
pr = verf_layer.dataProvider()
hinzugefuegt = 0
bereits_vorhanden = 0
fehlgeschlagen = 0
neue_geoms = []
neue_kz = []
neue_gemarkungen: set = set()
for src_feat in source_layer.selectedFeatures():
try:
kz = str(src_feat[FELD_KENNZEICHEN] or "")
if kz in vorhandene_ids:
bereits_vorhanden += 1
continue
geom = src_feat.geometry()
if not geom or geom.isEmpty():
fehlgeschlagen += 1
continue
neue_geoms.append(geom)
if kz:
neue_kz.append(kz)
gm = str(src_feat[FELD_GEMARKUNG] or "")
if gm:
neue_gemarkungen.add(gm)
hinzugefuegt += 1
except Exception:
fehlgeschlagen += 1
if neue_geoms:
existing_feats = list(verf_layer.getFeatures())
if existing_feats:
existing_feat = existing_feats[0]
all_geoms = [existing_feat.geometry()] + neue_geoms
new_union = QgsGeometry.unaryUnion(all_geoms)
alle_kz = sorted(vorhandene_ids | set(neue_kz))
existing_gm_str = existing_feat[FELD_GEMARKUNG] or ""
alle_gm = sorted(
{g.strip() for g in existing_gm_str.split(",") if g.strip()} | neue_gemarkungen
)
kz_idx = verf_layer.fields().indexFromName(FELD_KENNZEICHEN)
gm_idx = verf_layer.fields().indexFromName(FELD_GEMARKUNG)
pr.changeAttributeValues({existing_feat.id(): {
kz_idx: ",".join(alle_kz),
gm_idx: ",".join(alle_gm),
}})
pr.changeGeometryValues({existing_feat.id(): new_union})
else:
# Kein vorhandenes Feature – neues anlegen
union_geom = QgsGeometry.unaryUnion(neue_geoms)
if union_geom and not union_geom.isEmpty():
feat = QgsFeature()
feat.setGeometry(union_geom)
feat.setAttributes([
",".join(sorted(neue_kz)),
",".join(sorted(neue_gemarkungen)),
])
pr.addFeatures([feat])
verf_layer.updateExtents()
verf_layer.triggerRepaint()
return verf_layer, hinzugefuegt, bereits_vorhanden, fehlgeschlagen
def entferne_aus_verfahrensgebiet(
verf_layer,
source_layer,
) -> Tuple[object, int, int, int]:
"""Entfernt Flurstücke aus dem Verfahrensgebiet.
Die Geometrien der zu entfernenden Flurstücke werden per ``difference()``
von der bestehenden Union-Geometrie subtrahiert.
Nicht enthaltene Flurstücke werden gezählt, aber ignoriert.
Parameters
----------
verf_layer :
Bestehender Verfahrensgebiet-Memory-Layer (Single-Feature).
source_layer :
Flurstück-Layer mit aktiver Auswahl.
Returns
-------
Tuple[verf_layer, entfernt, nicht_enthalten, fehlgeschlagen]
"""
existing_feats = list(verf_layer.getFeatures())
if not existing_feats:
auswahl_kz = {
str(f[FELD_KENNZEICHEN] or "")
for f in source_layer.selectedFeatures()
if f[FELD_KENNZEICHEN]
}
return verf_layer, 0, len(auswahl_kz), 0
existing_feat = existing_feats[0]
existing_kz_str = existing_feat[FELD_KENNZEICHEN] or ""
vorhandene_ids = {k.strip() for k in existing_kz_str.split(",") if k.strip()}
auswahl_kz = {
str(f[FELD_KENNZEICHEN] or "")
for f in source_layer.selectedFeatures()
if f[FELD_KENNZEICHEN]
}
nicht_enthalten = len(auswahl_kz - vorhandene_ids)
zu_entfernen_kz = auswahl_kz & vorhandene_ids
if not zu_entfernen_kz:
return verf_layer, 0, nicht_enthalten, 0
result_geom = existing_feat.geometry()
fehlgeschlagen = 0
tatsaechlich_entfernt = 0
for src_feat in source_layer.selectedFeatures():
kz = str(src_feat[FELD_KENNZEICHEN] or "")
if kz not in zu_entfernen_kz:
continue
try:
geom = src_feat.geometry()
if geom and not geom.isEmpty():
result_geom = result_geom.difference(geom)
tatsaechlich_entfernt += 1
except Exception:
fehlgeschlagen += 1
pr = verf_layer.dataProvider()
verbleibende_kz = sorted(vorhandene_ids - zu_entfernen_kz)
if not verbleibende_kz or result_geom is None or result_geom.isEmpty():
pr.deleteFeatures([existing_feat.id()])
else:
kz_idx = verf_layer.fields().indexFromName(FELD_KENNZEICHEN)
pr.changeGeometryValues({existing_feat.id(): result_geom})
pr.changeAttributeValues({existing_feat.id(): {
kz_idx: ",".join(verbleibende_kz),
}})
verf_layer.updateExtents()
verf_layer.triggerRepaint()
return verf_layer, tatsaechlich_entfernt, nicht_enthalten, fehlgeschlagen
def ersetze_verfahrensgebiet(
source_layer,
) -> Tuple[Optional[object], int, int]:
"""Erstellt einen neuen Verfahrensgebiet-Layer aus der Auswahl (Ersetzen-Semantik).
Der Aufrufer ist verantwortlich, den alten Layer aus dem Projekt zu entfernen,
bevor der neue Layer hinzugefügt wird.
Returns
-------
Tuple[layer, erfolgreich, fehlgeschlagen]
Entspricht ``erstelle_verfahrensgebiet_aus_auswahl``.
"""
return erstelle_verfahrensgebiet_aus_auswahl(source_layer)