Verfahrensgebiet-Unterschiedung bei ähnlicher VKZ ergänzt
This commit is contained in:
@@ -9,10 +9,23 @@ from sn_basis.functions import (
|
||||
from sn_basis.functions import QMessageBox
|
||||
from sn_basis.functions.qt_wrapper import QVariant
|
||||
from sn_basis.functions.variable_wrapper import get_variable
|
||||
from sn_basis.functions.dialog_wrapper import ask_bror_kombination
|
||||
|
||||
alkis_sf_url = "https://geodienste.sachsen.de/aaa/public_alkis/sf/wfs"
|
||||
typename = "adv:AX_BauRaumOderBodenordnungsrecht"
|
||||
|
||||
#: Attribute, die aus dem BROR-WFS-Dienst geladen und im Memory-Layer gespeichert werden.
|
||||
BROR_ATTRIBUTE_FELDER = [
|
||||
"artderfestlegung",
|
||||
"ausfuehrendestelle_ax_dienststelle_schluessel_stelle",
|
||||
"adv_name",
|
||||
"bezeichnung",
|
||||
"datumanordnung",
|
||||
"datumbesitzeinweisung",
|
||||
"datumrechtskraft",
|
||||
"datumabgabe",
|
||||
]
|
||||
|
||||
class LoadStatus(Enum):
|
||||
"""Status eines Layer-Ladevorgangs.
|
||||
|
||||
@@ -83,11 +96,13 @@ def verfahrensgebiet_alkis(tab_widget):
|
||||
if status == LoadStatus.KEEP:
|
||||
return existing, status, []
|
||||
|
||||
# WFS mit Filter laden
|
||||
# WFS mit Filter laden – nur erste 5 Stellen der VKZ für den Vergleich verwenden,
|
||||
# da die Daten im Dienst ggf. abweichende Suffixe führen.
|
||||
vkz_prefix = verfahrensnummer[:5]
|
||||
wfs_uri_find = (
|
||||
f"url='{alkis_sf_url}' typename='{typename}' srsname='EPSG:25833' "
|
||||
f"sql=SELECT * FROM AX_BauRaumOderBodenordnungsrecht "
|
||||
f"WHERE bezeichnung LIKE '{verfahrensnummer}%'"
|
||||
f"WHERE bezeichnung LIKE '{vkz_prefix}%'"
|
||||
)
|
||||
temp_layer = QgsVectorLayer(wfs_uri_find, "TempVerfahrensgebiet", "WFS")
|
||||
|
||||
@@ -101,7 +116,32 @@ def verfahrensgebiet_alkis(tab_widget):
|
||||
f"Verfahrenskennzeichen {verfahrensnummer} nicht im WFS gefunden.")
|
||||
return None, LoadStatus.NONE, []
|
||||
|
||||
union_geom = QgsGeometry.unaryUnion([f.geometry() for f in features])
|
||||
# Objekte nach (adv_name, bezeichnung) gruppieren
|
||||
kombinationen: dict[tuple[str, str], list] = {}
|
||||
for feat in features:
|
||||
try:
|
||||
adv_name_val = str(feat.attribute("adv_name") or "")
|
||||
bezeichnung_val = str(feat.attribute("bezeichnung") or "")
|
||||
except Exception:
|
||||
adv_name_val = ""
|
||||
bezeichnung_val = ""
|
||||
key = (adv_name_val, bezeichnung_val)
|
||||
if key not in kombinationen:
|
||||
kombinationen[key] = []
|
||||
kombinationen[key].append(feat)
|
||||
|
||||
if len(kombinationen) == 1:
|
||||
selected_key = next(iter(kombinationen))
|
||||
selected_feats = kombinationen[selected_key]
|
||||
else:
|
||||
# Mehrere Kombinationen → Nutzerauswahl
|
||||
key_list = sorted(kombinationen.keys())
|
||||
selected_key = ask_bror_kombination(tab_widget, key_list)
|
||||
if selected_key is None:
|
||||
return None, LoadStatus.NONE, []
|
||||
selected_feats = kombinationen[selected_key]
|
||||
|
||||
union_geom = QgsGeometry.unaryUnion([f.geometry() for f in selected_feats])
|
||||
if union_geom is None or union_geom.isEmpty():
|
||||
QMessageBox.critical(tab_widget, "Fehler", "Keine Geometrien zum Verschmelzen gefunden.")
|
||||
return None, LoadStatus.NONE, []
|
||||
@@ -110,14 +150,24 @@ def verfahrensgebiet_alkis(tab_widget):
|
||||
crs = temp_layer.crs().authid()
|
||||
memory_layer = QgsVectorLayer(f"Polygon?crs={crs}", "Verfahrensgebiet", "memory")
|
||||
pr = memory_layer.dataProvider()
|
||||
pr.addAttributes([QgsField("VKZ", QVariant.String)])
|
||||
pr.addAttributes([QgsField(attr, QVariant.String) for attr in BROR_ATTRIBUTE_FELDER])
|
||||
memory_layer.updateFields()
|
||||
|
||||
# Attributwerte aus dem ersten Feature der gewählten Kombination übernehmen
|
||||
first_feat = selected_feats[0]
|
||||
attrs = []
|
||||
for attr in BROR_ATTRIBUTE_FELDER:
|
||||
try:
|
||||
val = first_feat.attribute(attr)
|
||||
attrs.append(str(val) if val is not None else "")
|
||||
except Exception:
|
||||
attrs.append("")
|
||||
|
||||
feat_union = QgsFeature()
|
||||
feat_union.setGeometry(union_geom)
|
||||
feat_union.setAttributes([verfahrensnummer])
|
||||
feat_union.setAttributes(attrs)
|
||||
pr.addFeatures([feat_union])
|
||||
memory_layer.updateExtents()
|
||||
|
||||
# Status: FIRST oder RELOAD; features für Gemarkungsermittlung weitergeben
|
||||
return memory_layer, status, features
|
||||
# Status: FIRST oder RELOAD; selected_feats für Gemarkungsermittlung weitergeben
|
||||
return memory_layer, status, selected_feats
|
||||
|
||||
@@ -61,12 +61,16 @@ class _DummyGeom:
|
||||
|
||||
|
||||
class _DummyFeature:
|
||||
def __init__(self, geom=None):
|
||||
def __init__(self, geom=None, attrs=None):
|
||||
self._geom = geom or _DummyGeom()
|
||||
self._attrs = attrs or {}
|
||||
|
||||
def geometry(self):
|
||||
return self._geom
|
||||
|
||||
def attribute(self, name):
|
||||
return self._attrs.get(name)
|
||||
|
||||
|
||||
class _DummyTempLayer:
|
||||
def __init__(self, valid=True, features=None):
|
||||
@@ -139,7 +143,7 @@ class TestVerfahrensgebietAlkis(unittest.TestCase):
|
||||
def test_verfahrensgebiet_alkis_fails_when_verfahrensnummer_missing(self, _mock_get_var):
|
||||
_DummyMessageBox.last_critical = None
|
||||
|
||||
layer, status = vg.verfahrensgebiet_alkis(tab_widget=None)
|
||||
layer, status, feats = vg.verfahrensgebiet_alkis(tab_widget=None)
|
||||
|
||||
self.assertIsNone(layer)
|
||||
self.assertEqual(status, vg.LoadStatus.NONE)
|
||||
@@ -153,7 +157,7 @@ class TestVerfahrensgebietAlkis(unittest.TestCase):
|
||||
_DummyMessageBox.last_critical = None
|
||||
mock_qgsvectorlayer.return_value = _DummyTempLayer(valid=False)
|
||||
|
||||
layer, status = vg.verfahrensgebiet_alkis(tab_widget=None)
|
||||
layer, status, feats = vg.verfahrensgebiet_alkis(tab_widget=None)
|
||||
|
||||
self.assertIsNone(layer)
|
||||
self.assertEqual(status, vg.LoadStatus.NONE)
|
||||
@@ -165,8 +169,10 @@ class TestVerfahrensgebietAlkis(unittest.TestCase):
|
||||
@patch("sn_verfahrensgebiet.functions.verfahrensgebiet_alkis.get_variable", return_value="VKZ123")
|
||||
@patch("sn_verfahrensgebiet.functions.verfahrensgebiet_alkis.QgsVectorLayer")
|
||||
@patch("sn_verfahrensgebiet.functions.verfahrensgebiet_alkis.QMessageBox", new=_DummyMessageBox)
|
||||
def test_verfahrensgebiet_alkis_success_returns_memory_layer(self, mock_qgsvectorlayer, _mock_get_var, _mock_check, _mock_union, mock_feature_cls):
|
||||
temp_layer = _DummyTempLayer(valid=True, features=[_DummyFeature()])
|
||||
def test_verfahrensgebiet_alkis_success_single_kombination(self, mock_qgsvectorlayer, _mock_get_var, _mock_check, _mock_union, mock_feature_cls):
|
||||
"""Einzelne (adv_name, bezeichnung)-Kombination → kein Dialog, Memory-Layer zurück."""
|
||||
feat = _DummyFeature(attrs={"adv_name": "Bodensanierung", "bezeichnung": "VKZ12/001"})
|
||||
temp_layer = _DummyTempLayer(valid=True, features=[feat])
|
||||
memory_layer = _DummyMemoryLayer()
|
||||
mock_qgsvectorlayer.side_effect = [temp_layer, memory_layer]
|
||||
|
||||
@@ -174,10 +180,63 @@ class TestVerfahrensgebietAlkis(unittest.TestCase):
|
||||
mock_feature.setGeometry.return_value = None
|
||||
mock_feature.setAttributes.return_value = None
|
||||
|
||||
layer, status = vg.verfahrensgebiet_alkis(tab_widget=None)
|
||||
layer, status, feats = vg.verfahrensgebiet_alkis(tab_widget=None)
|
||||
|
||||
self.assertIs(layer, memory_layer)
|
||||
self.assertEqual(status, vg.LoadStatus.FIRST)
|
||||
self.assertEqual(len(feats), 1)
|
||||
|
||||
@patch("sn_verfahrensgebiet.functions.verfahrensgebiet_alkis.ask_bror_kombination")
|
||||
@patch("sn_verfahrensgebiet.functions.verfahrensgebiet_alkis.QgsFeature")
|
||||
@patch("sn_verfahrensgebiet.functions.verfahrensgebiet_alkis.QgsGeometry.unaryUnion", return_value=_DummyGeom(empty=False))
|
||||
@patch("sn_verfahrensgebiet.functions.verfahrensgebiet_alkis._check_existing_layer", return_value=(None, vg.LoadStatus.FIRST))
|
||||
@patch("sn_verfahrensgebiet.functions.verfahrensgebiet_alkis.get_variable", return_value="VKZ123")
|
||||
@patch("sn_verfahrensgebiet.functions.verfahrensgebiet_alkis.QgsVectorLayer")
|
||||
@patch("sn_verfahrensgebiet.functions.verfahrensgebiet_alkis.QMessageBox", new=_DummyMessageBox)
|
||||
def test_verfahrensgebiet_alkis_multi_kombination_dialog_auswahl(
|
||||
self, mock_qgsvectorlayer, _mock_get_var, _mock_check, _mock_union, mock_feature_cls, mock_dialog
|
||||
):
|
||||
"""Zwei verschiedene Kombinationen → Dialog wird gezeigt, gewählte Features genutzt."""
|
||||
feat1 = _DummyFeature(attrs={"adv_name": "ADV-A", "bezeichnung": "VKZ12/001"})
|
||||
feat2 = _DummyFeature(attrs={"adv_name": "ADV-B", "bezeichnung": "VKZ12/002"})
|
||||
temp_layer = _DummyTempLayer(valid=True, features=[feat1, feat2])
|
||||
memory_layer = _DummyMemoryLayer()
|
||||
mock_qgsvectorlayer.side_effect = [temp_layer, memory_layer]
|
||||
mock_dialog.return_value = ("ADV-A", "VKZ12/001")
|
||||
|
||||
mock_feature = mock_feature_cls.return_value
|
||||
mock_feature.setGeometry.return_value = None
|
||||
mock_feature.setAttributes.return_value = None
|
||||
|
||||
layer, status, feats = vg.verfahrensgebiet_alkis(tab_widget=None)
|
||||
|
||||
mock_dialog.assert_called_once()
|
||||
self.assertIs(layer, memory_layer)
|
||||
self.assertEqual(status, vg.LoadStatus.FIRST)
|
||||
# Nur feat1 gehört zur gewählten Kombination
|
||||
self.assertEqual(feats, [feat1])
|
||||
|
||||
@patch("sn_verfahrensgebiet.functions.verfahrensgebiet_alkis.ask_bror_kombination")
|
||||
@patch("sn_verfahrensgebiet.functions.verfahrensgebiet_alkis._check_existing_layer", return_value=(None, vg.LoadStatus.FIRST))
|
||||
@patch("sn_verfahrensgebiet.functions.verfahrensgebiet_alkis.get_variable", return_value="VKZ123")
|
||||
@patch("sn_verfahrensgebiet.functions.verfahrensgebiet_alkis.QgsVectorLayer")
|
||||
@patch("sn_verfahrensgebiet.functions.verfahrensgebiet_alkis.QMessageBox", new=_DummyMessageBox)
|
||||
def test_verfahrensgebiet_alkis_multi_kombination_dialog_abbruch(
|
||||
self, mock_qgsvectorlayer, _mock_get_var, _mock_check, mock_dialog
|
||||
):
|
||||
"""Zwei verschiedene Kombinationen → Nutzer bricht Dialog ab → NONE."""
|
||||
feat1 = _DummyFeature(attrs={"adv_name": "ADV-A", "bezeichnung": "VKZ12/001"})
|
||||
feat2 = _DummyFeature(attrs={"adv_name": "ADV-B", "bezeichnung": "VKZ12/002"})
|
||||
temp_layer = _DummyTempLayer(valid=True, features=[feat1, feat2])
|
||||
mock_qgsvectorlayer.return_value = temp_layer
|
||||
mock_dialog.return_value = None # Nutzer bricht ab
|
||||
|
||||
layer, status, feats = vg.verfahrensgebiet_alkis(tab_widget=None)
|
||||
|
||||
mock_dialog.assert_called_once()
|
||||
self.assertIsNone(layer)
|
||||
self.assertEqual(status, vg.LoadStatus.NONE)
|
||||
self.assertEqual(feats, [])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
Reference in New Issue
Block a user