359 lines
12 KiB
Python
359 lines
12 KiB
Python
"""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)
|