From bdc7d265e8b9ff7aa6624cd5eadcacae159f62b0 Mon Sep 17 00:00:00 2001 From: Michael Otto Date: Thu, 2 Jul 2026 11:26:45 +0200 Subject: [PATCH 1/7] auf neuen Release-Workflow umgestellt --- metadata.txt => .gitea/templates/metadata.template | 0 1 file changed, 0 insertions(+), 0 deletions(-) rename metadata.txt => .gitea/templates/metadata.template (100%) diff --git a/metadata.txt b/.gitea/templates/metadata.template similarity index 100% rename from metadata.txt rename to .gitea/templates/metadata.template From 978c259496c207cbb7359132a226f78ecb36267b Mon Sep 17 00:00:00 2001 From: daniel Date: Fri, 10 Jul 2026 07:33:31 +0200 Subject: [PATCH 2/7] =?UTF-8?q?knickpunktpr=C3=BCfung=20fehlende=20Knickup?= =?UTF-8?q?nkte=20auf=20Abschnitte=20auf=20dem=20BROBOR-Objekt=20umgestell?= =?UTF-8?q?t?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- functions/knickpunkt_pruefung.py | 287 ++++++++++++++++++++++-------- tests/test_knickpunkt_pruefung.py | 204 +++++++++++++++++++-- 2 files changed, 397 insertions(+), 94 deletions(-) diff --git a/functions/knickpunkt_pruefung.py b/functions/knickpunkt_pruefung.py index 4044c74..072c3ac 100644 --- a/functions/knickpunkt_pruefung.py +++ b/functions/knickpunkt_pruefung.py @@ -6,8 +6,10 @@ 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. +- **Fehlende Knickpunkte** (orange): Ein PunktortTA liegt innerhalb des + 3-cm-Puffers einer Verbindungslinie zwischen zwei aufeinanderfolgenden + gematchten PunktortTA am BROR-Umring, hat aber keinen entsprechenden + Knickpunkt im BROR. **Algorithmus:** @@ -16,9 +18,11 @@ identifiziert: 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. +4. Entlang der BROR-Außengrenze aufeinanderfolgende gematchte PunktortTA + ermitteln und für jedes konsekutive Paar eine gerade Verbindungslinie + bilden; diese um :data:`VERBINDUNGSLINIE_PUFFER_M` puffern. +5. Nicht-gematchte PunktortTA innerhalb der Puffer → orange. + Verbindungslinien- und Puffer-Layer werden dem QGIS-Projekt hinzugefügt. **Rückgabe:** ``(result_layer, identisch, kein_pta, fehlender_knick)`` """ @@ -42,8 +46,12 @@ from sn_basis.functions.qt_wrapper import QVariant KOORDINATEN_TOLERANZ_M: float = 0.01 #: Puffer um den BROR-Umring für die Suche nach fehlenden Knickpunkten (m). +#: Wird aus Kompatibilitätsgründen behalten, wird aber nicht mehr verwendet. BOUNDARY_PUFFER_M: float = 0.5 +#: Puffer um die Verbindungslinie zwischen zwei aufeinanderfolgenden PunktortTA (m). +VERBINDUNGSLINIE_PUFFER_M: float = 0.03 + #: Rundungsgenauigkeit für Vertex-Koordinaten im Flurstück-Set (Dezimalstellen). _VERTEX_ROUND = 3 @@ -216,55 +224,198 @@ def _baue_pta_index(punktort_layer) -> Tuple[object, Dict[int, object]]: 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. +def _fehlende_knickpunkte_via_verbindungslinien( + bror_union, + pta_index, + geom_map: Dict[int, object], + matched_pta_ids: Set[int], + punktort_layer, + crs_authid: str, +) -> Tuple[int, List[object]]: + """Ermittelt fehlende Knickpunkte via Verbindungslinien-Puffer. + + Bestimmt entlang der BROR-Außengrenze die Reihenfolge der gematchten + ``AX_PunktortTA``. Für jedes konsekutive Paar wird eine gerade + Verbindungslinie erzeugt und um :data:`VERBINDUNGSLINIE_PUFFER_M` gepuffert. + Nicht-gematchte PunktortTA innerhalb dieses Puffers werden als fehlende + Knickpunkte (orange) gewertet. + + Fügt dem QGIS-Projekt zwei Hilfs-Layer hinzu (falls ein Projekt-Kontext + vorhanden ist): + + - *Knickpunktprüfung – Verbindungslinien* + - *Knickpunktprüfung – Puffer* Parameters ---------- - flst_layer : - Flurstücke-Layer. - bror_buffer_geom : - Gepufferter BROR-Umring (Polygon-Geometrie). + bror_union : + Vereinigungsgeometrie des BROR-Polygons (``QgsGeometry``). + pta_index : + Spatial-Index der PunktortTA-Features (``QgsSpatialIndex``). + geom_map : + Feature-ID → ``QgsGeometry``-Mapping für PunktortTA. + matched_pta_ids : + IDs der bereits gematchten PunktortTA (identisch mit BROR-Knickpunkten). + punktort_layer : + Geladener ``AX_PunktortTA``-Layer (für zukünftige Erweiterungen + reserviert; wird derzeit nicht direkt verwendet). + crs_authid : + CRS-Kennung des BROR-Layers, z.B. ``"EPSG:25833"``. Returns ------- - Set[Tuple[float, float]] - Gerundete Vertex-Koordinaten. + Tuple[int, List[QgsFeature]] + ``(fehlender_knick_count, orange_feats)`` – Anzahl gefundener fehlender + Knickpunkte und Liste der zugehörigen ``QgsFeature``-Objekte. """ - 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(): + from qgis.core import QgsPointXY, QgsProject # type: ignore[import] + except ImportError: + return 0, [] + + if bror_union is None or bror_union.isEmpty(): + return 0, [] + + # ------------------------------------------------------------------ + # Außenringe extrahieren + # ------------------------------------------------------------------ + exterior_rings: List[List[object]] = [] + if bror_union.isMultipart(): + for poly in bror_union.asMultiPolygon(): + if poly: + exterior_rings.append([QgsPointXY(x, y) for x, y in poly[0]]) + else: + poly = bror_union.asPolygon() + if poly: + exterior_rings.append([QgsPointXY(x, y) for x, y in poly[0]]) + + # Schlusspunkt (= Startpunkt, Ring geschlossen) entfernen + cleaned_rings: List[List[object]] = [] + for ring in exterior_rings: + if len(ring) >= 2: + if ( + abs(ring[-1].x() - ring[0].x()) < 1e-9 + and abs(ring[-1].y() - ring[0].y()) < 1e-9 + ): + ring = ring[:-1] + if len(ring) >= 2: + cleaned_rings.append(ring) + + line_geoms: List[object] = [] + buffer_geoms: List[object] = [] + orange_ids: Set[int] = set() + + # ------------------------------------------------------------------ + # Pro Ring: gematchte PunktortTA in Reihenfolge, Paare bilden + # ------------------------------------------------------------------ + for ring_pts in cleaned_rings: + # Gematchte PunktortTA auf diesem Ring lokalisieren + pta_on_ring: List[Tuple[int, int, object]] = [] # (ring_idx, fid, pt) + for fid in matched_pta_ids: + pta_geom = geom_map.get(fid) + if pta_geom is None: 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: + pta_pt = pta_geom.asPoint() + for i, ring_pt in enumerate(ring_pts): + dist_sq = ( + (ring_pt.x() - pta_pt.x()) ** 2 + + (ring_pt.y() - pta_pt.y()) ** 2 + ) + if dist_sq <= KOORDINATEN_TOLERANZ_M ** 2: + pta_on_ring.append((i, fid, ring_pt)) + break + + if len(pta_on_ring) < 2: + continue + + pta_on_ring.sort(key=lambda t: t[0]) + num_pta = len(pta_on_ring) + + for k in range(num_pta): + _, _, pt1 = pta_on_ring[k] + _, _, pt2 = pta_on_ring[(k + 1) % num_pta] + + # Gerade Verbindungslinie zwischen den zwei PunktortTA + line_geom = QgsGeometry.fromPolylineXY([pt1, pt2]) + if line_geom is None or line_geom.isEmpty(): continue + + line_geoms.append(line_geom) + + buffer_geom = line_geom.buffer(VERBINDUNGSLINIE_PUFFER_M, 8) + if buffer_geom is None or buffer_geom.isEmpty(): + continue + + buffer_geoms.append(buffer_geom) + + # Nicht-gematchte PunktortTA im Puffer suchen (via Spatial-Index) + bb = buffer_geom.boundingBox() + kandidaten = pta_index.intersects(bb) + for fid in kandidaten: + if fid in matched_pta_ids or fid in orange_ids: + continue + candidate_geom = geom_map.get(fid) + if candidate_geom is None: + continue + try: + if candidate_geom.within(buffer_geom): + orange_ids.add(fid) + except Exception: + continue + + # ------------------------------------------------------------------ + # Orange Features erstellen + # ------------------------------------------------------------------ + orange_feats: List[object] = [] + for fid in orange_ids: + pta_geom = geom_map.get(fid) + if pta_geom is None: + continue + feat = QgsFeature() + feat.setGeometry(pta_geom) + feat.setAttributes([WERT_FEHLENDER_KNICK]) + orange_feats.append(feat) + + # ------------------------------------------------------------------ + # Hilfs-Layer erstellen und zum Projekt hinzufügen + # ------------------------------------------------------------------ + try: + lines_layer = QgsVectorLayer( + f"LineString?crs={crs_authid}", + "Knickpunktprüfung – Verbindungslinien", + "memory", + ) + lines_pr = lines_layer.dataProvider() + line_feats_list = [] + for lg in line_geoms: + f = QgsFeature() + f.setGeometry(lg) + line_feats_list.append(f) + lines_pr.addFeatures(line_feats_list) + lines_layer.updateExtents() + + buffer_layer = QgsVectorLayer( + f"Polygon?crs={crs_authid}", + "Knickpunktprüfung – Puffer", + "memory", + ) + buffer_pr = buffer_layer.dataProvider() + buffer_feats_list = [] + for bg in buffer_geoms: + f = QgsFeature() + f.setGeometry(bg) + buffer_feats_list.append(f) + buffer_pr.addFeatures(buffer_feats_list) + buffer_layer.updateExtents() + + project = QgsProject.instance() + if project is not None: + project.addMapLayer(lines_layer) + project.addMapLayer(buffer_layer) except Exception: - pass - return vertex_set + pass # Kein QGIS-Kontext (Tests) → Layer-Erstellung überspringen + + return len(orange_feats), orange_feats # --------------------------------------------------------------------------- @@ -290,7 +441,8 @@ def knickpunkt_pruefung( punktort_layer : Geladener ``AX_PunktortTA``-Layer (Ergebnis aus ``grenzpunkt_alkis``). flst_layer : - Flurstücke-Layer für die Orange-Prüfung. Kann ``None`` sein. + Nicht mehr verwendet. Wird aus Kompatibilitätsgründen weiterhin + akzeptiert. progress : Optionaler ``ProgressDialog`` für Fortschrittsanzeige. knickpunkte : @@ -407,50 +559,33 @@ def knickpunkt_pruefung( return None, 0, 0, 0 # ------------------------------------------------------------------ - # Schritt n+2: Fehlende Knickpunkte (orange) ermitteln + # Schritt n+2: Fehlende Knickpunkte (orange) via Verbindungslinien # ------------------------------------------------------------------ _schritt("Knickpunktprüfung: Fehlende Knickpunkte ermitteln …", n + 2) fehlender_knick_count = 0 - # BROR-Umring ermitteln und puffern try: - bror_feats = [ + bror_feats_union = [ 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 + bror_union_geom = QgsGeometry.unaryUnion( + [f.geometry() for f in bror_feats_union] ) except Exception: - bror_buffer_geom = None + bror_union_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 not _canceled(): + fehlender_knick_count, orange_feats = _fehlende_knickpunkte_via_verbindungslinien( + bror_union_geom, + pta_index, + geom_map, + matched_pta_ids, + punktort_layer, + crs_authid, + ) + feats_to_add.extend(orange_feats) if _canceled(): return None, 0, 0, 0 diff --git a/tests/test_knickpunkt_pruefung.py b/tests/test_knickpunkt_pruefung.py index 882208a..213ecad 100644 --- a/tests/test_knickpunkt_pruefung.py +++ b/tests/test_knickpunkt_pruefung.py @@ -14,7 +14,7 @@ knickpunkt_pruefung: - 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 +- PunktortTA im 3cm-Puffer einer Verbindungslinie zwischen zwei gematchten PunktortTA → orange gezählt - Abbruch via progress.is_canceled() → (None, 0, 0, 0) """ @@ -278,29 +278,197 @@ class TestKnickpunktPruefungKlassifizierung(unittest.TestCase): self.assertEqual(kein_pta, 1) + # --------------------------------------------------------------------------- -# Tests: _baue_flst_vertex_set +# Tests: _fehlende_knickpunkte_via_verbindungslinien # --------------------------------------------------------------------------- -class TestBaueFlstVertexSet(unittest.TestCase): +class TestFehlendKnickpunkteViaVerbindungslinien(unittest.TestCase): + """Direkte Tests für :func:`_fehlende_knickpunkte_via_verbindungslinien`.""" - 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 _fn(self, *args, **kwargs): + from sn_verfahrensgebiet.functions.knickpunkt_pruefung import ( + _fehlende_knickpunkte_via_verbindungslinien, + ) + return _fehlende_knickpunkte_via_verbindungslinien(*args, **kwargs) - 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] + def test_leere_union_gibt_null_zurueck(self): + """bror_union.isEmpty() == True → (0, []).""" + bror_union = MagicMock() + bror_union.isEmpty.return_value = True + with patch("qgis.core.QgsPointXY", create=True), \ + patch("qgis.core.QgsProject", create=True): + count, feats = self._fn( + bror_union, MagicMock(), {}, set(), MagicMock(), "EPSG:25833" + ) + self.assertEqual(count, 0) + self.assertEqual(feats, []) + + def test_weniger_als_zwei_pta_auf_ring_kein_orange(self): + """Nur 1 gematchter PunktortTA auf Ring → keine fehlenden Knickpunkte.""" + bror_union = MagicMock() + bror_union.isEmpty.return_value = False + bror_union.isMultipart.return_value = False + bror_union.asPolygon.return_value = [ + [(0.0, 0.0), (10.0, 0.0), (10.0, 10.0), (0.0, 0.0)] + ] + + def _mock_ptxy(x, y): + pt = MagicMock() + pt.x.return_value = x + pt.y.return_value = y + return pt + + pta_geom = MagicMock() + pta_pt = MagicMock() + pta_pt.x.return_value = 0.0 + pta_pt.y.return_value = 0.0 + pta_geom.asPoint.return_value = pta_pt + geom_map = {1: pta_geom} + matched_pta_ids = {1} # nur 1 → < 2 + + with patch("qgis.core.QgsPointXY", create=True, side_effect=_mock_ptxy), \ + patch("qgis.core.QgsProject", create=True): + count, feats = self._fn( + bror_union, MagicMock(), geom_map, matched_pta_ids, + MagicMock(), "EPSG:25833" + ) + self.assertEqual(count, 0) + self.assertEqual(len(feats), 0) + + @patch(f"{_MOD}.QgsGeometry") + @patch(f"{_MOD}.QgsVectorLayer") + def test_ungematchter_pta_im_puffer_wird_orange( + self, mock_vl_cls, mock_geom_cls + ): + """Nicht-gematchter PunktortTA innerhalb 3-cm-Puffer → orange gezählt.""" + bror_union = MagicMock() + bror_union.isEmpty.return_value = False + bror_union.isMultipart.return_value = False + bror_union.asPolygon.return_value = [ + [(0.0, 0.0), (10.0, 0.0), (10.0, 10.0), (0.0, 0.0)] + ] + + def _mock_ptxy(x, y): + pt = MagicMock() + pt.x.return_value = x + pt.y.return_value = y + return pt + + # Zwei gematchte PunktortTA (fid 1 und 2) + pta_geom_1 = MagicMock() + pta_pt_1 = MagicMock() + pta_pt_1.x.return_value = 0.0 + pta_pt_1.y.return_value = 0.0 + pta_geom_1.asPoint.return_value = pta_pt_1 + + pta_geom_2 = MagicMock() + pta_pt_2 = MagicMock() + pta_pt_2.x.return_value = 10.0 + pta_pt_2.y.return_value = 0.0 + pta_geom_2.asPoint.return_value = pta_pt_2 + + # Nicht-gematchter PunktortTA (fid=3) liegt im Puffer + pta_geom_3 = MagicMock() + pta_geom_3.isEmpty.return_value = False + pta_geom_3.within.return_value = True + + geom_map = {1: pta_geom_1, 2: pta_geom_2, 3: pta_geom_3} + matched_pta_ids = {1, 2} + + # Spatial-Index liefert fid=3 als Kandidat + pta_index = MagicMock() buffer_geom = MagicMock() - result = _baue_flst_vertex_set(layer, buffer_geom) - self.assertEqual(result, set()) + buffer_geom.isEmpty.return_value = False + buffer_geom.boundingBox.return_value = MagicMock() + pta_index.intersects.return_value = [3] + + line_geom = MagicMock() + line_geom.isEmpty.return_value = False + line_geom.buffer.return_value = buffer_geom + mock_geom_cls.fromPolylineXY.return_value = line_geom + + pr_mock = MagicMock() + pr_mock.addFeatures.return_value = True + layer_mock = MagicMock() + layer_mock.dataProvider.return_value = pr_mock + mock_vl_cls.return_value = layer_mock + + with patch("qgis.core.QgsPointXY", create=True, side_effect=_mock_ptxy), \ + patch("qgis.core.QgsProject", create=True) as mock_proj: + mock_proj.instance.return_value = MagicMock() + count, feats = self._fn( + bror_union, pta_index, geom_map, matched_pta_ids, + MagicMock(), "EPSG:25833" + ) + + self.assertEqual(count, 1) + self.assertEqual(len(feats), 1) + + @patch(f"{_MOD}.QgsGeometry") + @patch(f"{_MOD}.QgsVectorLayer") + def test_ungematchter_pta_ausserhalb_puffer_kein_orange( + self, mock_vl_cls, mock_geom_cls + ): + """Nicht-gematchter PunktortTA außerhalb Puffer → nicht orange gezählt.""" + bror_union = MagicMock() + bror_union.isEmpty.return_value = False + bror_union.isMultipart.return_value = False + bror_union.asPolygon.return_value = [ + [(0.0, 0.0), (10.0, 0.0), (10.0, 10.0), (0.0, 0.0)] + ] + + def _mock_ptxy(x, y): + pt = MagicMock() + pt.x.return_value = x + pt.y.return_value = y + return pt + + pta_geom_1 = MagicMock() + pta_pt_1 = MagicMock() + pta_pt_1.x.return_value = 0.0 + pta_pt_1.y.return_value = 0.0 + pta_geom_1.asPoint.return_value = pta_pt_1 + + pta_geom_2 = MagicMock() + pta_pt_2 = MagicMock() + pta_pt_2.x.return_value = 10.0 + pta_pt_2.y.return_value = 0.0 + pta_geom_2.asPoint.return_value = pta_pt_2 + + # fid=3 liegt NICHT im Puffer + pta_geom_3 = MagicMock() + pta_geom_3.within.return_value = False + + geom_map = {1: pta_geom_1, 2: pta_geom_2, 3: pta_geom_3} + matched_pta_ids = {1, 2} + + pta_index = MagicMock() + buffer_geom = MagicMock() + buffer_geom.isEmpty.return_value = False + buffer_geom.boundingBox.return_value = MagicMock() + pta_index.intersects.return_value = [3] + + line_geom = MagicMock() + line_geom.isEmpty.return_value = False + line_geom.buffer.return_value = buffer_geom + mock_geom_cls.fromPolylineXY.return_value = line_geom + + pr_mock = MagicMock() + layer_mock = MagicMock() + layer_mock.dataProvider.return_value = pr_mock + mock_vl_cls.return_value = layer_mock + + with patch("qgis.core.QgsPointXY", create=True, side_effect=_mock_ptxy), \ + patch("qgis.core.QgsProject", create=True) as mock_proj: + mock_proj.instance.return_value = MagicMock() + count, feats = self._fn( + bror_union, pta_index, geom_map, matched_pta_ids, + MagicMock(), "EPSG:25833" + ) + + self.assertEqual(count, 0) + self.assertEqual(len(feats), 0) if __name__ == "__main__": From ac12038a8a4db9bb6d3e238f6e7c9d9e9871269e Mon Sep 17 00:00:00 2001 From: daniel Date: Fri, 10 Jul 2026 08:24:10 +0200 Subject: [PATCH 3/7] Verfahrensgebietsbildung mit korrigierten Knickpunkten --- functions/knickpunkt_pruefung.py | 204 +++++++++++++++++++ functions/verfahrensgebiet_alkis_komplett.py | 26 ++- 2 files changed, 229 insertions(+), 1 deletion(-) diff --git a/functions/knickpunkt_pruefung.py b/functions/knickpunkt_pruefung.py index 072c3ac..afdf23b 100644 --- a/functions/knickpunkt_pruefung.py +++ b/functions/knickpunkt_pruefung.py @@ -600,3 +600,207 @@ def knickpunkt_pruefung( _setze_kategorien_symbologie(ergebnis_layer) return ergebnis_layer, identisch_count, kein_pta_count, fehlender_knick_count + + +def erstelle_korrigiertes_polygon( + bror_layer, + result_layer, + crs_authid: str, +) -> Tuple[Optional[object], Optional[object]]: + """Erstellt korrigierte Verbindungslinien und ein Polygon aus identischen + und gefundenen fehlenden Knickpunkten. + + Kombiniert die gematchten PunktortTA (``identisch``, grün) mit den + gefundenen fehlenden Knickpunkten (``fehlender_knick``, orange) zu einem + korrigierten Verlauf entlang der BROR-Außengrenze. Orange Punkte werden + nach ihrer Projektion auf die Verbindungsstrecke zwischen ihren zwei + umgebenden identischen PunktortTA einsortiert. + + Parameters + ---------- + bror_layer : + BauRaumOderBodenordnungsrecht-Polygon-Layer (für die Ringreihenfolge). + result_layer : + Punkt-Ergebnislayer aus :func:`knickpunkt_pruefung` mit Feld ``typ``. + crs_authid : + CRS-Kennung, z.B. ``"EPSG:25833"``. + + Returns + ------- + Tuple[Optional[QgsVectorLayer], Optional[QgsVectorLayer]] + ``(linien_layer, polygon_layer)`` – Memory-Layer mit den korrigierten + Verbindungslinien und dem abgeleiteten Polygon, oder ``(None, None)`` + bei fehlgeschlagener Verarbeitung. + """ + try: + from qgis.core import QgsPointXY # type: ignore[import] + except ImportError: + return None, None + + # ------------------------------------------------------------------ + # Punkte aus Ergebnislayer extrahieren + # ------------------------------------------------------------------ + identisch_pts: List[object] = [] + orange_pts: List[object] = [] + + try: + for feat in result_layer.getFeatures(): + geom = feat.geometry() + if geom is None or geom.isEmpty(): + continue + pt = geom.asPoint() + typ = feat.attribute(FELD_TYP) + if typ == WERT_IDENTISCH: + identisch_pts.append(QgsPointXY(pt.x(), pt.y())) + elif typ == WERT_FEHLENDER_KNICK: + orange_pts.append(QgsPointXY(pt.x(), pt.y())) + except Exception: + return None, None + + if len(identisch_pts) < 3: + return None, None + + # ------------------------------------------------------------------ + # Alle BROR-Außenringe extrahieren (Multipolygon-sicher) + # ------------------------------------------------------------------ + 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]) + if bror_union is None or bror_union.isEmpty(): + return None, None + + if bror_union.isMultipart(): + all_raw_rings = [ + poly[0] for poly in bror_union.asMultiPolygon() if poly + ] + else: + poly = bror_union.asPolygon() + all_raw_rings = [poly[0]] if poly else [] + except Exception: + return None, None + + if not all_raw_rings: + return None, None + + all_line_segments: List[Tuple[object, object]] = [] + all_polygon_feats: List[object] = [] + + # ------------------------------------------------------------------ + # Pro Polygon-Teil: Identisch-Punkte verorten, Verlauf aufbauen + # ------------------------------------------------------------------ + for raw_ring in all_raw_rings: + ring_pts = [QgsPointXY(x, y) for x, y in raw_ring] + if len(ring_pts) >= 2 and ( + abs(ring_pts[-1].x() - ring_pts[0].x()) < 1e-9 + and abs(ring_pts[-1].y() - ring_pts[0].y()) < 1e-9 + ): + ring_pts = ring_pts[:-1] + + if len(ring_pts) < 3: + continue + + # Identisch-Punkte auf diesem Ring lokalisieren + on_ring: List[Tuple[int, object]] = [] + for pt in identisch_pts: + for i, rpt in enumerate(ring_pts): + dist_sq = (rpt.x() - pt.x()) ** 2 + (rpt.y() - pt.y()) ** 2 + if dist_sq <= KOORDINATEN_TOLERANZ_M ** 2: + on_ring.append((i, pt)) + break + + on_ring.sort(key=lambda t: t[0]) + + if len(on_ring) < 3: + # Fallback: Original-Ring dieses Teilgebiets verwenden + ordered_pts: List[object] = ring_pts + for j in range(len(ring_pts)): + all_line_segments.append( + (ring_pts[j], ring_pts[(j + 1) % len(ring_pts)]) + ) + else: + # Korrigierten Verlauf für diesen Teil aufbauen + ordered_pts = [] + n_part = len(on_ring) + + for k in range(n_part): + _, pt1 = on_ring[k] + _, pt2 = on_ring[(k + 1) % n_part] + + dx = pt2.x() - pt1.x() + dy = pt2.y() - pt1.y() + len_sq = dx * dx + dy * dy + + between: List[Tuple[float, object]] = [] + if len_sq > 0: + for opt in orange_pts: + t = ( + (opt.x() - pt1.x()) * dx + (opt.y() - pt1.y()) * dy + ) / len_sq + clamped_t = max(0.0, min(1.0, t)) + foot_x = pt1.x() + clamped_t * dx + foot_y = pt1.y() + clamped_t * dy + dist = ( + (opt.x() - foot_x) ** 2 + (opt.y() - foot_y) ** 2 + ) ** 0.5 + if dist <= VERBINDUNGSLINIE_PUFFER_M: + between.append((t, opt)) + + between.sort(key=lambda x: x[0]) + + segment_pts = [pt1] + [p for _, p in between] + ordered_pts.extend(segment_pts) + + all_pts_in_seg = segment_pts + [pt2] + for j in range(len(all_pts_in_seg) - 1): + all_line_segments.append( + (all_pts_in_seg[j], all_pts_in_seg[j + 1]) + ) + + if len(ordered_pts) < 3: + continue + + polygon_ring = ordered_pts + [ordered_pts[0]] + polygon_geom = QgsGeometry.fromPolygonXY([polygon_ring]) + if polygon_geom and not polygon_geom.isEmpty(): + pf = QgsFeature() + pf.setGeometry(polygon_geom) + all_polygon_feats.append(pf) + + if not all_polygon_feats: + return None, None + + # ------------------------------------------------------------------ + # Linien-Layer erstellen + # ------------------------------------------------------------------ + linien_layer = QgsVectorLayer( + f"LineString?crs={crs_authid}", + "Korrigierte Verbindungslinien (Detailprüfung)", + "memory", + ) + lines_pr = linien_layer.dataProvider() + line_feats_list = [] + for p1, p2 in all_line_segments: + lgeom = QgsGeometry.fromPolylineXY([p1, p2]) + if lgeom and not lgeom.isEmpty(): + f = QgsFeature() + f.setGeometry(lgeom) + line_feats_list.append(f) + lines_pr.addFeatures(line_feats_list) + linien_layer.updateExtents() + + # ------------------------------------------------------------------ + # Polygon-Layer erstellen (ein Feature pro Teilgebiet) + # ------------------------------------------------------------------ + polygon_layer = QgsVectorLayer( + f"Polygon?crs={crs_authid}", + "Korrigiertes Polygon (Detailprüfung)", + "memory", + ) + poly_pr = polygon_layer.dataProvider() + poly_pr.addFeatures(all_polygon_feats) + polygon_layer.updateExtents() + + return linien_layer, polygon_layer diff --git a/functions/verfahrensgebiet_alkis_komplett.py b/functions/verfahrensgebiet_alkis_komplett.py index efae825..f069368 100644 --- a/functions/verfahrensgebiet_alkis_komplett.py +++ b/functions/verfahrensgebiet_alkis_komplett.py @@ -22,7 +22,10 @@ optionaler Umringsprüfung: - Grenzpunkte (AX_PunktortTA) per BBox laden. - Knickpunkte des BROR mit PunktortTA vergleichen (grün/rot/orange). - - Ergebnislayer (Punkte) ins Projekt einfügen. + - Korrigiertes Polygon aus identischen + gefundenen PunktortTA ableiten. + - Flurstücke mit korrigiertem Polygon neu selektieren und enriched Layer + erzeugen. + - Ergebnislayer (Punkte), Verbindungslinien und Polygon ins Projekt einfügen. - Abschlussmeldung mit Zählung. Das BROR-Objekt (``AX_BauRaumOderBodenordnungsrecht``) bleibt bei Erfolg als @@ -66,6 +69,7 @@ from sn_verfahrensgebiet.functions.grenzpunkt_alkis import lade_punktort_ta from sn_verfahrensgebiet.functions.knickpunkt_pruefung import ( knickpunkt_pruefung, _extrahiere_knickpunkte, + erstelle_korrigiertes_polygon, ) # Feldname für das Verfahrenskennzeichen im enriched Layer @@ -523,6 +527,26 @@ def verfahrensgebiet_alkis_komplett( detail_progress.set_value(gesamt_schritte) detail_progress.close() + # ------------------------------------------------------------------ + # Korrigiertes Polygon aus identischen + fehlenden Knickpunkten + # ------------------------------------------------------------------ + if ergebnis_layer and ergebnis_layer.isValid(): + linien_layer_korr, polygon_layer_korr = erstelle_korrigiertes_polygon( + alkis_vg_layer, ergebnis_layer, alkis_vg_layer.crs().authid() + ) + if polygon_layer_korr is not None and polygon_layer_korr.isValid(): + anzahl_korr = _select_by_location(flst_layer, polygon_layer_korr) + if anzahl_korr > 0: + enriched_layer_neu, _, _ = erstelle_verfahrensgebiet_aus_auswahl( + flst_layer + ) + if enriched_layer_neu and enriched_layer_neu.isValid(): + _fuege_vkz_feld_hinzu(enriched_layer_neu, vkz) + enriched_layer = enriched_layer_neu + 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) From 0af4a79f8d6300adc1c471dd4746b80ada7f1b68 Mon Sep 17 00:00:00 2001 From: Daniel Helbig Date: Fri, 10 Jul 2026 08:29:11 +0200 Subject: [PATCH 4/7] changelog.txt aktualisiert --- changelog.txt | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/changelog.txt b/changelog.txt index 2123cd4..fdbec9d 100644 --- a/changelog.txt +++ b/changelog.txt @@ -1,3 +1,7 @@ +- Knickpunktprüfung "fehlende Knickpunkte" auf Extraktion der Verbindungslinien zwischen bekannten Punktorten umgestellt +- Umringsdetektion und Ladepuffer Punktort_TA als Layer im Projekt bei der Knickpunktprüfung ergänzt +- Verfahrensgebiet-Bildung mit korrigierten Knickpunkten (keine manuelle Korrektur der Flurstückszugehörigkeit mehr erforderlich) +--- - Gemarkungen aus ALKIS (vereinfacht) laden ergänzt - Flurstücke aus Gemarkungsauswahl laden ergänzt --- From 0c5e6df736d91cf0cede0e9488dfa615792553cb Mon Sep 17 00:00:00 2001 From: daniel Date: Fri, 10 Jul 2026 09:29:25 +0200 Subject: [PATCH 5/7] =?UTF-8?q?fix:=20innere=20Ringe=20bleiben=20unber?= =?UTF-8?q?=C3=BCcksichtigt?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- functions/knickpunkt_pruefung.py | 159 ++++++++++++++++--------------- 1 file changed, 81 insertions(+), 78 deletions(-) diff --git a/functions/knickpunkt_pruefung.py b/functions/knickpunkt_pruefung.py index afdf23b..edf635a 100644 --- a/functions/knickpunkt_pruefung.py +++ b/functions/knickpunkt_pruefung.py @@ -277,21 +277,20 @@ def _fehlende_knickpunkte_via_verbindungslinien( return 0, [] # ------------------------------------------------------------------ - # Außenringe extrahieren + # Alle Ringe extrahieren (Außen- und Innenringe) # ------------------------------------------------------------------ - exterior_rings: List[List[object]] = [] + all_rings: List[List[object]] = [] if bror_union.isMultipart(): for poly in bror_union.asMultiPolygon(): - if poly: - exterior_rings.append([QgsPointXY(x, y) for x, y in poly[0]]) + for ring in poly: + all_rings.append([QgsPointXY(x, y) for x, y in ring]) else: - poly = bror_union.asPolygon() - if poly: - exterior_rings.append([QgsPointXY(x, y) for x, y in poly[0]]) + for ring in bror_union.asPolygon(): + all_rings.append([QgsPointXY(x, y) for x, y in ring]) # Schlusspunkt (= Startpunkt, Ring geschlossen) entfernen cleaned_rings: List[List[object]] = [] - for ring in exterior_rings: + for ring in all_rings: if len(ring) >= 2: if ( abs(ring[-1].x() - ring[0].x()) < 1e-9 @@ -661,7 +660,7 @@ def erstelle_korrigiertes_polygon( return None, None # ------------------------------------------------------------------ - # Alle BROR-Außenringe extrahieren (Multipolygon-sicher) + # Alle Polygon-Teile mit allen Ringen (Außen- und Innenringe) # ------------------------------------------------------------------ try: bror_feats = [ @@ -673,97 +672,101 @@ def erstelle_korrigiertes_polygon( return None, None if bror_union.isMultipart(): - all_raw_rings = [ - poly[0] for poly in bror_union.asMultiPolygon() if poly - ] + all_poly_parts = [poly for poly in bror_union.asMultiPolygon() if poly] else: poly = bror_union.asPolygon() - all_raw_rings = [poly[0]] if poly else [] + all_poly_parts = [poly] if poly else [] except Exception: return None, None - if not all_raw_rings: + if not all_poly_parts: return None, None all_line_segments: List[Tuple[object, object]] = [] all_polygon_feats: List[object] = [] # ------------------------------------------------------------------ - # Pro Polygon-Teil: Identisch-Punkte verorten, Verlauf aufbauen + # Pro Polygon-Teil: alle Ringe (Außen + Innen) verarbeiten # ------------------------------------------------------------------ - for raw_ring in all_raw_rings: - ring_pts = [QgsPointXY(x, y) for x, y in raw_ring] - if len(ring_pts) >= 2 and ( - abs(ring_pts[-1].x() - ring_pts[0].x()) < 1e-9 - and abs(ring_pts[-1].y() - ring_pts[0].y()) < 1e-9 - ): - ring_pts = ring_pts[:-1] + for poly_rings in all_poly_parts: + corrected_qgs_rings: List[List[object]] = [] - if len(ring_pts) < 3: - continue + for raw_ring in poly_rings: + ring_pts = [QgsPointXY(x, y) for x, y in raw_ring] + if len(ring_pts) >= 2 and ( + abs(ring_pts[-1].x() - ring_pts[0].x()) < 1e-9 + and abs(ring_pts[-1].y() - ring_pts[0].y()) < 1e-9 + ): + ring_pts = ring_pts[:-1] - # Identisch-Punkte auf diesem Ring lokalisieren - on_ring: List[Tuple[int, object]] = [] - for pt in identisch_pts: - for i, rpt in enumerate(ring_pts): - dist_sq = (rpt.x() - pt.x()) ** 2 + (rpt.y() - pt.y()) ** 2 - if dist_sq <= KOORDINATEN_TOLERANZ_M ** 2: - on_ring.append((i, pt)) - break + if len(ring_pts) < 3: + continue - on_ring.sort(key=lambda t: t[0]) + # Identisch-Punkte auf diesem Ring lokalisieren + on_ring: List[Tuple[int, object]] = [] + for pt in identisch_pts: + for i, rpt in enumerate(ring_pts): + dist_sq = (rpt.x() - pt.x()) ** 2 + (rpt.y() - pt.y()) ** 2 + if dist_sq <= KOORDINATEN_TOLERANZ_M ** 2: + on_ring.append((i, pt)) + break - if len(on_ring) < 3: - # Fallback: Original-Ring dieses Teilgebiets verwenden - ordered_pts: List[object] = ring_pts - for j in range(len(ring_pts)): - all_line_segments.append( - (ring_pts[j], ring_pts[(j + 1) % len(ring_pts)]) - ) - else: - # Korrigierten Verlauf für diesen Teil aufbauen - ordered_pts = [] - n_part = len(on_ring) + on_ring.sort(key=lambda t: t[0]) - for k in range(n_part): - _, pt1 = on_ring[k] - _, pt2 = on_ring[(k + 1) % n_part] - - dx = pt2.x() - pt1.x() - dy = pt2.y() - pt1.y() - len_sq = dx * dx + dy * dy - - between: List[Tuple[float, object]] = [] - if len_sq > 0: - for opt in orange_pts: - t = ( - (opt.x() - pt1.x()) * dx + (opt.y() - pt1.y()) * dy - ) / len_sq - clamped_t = max(0.0, min(1.0, t)) - foot_x = pt1.x() + clamped_t * dx - foot_y = pt1.y() + clamped_t * dy - dist = ( - (opt.x() - foot_x) ** 2 + (opt.y() - foot_y) ** 2 - ) ** 0.5 - if dist <= VERBINDUNGSLINIE_PUFFER_M: - between.append((t, opt)) - - between.sort(key=lambda x: x[0]) - - segment_pts = [pt1] + [p for _, p in between] - ordered_pts.extend(segment_pts) - - all_pts_in_seg = segment_pts + [pt2] - for j in range(len(all_pts_in_seg) - 1): + if len(on_ring) < 3: + # Fallback: Original-Ring verwenden + ordered_pts: List[object] = ring_pts + for j in range(len(ring_pts)): all_line_segments.append( - (all_pts_in_seg[j], all_pts_in_seg[j + 1]) + (ring_pts[j], ring_pts[(j + 1) % len(ring_pts)]) ) + else: + # Korrigierten Verlauf für diesen Ring aufbauen + ordered_pts = [] + n_part = len(on_ring) - if len(ordered_pts) < 3: + for k in range(n_part): + _, pt1 = on_ring[k] + _, pt2 = on_ring[(k + 1) % n_part] + + dx = pt2.x() - pt1.x() + dy = pt2.y() - pt1.y() + len_sq = dx * dx + dy * dy + + between: List[Tuple[float, object]] = [] + if len_sq > 0: + for opt in orange_pts: + t = ( + (opt.x() - pt1.x()) * dx + (opt.y() - pt1.y()) * dy + ) / len_sq + clamped_t = max(0.0, min(1.0, t)) + foot_x = pt1.x() + clamped_t * dx + foot_y = pt1.y() + clamped_t * dy + dist = ( + (opt.x() - foot_x) ** 2 + (opt.y() - foot_y) ** 2 + ) ** 0.5 + if dist <= VERBINDUNGSLINIE_PUFFER_M: + between.append((t, opt)) + + between.sort(key=lambda x: x[0]) + + segment_pts = [pt1] + [p for _, p in between] + ordered_pts.extend(segment_pts) + + all_pts_in_seg = segment_pts + [pt2] + for j in range(len(all_pts_in_seg) - 1): + all_line_segments.append( + (all_pts_in_seg[j], all_pts_in_seg[j + 1]) + ) + + if len(ordered_pts) >= 3: + corrected_qgs_rings.append(ordered_pts + [ordered_pts[0]]) + + if not corrected_qgs_rings: continue - polygon_ring = ordered_pts + [ordered_pts[0]] - polygon_geom = QgsGeometry.fromPolygonXY([polygon_ring]) + # Erstes Element = Außenring, weitere = Innenringe (Löcher) + polygon_geom = QgsGeometry.fromPolygonXY(corrected_qgs_rings) if polygon_geom and not polygon_geom.isEmpty(): pf = QgsFeature() pf.setGeometry(polygon_geom) From f37f33584dbd9184ba86bdd077ac2fa227b81650 Mon Sep 17 00:00:00 2001 From: daniel Date: Wed, 15 Jul 2026 15:29:48 +0200 Subject: [PATCH 6/7] =?UTF-8?q?shape=20Verfahrensgebiet=20erg=C3=A4nzt?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- functions/verfahrensgebiet_alkis.py | 12 +- functions/verfahrensgebiet_shape_komplett.py | 587 +++++++++++ metadata.txt | 13 + tests/test_verfahrensgebiet_shape_komplett.py | 932 ++++++++++++++++++ tests/test_working_tab.py | 257 +++++ ui/tabs/working_tab.py | 62 +- 6 files changed, 1856 insertions(+), 7 deletions(-) create mode 100644 functions/verfahrensgebiet_shape_komplett.py create mode 100644 metadata.txt create mode 100644 tests/test_verfahrensgebiet_shape_komplett.py diff --git a/functions/verfahrensgebiet_alkis.py b/functions/verfahrensgebiet_alkis.py index dae1c29..be56b8e 100644 --- a/functions/verfahrensgebiet_alkis.py +++ b/functions/verfahrensgebiet_alkis.py @@ -72,7 +72,7 @@ def _check_existing_layer(tab_widget): return existing_layers[0], LoadStatus.KEEP -def verfahrensgebiet_alkis(tab_widget): +def verfahrensgebiet_alkis(tab_widget, skip_existing_check: bool = False): """Lädt Verfahrensgebiet aus ALKIS-WFS basierend auf Verfahrensnummer. Args: @@ -92,10 +92,12 @@ def verfahrensgebiet_alkis(tab_widget): "In den Projekteigenschaften ist noch keine Verfahrensnummer eingetragen!") return None, LoadStatus.NONE, [] - # Vorhandenen Layer prüfen - existing, status = _check_existing_layer(tab_widget) - if status == LoadStatus.KEEP: - return existing, status, [] + # Vorhandenen Layer prüfen (überspringbar für die Shape-Pipeline) + status = LoadStatus.FIRST + if not skip_existing_check: + existing, status = _check_existing_layer(tab_widget) + if status == LoadStatus.KEEP: + return existing, status, [] # 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. diff --git a/functions/verfahrensgebiet_shape_komplett.py b/functions/verfahrensgebiet_shape_komplett.py new file mode 100644 index 0000000..31c1c51 --- /dev/null +++ b/functions/verfahrensgebiet_shape_komplett.py @@ -0,0 +1,587 @@ +"""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 diff --git a/metadata.txt b/metadata.txt new file mode 100644 index 0000000..9695e79 --- /dev/null +++ b/metadata.txt @@ -0,0 +1,13 @@ +[general] +version=26.7.3-testing +name=LNO Sachsen | Verfahrensgebiet +description=Plugin zum Erzeugen eines Objektes "Verfahrensgebiet" +author=Daniel Helbig, Michael Otto +homepage=https://entwicklung.flurneuordnung-sachsen.de/AG_QGIS/Plugin_SN_Verfahrensgebiet +tracker=https://entwicklung.flurneuordnung-sachsen.de/AG_QGIS/Plugin_SN_Verfahrensgebiet/issues +repository=https://entwicklung.flurneuordnung-sachsen.de/AG_QGIS/Plugin_SN_Verfahrensgebiet/src/branch/testing/ +QGISMaximumVersion=4.99 +qgisMinimumVersion=3.40 +experimental=true +deprecated=false +supportsQt6=true \ No newline at end of file diff --git a/tests/test_verfahrensgebiet_shape_komplett.py b/tests/test_verfahrensgebiet_shape_komplett.py new file mode 100644 index 0000000..0ad6f0a --- /dev/null +++ b/tests/test_verfahrensgebiet_shape_komplett.py @@ -0,0 +1,932 @@ +"""sn_verfahrensgebiet/tests/test_verfahrensgebiet_shape_komplett.py + +Unit-Tests für :mod:`sn_verfahrensgebiet.functions.verfahrensgebiet_shape_komplett`. + +Getestete Szenarien: + +lade_vg_aus_zip: +- Dateidialog abgebrochen (leerer Pfad) → None +- ZIP-Entpacken schlägt fehl (Exception) → None + QMessageBox.warning +- Kein .shp in ZIP → None + Warning "gültiges Shape-Dateiset angeben" +- QgsVectorLayer nicht valid → None + Warning +- Leere Geometrien im Shapefile → None + Warning +- Erfolgspfad → Memory-Layer mit Union-Geometrie, Name "Verfahrensgebiet" + +verfahrensgebiet_shape_komplett: +- lade_vg_aus_zip None (Abbruch) → (None, None, None, NONE) +- flurstueck_alkis NONE → (None, None, None, NONE) +- Keine Flurstücke selektiert → (None, flst, None, NONE) + QMessageBox.warning +- erstelle_verfahrensgebiet_aus_auswahl ungültig → Workflow noch immer abgeschlossen + (temp_enriched-Fehler: stimmt_ueberein=True wegen 0-Fläche) +- Flächenübereinstimmung → show_info_dialog, return (shape_vg, flst, None, FIRST) +- Flächenabweichung + abbrechen → return (shape_vg, flst, None, FIRST) +- Flächenabweichung + detailpruefung → lade_punktort_ta + knickpunkt_pruefung aufgerufen +- Cleanup: TempVerfahrensgebiet stets aus Projekt entfernt +""" + +import sys +import os +import types +import unittest +from unittest.mock import patch, MagicMock, call + +# --------------------------------------------------------------------------- +# Minimales QGIS-Mock-Setup +# --------------------------------------------------------------------------- +if "qgis" not in sys.modules: + qgis_mod = types.ModuleType("qgis") + qgis_core = types.ModuleType("qgis.core") + 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.verfahrensgebiet_alkis import LoadStatus +from sn_verfahrensgebiet.functions.verfahrensgebiet_shape_komplett import ( + verfahrensgebiet_shape_komplett, + lade_vg_aus_zip, + _finde_shp, + _erstelle_memory_layer_aus_shp, +) + + +# --------------------------------------------------------------------------- +# Hilfs-Dummies +# --------------------------------------------------------------------------- + +def _make_layer(valid=True, name="Layer", lid="L1", feature_count=1, area=1000.0): + layer = MagicMock() + layer.isValid.return_value = valid + layer.id.return_value = lid + layer.name.return_value = name + layer.featureCount.return_value = feature_count + layer.selectedFeatureCount.return_value = feature_count + + mock_geom = MagicMock() + mock_geom.isEmpty.return_value = False + mock_geom.area.return_value = area + mock_feat = MagicMock() + mock_feat.geometry.return_value = mock_geom + layer.getFeatures.return_value = [mock_feat] + + mock_crs = MagicMock() + mock_crs.authid.return_value = "EPSG:25833" + layer.crs.return_value = mock_crs + return layer + + +class _DummyProject: + def __init__(self): + self.added = [] + self.removed = [] + self._layers_by_name: dict = {} + + def addMapLayer(self, layer): + self.added.append(layer) + + def removeMapLayer(self, lid): + self.removed.append(lid) + + def mapLayersByName(self, name): + return self._layers_by_name.get(name, []) + + def mapLayer(self, lid): + return None + + def instance(self): + return self + + +def _make_progress_mock(canceled=False): + prog = MagicMock() + prog.is_canceled.return_value = canceled + return prog + + +# --------------------------------------------------------------------------- +# Patch-Konstanten +# --------------------------------------------------------------------------- + +_MOD = "sn_verfahrensgebiet.functions.verfahrensgebiet_shape_komplett" +_PATCH_LADE_ZIP = f"{_MOD}.lade_vg_aus_zip" +_PATCH_GEM = f"{_MOD}.gemarkung_schluessel_und_namen_von_vg_geometrie" +_PATCH_FLST = f"{_MOD}.flurstueck_alkis" +_PATCH_SELECT = f"{_MOD}._select_by_location" +_PATCH_ERSTELLE = f"{_MOD}.erstelle_verfahrensgebiet_aus_auswahl" +_PATCH_PROJECT = f"{_MOD}.QgsProject" +_PATCH_MSGBOX = f"{_MOD}.QMessageBox" +_PATCH_WARNING = f"{_MOD}.warning" +_PATCH_PROGRESS = f"{_MOD}.create_progress_dialog" +_PATCH_VERGLEICHE = f"{_MOD}._vergleiche_flaechen" +_PATCH_SHOW_INFO = f"{_MOD}.show_info_dialog" +_PATCH_ASK_DIALOG = f"{_MOD}.ask_detailpruefung_oder_abbrechen" +_PATCH_GP_ALKIS = f"{_MOD}.lade_punktort_ta" +_PATCH_KNICK = f"{_MOD}.knickpunkt_pruefung" +_PATCH_EXTRAHIERE = f"{_MOD}._extrahiere_knickpunkte" +_PATCH_LESE_TOL = f"{_MOD}._lese_flaechen_toleranz" + + +# --------------------------------------------------------------------------- +# Tests: _finde_shp +# --------------------------------------------------------------------------- + +class TestFindShp(unittest.TestCase): + + def test_findet_shp_in_root(self): + with patch("os.walk") as mock_walk: + mock_walk.return_value = [ + ("/extract", [], ["shape.shp", "shape.dbf"]), + ] + result = _finde_shp("/extract") + self.assertEqual(result, os.path.join("/extract", "shape.shp")) + + def test_findet_shp_in_unterordner(self): + with patch("os.walk") as mock_walk: + mock_walk.return_value = [ + ("/extract", ["sub"], []), + ("/extract/sub", [], ["daten.shp"]), + ] + result = _finde_shp("/extract") + self.assertEqual(result, os.path.join("/extract/sub", "daten.shp")) + + def test_kein_shp_gibt_none_zurueck(self): + with patch("os.walk") as mock_walk: + mock_walk.return_value = [ + ("/extract", [], ["shape.dbf", "shape.prj"]), + ] + result = _finde_shp("/extract") + self.assertIsNone(result) + + def test_case_insensitive_erkennung(self): + with patch("os.walk") as mock_walk: + mock_walk.return_value = [ + ("/extract", [], ["DATEN.SHP"]), + ] + result = _finde_shp("/extract") + self.assertIsNotNone(result) + + +# --------------------------------------------------------------------------- +# Tests: _erstelle_memory_layer_aus_shp +# --------------------------------------------------------------------------- + +class TestErstelleMemoryLayerAusShp(unittest.TestCase): + + def test_leere_geometrien_geben_none(self): + shp_layer = MagicMock() + geom = MagicMock() + geom.isEmpty.return_value = True + feat = MagicMock() + feat.geometry.return_value = geom + shp_layer.getFeatures.return_value = [feat] + shp_layer.crs.return_value.authid.return_value = "EPSG:25833" + + result = _erstelle_memory_layer_aus_shp(shp_layer) + self.assertIsNone(result) + + def test_keine_features_gibt_none(self): + shp_layer = MagicMock() + shp_layer.getFeatures.return_value = [] + shp_layer.crs.return_value.authid.return_value = "EPSG:25833" + + result = _erstelle_memory_layer_aus_shp(shp_layer) + self.assertIsNone(result) + + @patch(f"{_MOD}.QgsGeometry") + @patch(f"{_MOD}.QgsVectorLayer") + @patch(f"{_MOD}.QgsFeature") + def test_attribute_von_erstem_feature_werden_gesetzt( + self, mock_feat_cls, mock_vl_cls, mock_geom_cls + ): + """Attributwerte des ersten Features werden auf das Union-Feature übertragen.""" + # Geometrie-Mock + mock_union = MagicMock() + mock_union.isEmpty.return_value = False + mock_geom_cls.unaryUnion.return_value = mock_union + + # Memory-Layer-Mock + mock_mem = MagicMock() + mock_mem.isValid.return_value = True + mock_mem.featureCount.return_value = 1 + mock_vl_cls.return_value = mock_mem + mock_pr = MagicMock() + mock_mem.dataProvider.return_value = mock_pr + + # Feature-Mock (Output) + mock_feat_out = MagicMock() + mock_feat_cls.return_value = mock_feat_out + + # Shapefile-Layer-Mock + shp_layer = MagicMock() + shp_layer.crs.return_value.authid.return_value = "EPSG:25833" + shp_layer.fields.return_value = [MagicMock(), MagicMock()] # list → iterable + + mock_geom = MagicMock() + mock_geom.isEmpty.return_value = False + src_feat = MagicMock() + src_feat.geometry.return_value = mock_geom + src_feat.attributes.return_value = ["wert_a", 42] + shp_layer.getFeatures.return_value = [src_feat] + + result = _erstelle_memory_layer_aus_shp(shp_layer) + + # Felder sollen übernommen werden + mock_pr.addAttributes.assert_called_once() + # Attributwerte sollen vom ersten Feature gesetzt werden + mock_feat_out.setAttributes.assert_called_once_with(["wert_a", 42]) + self.assertIs(result, mock_mem) + + +# --------------------------------------------------------------------------- +# Tests: lade_vg_aus_zip +# --------------------------------------------------------------------------- + +class TestLadeVgAusZip(unittest.TestCase): + + def _patch(self, **kwargs): + """Hilfsmethode: kombinierter Patch-Kontext für lade_vg_aus_zip.""" + return kwargs + + @patch(f"{_MOD}.QFileDialog") + def test_dialog_abgebrochen_gibt_none_zurueck(self, mock_fd): + mock_fd.getOpenFileName.return_value = ("", "") + result = lade_vg_aus_zip(None) + self.assertIsNone(result) + + @patch(f"{_MOD}.QMessageBox") + @patch(f"{_MOD}.zipfile") + @patch(f"{_MOD}.QFileDialog") + def test_zip_fehler_gibt_none_und_warning(self, mock_fd, mock_zf, mock_mb): + mock_fd.getOpenFileName.return_value = ("/tmp/vg.zip", "") + mock_zf.ZipFile.side_effect = Exception("bad zip") + + with patch("os.path.dirname", return_value="/tmp"), \ + patch("os.path.splitext", return_value=("/tmp/vg", ".zip")), \ + patch("os.path.basename", return_value="vg.zip"), \ + patch("os.path.join", return_value="/tmp/vg"): + result = lade_vg_aus_zip(None) + + self.assertIsNone(result) + mock_mb.warning.assert_called_once() + + @patch(f"{_MOD}.QMessageBox") + @patch(f"{_MOD}._finde_shp", return_value=None) + @patch(f"{_MOD}.zipfile") + @patch(f"{_MOD}.QFileDialog") + def test_kein_shp_in_zip_gibt_none_und_warning(self, mock_fd, mock_zf, mock_find, mock_mb): + mock_fd.getOpenFileName.return_value = ("/tmp/vg.zip", "") + mock_zf.ZipFile.return_value.__enter__ = MagicMock(return_value=MagicMock()) + mock_zf.ZipFile.return_value.__exit__ = MagicMock(return_value=False) + + with patch("os.path.dirname", return_value="/tmp"), \ + patch("os.path.splitext", return_value=("/tmp/vg", ".zip")), \ + patch("os.path.basename", return_value="vg.zip"), \ + patch("os.path.join", return_value="/tmp/vg"): + result = lade_vg_aus_zip(None) + + self.assertIsNone(result) + mock_mb.warning.assert_called_once() + args = mock_mb.warning.call_args[0] + self.assertIn("gültiges Shape-Dateiset angeben", args) + + @patch(f"{_MOD}.QMessageBox") + @patch(f"{_MOD}.QgsVectorLayer") + @patch(f"{_MOD}._finde_shp", return_value="/tmp/vg/shape.shp") + @patch(f"{_MOD}.zipfile") + @patch(f"{_MOD}.QFileDialog") + def test_invalider_qgslayer_gibt_none_und_warning( + self, mock_fd, mock_zf, mock_find, mock_vl, mock_mb + ): + mock_fd.getOpenFileName.return_value = ("/tmp/vg.zip", "") + mock_zf.ZipFile.return_value.__enter__ = MagicMock(return_value=MagicMock()) + mock_zf.ZipFile.return_value.__exit__ = MagicMock(return_value=False) + + invalid_layer = MagicMock() + invalid_layer.isValid.return_value = False + mock_vl.return_value = invalid_layer + + with patch("os.path.dirname", return_value="/tmp"), \ + patch("os.path.splitext", return_value=("/tmp/vg", ".zip")), \ + patch("os.path.basename", return_value="vg.zip"), \ + patch("os.path.join", return_value="/tmp/vg"): + result = lade_vg_aus_zip(None) + + self.assertIsNone(result) + mock_mb.warning.assert_called_once() + args = mock_mb.warning.call_args[0] + self.assertIn("gültiges Shape-Dateiset angeben", args) + + @patch(f"{_MOD}.QMessageBox") + @patch(f"{_MOD}._erstelle_memory_layer_aus_shp", return_value=None) + @patch(f"{_MOD}.QgsVectorLayer") + @patch(f"{_MOD}._finde_shp", return_value="/tmp/vg/shape.shp") + @patch(f"{_MOD}.zipfile") + @patch(f"{_MOD}.QFileDialog") + def test_leere_geometrien_geben_none_und_warning( + self, mock_fd, mock_zf, mock_find, mock_vl, mock_mem, mock_mb + ): + mock_fd.getOpenFileName.return_value = ("/tmp/vg.zip", "") + mock_zf.ZipFile.return_value.__enter__ = MagicMock(return_value=MagicMock()) + mock_zf.ZipFile.return_value.__exit__ = MagicMock(return_value=False) + valid_layer = MagicMock() + valid_layer.isValid.return_value = True + mock_vl.return_value = valid_layer + + with patch("os.path.dirname", return_value="/tmp"), \ + patch("os.path.splitext", return_value=("/tmp/vg", ".zip")), \ + patch("os.path.basename", return_value="vg.zip"), \ + patch("os.path.join", return_value="/tmp/vg"): + result = lade_vg_aus_zip(None) + + self.assertIsNone(result) + mock_mb.warning.assert_called_once() + + @patch(f"{_MOD}._erstelle_memory_layer_aus_shp") + @patch(f"{_MOD}.QgsVectorLayer") + @patch(f"{_MOD}._finde_shp", return_value="/tmp/vg/shape.shp") + @patch(f"{_MOD}.zipfile") + @patch(f"{_MOD}.QFileDialog") + def test_erfolg_gibt_memory_layer_zurueck( + self, mock_fd, mock_zf, mock_find, mock_vl, mock_mem + ): + mock_fd.getOpenFileName.return_value = ("/tmp/vg.zip", "") + mock_zf.ZipFile.return_value.__enter__ = MagicMock(return_value=MagicMock()) + mock_zf.ZipFile.return_value.__exit__ = MagicMock(return_value=False) + valid_layer = MagicMock() + valid_layer.isValid.return_value = True + mock_vl.return_value = valid_layer + + mem_layer = _make_layer(name="Verfahrensgebiet", lid="MEM1") + mock_mem.return_value = mem_layer + + with patch("os.path.dirname", return_value="/tmp"), \ + patch("os.path.splitext", return_value=("/tmp/vg", ".zip")), \ + patch("os.path.basename", return_value="vg.zip"), \ + patch("os.path.join", return_value="/tmp/vg"): + result = lade_vg_aus_zip(None) + + self.assertIs(result, mem_layer) + + +# --------------------------------------------------------------------------- +# Tests: verfahrensgebiet_shape_komplett (Hauptworkflow) +# --------------------------------------------------------------------------- + +class TestVerfahrensgebietShapeKomplett(unittest.TestCase): + + # ------------------------------------------------------------------ # + # Abbruchpfade + # ------------------------------------------------------------------ # + + @patch(_PATCH_PROGRESS) + @patch(_PATCH_PROJECT) + @patch(_PATCH_LADE_ZIP) + def test_lade_zip_none_aborts(self, mock_lade, mock_proj, mock_prog): + mock_proj.instance.return_value = _DummyProject() + mock_lade.return_value = None + mock_prog.return_value = _make_progress_mock() + + enriched, flst, bror, status = verfahrensgebiet_shape_komplett(None) + + self.assertIsNone(enriched) + self.assertIsNone(flst) + self.assertIsNone(bror) + self.assertEqual(status, LoadStatus.NONE) + + @patch(_PATCH_PROGRESS) + @patch(_PATCH_PROJECT) + @patch(_PATCH_GEM) + @patch(_PATCH_LADE_ZIP) + def test_shape_vg_sofort_ins_projekt_nach_laden(self, mock_lade, mock_gem, mock_proj, mock_prog): + """Shape-VG wird unmittelbar nach dem Laden ins Projekt eingefügt.""" + proj = _DummyProject() + mock_proj.instance.return_value = proj + shape_vg = _make_layer(name="Verfahrensgebiet", lid="S1") + mock_lade.return_value = shape_vg + mock_gem.return_value = [] + mock_prog.return_value = _make_progress_mock() + + with patch(_PATCH_FLST, return_value=(None, LoadStatus.NONE)): + verfahrensgebiet_shape_komplett(None) + + # Shape-VG muss ins Projekt eingefügt worden sein + self.assertIn(shape_vg, proj.added) + + @patch(_PATCH_PROGRESS) + @patch(_PATCH_PROJECT) + @patch(_PATCH_GEM) + @patch(_PATCH_LADE_ZIP) + def test_shape_vg_wird_bei_abbruch_aus_projekt_entfernt( + self, mock_lade, mock_gem, mock_proj, mock_prog + ): + """Shape-VG wird bei Fehler/Abbruch aus dem Projekt entfernt.""" + proj = _DummyProject() + mock_proj.instance.return_value = proj + shape_vg = _make_layer(name="Verfahrensgebiet", lid="S1") + mock_lade.return_value = shape_vg + mock_gem.return_value = [] + mock_prog.return_value = _make_progress_mock() + + with patch(_PATCH_FLST, return_value=(None, LoadStatus.NONE)): + _, _, _, status = verfahrensgebiet_shape_komplett(None) + + self.assertEqual(status, LoadStatus.NONE) + # Layer-ID muss in removed sein + self.assertIn("S1", proj.removed) + + @patch(_PATCH_PROGRESS) + @patch(_PATCH_PROJECT) + @patch(_PATCH_GEM) + @patch(_PATCH_LADE_ZIP) + def test_shape_vg_bleibt_im_projekt_bei_erfolg(self, mock_lade, mock_gem, mock_proj, mock_prog): + """Bei erfolgreichem Abschluss wird shape_vg NICHT aus dem Projekt entfernt.""" + proj = _DummyProject() + mock_proj.instance.return_value = proj + shape_vg = _make_layer(name="Verfahrensgebiet", lid="S1") + mock_lade.return_value = shape_vg + mock_gem.return_value = [] + flst_layer = _make_layer(name="Flurstücke", lid="F1") + mock_prog.return_value = _make_progress_mock() + + with patch(_PATCH_FLST, return_value=(flst_layer, LoadStatus.FIRST)), \ + patch(_PATCH_SELECT, return_value=5), \ + patch(_PATCH_ERSTELLE, return_value=(_make_layer(lid="T1"), 5, 0)), \ + patch(_PATCH_VERGLEICHE, return_value=(1000.0, 1000.0, True)), \ + patch(_PATCH_SHOW_INFO): + verfahrensgebiet_shape_komplett(None) + + # Layer-ID darf NICHT in removed sein + self.assertNotIn("S1", proj.removed) + + @patch(_PATCH_PROGRESS) + @patch(_PATCH_PROJECT) + @patch(_PATCH_GEM) + @patch(_PATCH_LADE_ZIP) + def test_flst_none_aborts(self, mock_lade, mock_gem, mock_proj, mock_prog): + mock_proj.instance.return_value = _DummyProject() + mock_lade.return_value = _make_layer(name="Verfahrensgebiet") + mock_gem.return_value = [("1234", "Muster")] + mock_prog.return_value = _make_progress_mock() + + with patch(_PATCH_FLST, return_value=(None, LoadStatus.NONE)): + enriched, flst, bror, status = verfahrensgebiet_shape_komplett(None) + + self.assertIsNone(enriched) + self.assertIsNone(flst) + self.assertEqual(status, LoadStatus.NONE) + + @patch(_PATCH_PROGRESS) + @patch(_PATCH_MSGBOX) + @patch(_PATCH_SELECT) + @patch(_PATCH_PROJECT) + @patch(_PATCH_GEM) + @patch(_PATCH_FLST) + @patch(_PATCH_LADE_ZIP) + def test_keine_flurstuecke_selektiert_gibt_warning( + self, mock_lade, mock_flst, mock_gem, mock_proj, mock_select, mock_mb, mock_prog + ): + mock_proj.instance.return_value = _DummyProject() + mock_lade.return_value = _make_layer(name="Verfahrensgebiet") + mock_gem.return_value = [] + flst_layer = _make_layer(name="Flurstücke") + mock_flst.return_value = (flst_layer, LoadStatus.FIRST) + mock_select.return_value = 0 + mock_prog.return_value = _make_progress_mock() + + enriched, flst, bror, status = verfahrensgebiet_shape_komplett(None) + + self.assertIsNone(enriched) + self.assertIs(flst, flst_layer) + self.assertIsNone(bror) + self.assertEqual(status, LoadStatus.NONE) + mock_mb.warning.assert_called_once() + + # ------------------------------------------------------------------ # + # Flächenübereinstimmung + # ------------------------------------------------------------------ # + + @patch(_PATCH_SHOW_INFO) + @patch(_PATCH_PROGRESS) + @patch(_PATCH_VERGLEICHE) + @patch(_PATCH_ERSTELLE) + @patch(_PATCH_SELECT) + @patch(_PATCH_PROJECT) + @patch(_PATCH_GEM) + @patch(_PATCH_FLST) + @patch(_PATCH_LADE_ZIP) + def test_flaechen_uebereinstimmung_gibt_info_und_kein_dialog( + self, mock_lade, mock_flst, mock_gem, mock_proj, + mock_select, mock_erstelle, mock_vergleiche, mock_prog, mock_info + ): + proj = _DummyProject() + mock_proj.instance.return_value = proj + shape_vg = _make_layer(name="Verfahrensgebiet", lid="S1") + mock_lade.return_value = shape_vg + mock_gem.return_value = [("1234", "Muster")] + flst_layer = _make_layer(name="Flurstücke", lid="F1") + mock_flst.return_value = (flst_layer, LoadStatus.FIRST) + mock_select.return_value = 5 + temp_enriched = _make_layer(name="Verfahrensgebiet", lid="T1") + mock_erstelle.return_value = (temp_enriched, 5, 0) + mock_vergleiche.return_value = (1000.0, 1000.0, True) + mock_prog.return_value = _make_progress_mock() + + enriched, flst, bror, status = verfahrensgebiet_shape_komplett(None) + + self.assertIs(enriched, shape_vg) + self.assertIs(flst, flst_layer) + self.assertIsNone(bror) + self.assertEqual(status, LoadStatus.FIRST) + mock_info.assert_called_once() + + # ------------------------------------------------------------------ # + # Flächenabweichung + Abbrechen + # ------------------------------------------------------------------ # + + @patch(_PATCH_ASK_DIALOG) + @patch(_PATCH_SHOW_INFO) + @patch(_PATCH_PROGRESS) + @patch(_PATCH_VERGLEICHE) + @patch(_PATCH_ERSTELLE) + @patch(_PATCH_SELECT) + @patch(_PATCH_PROJECT) + @patch(_PATCH_GEM) + @patch(_PATCH_FLST) + @patch(_PATCH_LADE_ZIP) + def test_abweichung_abbrechen_gibt_shape_vg_zurueck( + self, mock_lade, mock_flst, mock_gem, mock_proj, + mock_select, mock_erstelle, mock_vergleiche, mock_prog, mock_info, mock_ask + ): + proj = _DummyProject() + mock_proj.instance.return_value = proj + shape_vg = _make_layer(name="Verfahrensgebiet", lid="S1") + mock_lade.return_value = shape_vg + mock_gem.return_value = [] + flst_layer = _make_layer(name="Flurstücke", lid="F1") + mock_flst.return_value = (flst_layer, LoadStatus.FIRST) + mock_select.return_value = 5 + temp_enriched = _make_layer(name="Verfahrensgebiet", lid="T1") + mock_erstelle.return_value = (temp_enriched, 5, 0) + mock_vergleiche.return_value = (1000.0, 800.0, False) + mock_prog.return_value = _make_progress_mock() + mock_ask.return_value = "abbrechen" + + enriched, flst, bror, status = verfahrensgebiet_shape_komplett(None) + + self.assertIs(enriched, shape_vg) + self.assertIs(flst, flst_layer) + self.assertIsNone(bror) + self.assertEqual(status, LoadStatus.FIRST) + mock_info.assert_not_called() + + # ------------------------------------------------------------------ # + # Detailprüfung + # ------------------------------------------------------------------ # + + @patch(_PATCH_SHOW_INFO) + @patch(_PATCH_KNICK) + @patch(_PATCH_GP_ALKIS) + @patch(_PATCH_EXTRAHIERE) + @patch(_PATCH_ASK_DIALOG) + @patch(_PATCH_PROGRESS) + @patch(_PATCH_VERGLEICHE) + @patch(_PATCH_ERSTELLE) + @patch(_PATCH_SELECT) + @patch(_PATCH_PROJECT) + @patch(_PATCH_GEM) + @patch(_PATCH_FLST) + @patch(_PATCH_LADE_ZIP) + def test_abweichung_detailpruefung_ruft_knickpunkt_pruefung_auf( + self, mock_lade, mock_flst, mock_gem, mock_proj, + mock_select, mock_erstelle, mock_vergleiche, mock_prog, + mock_ask, mock_extrahiere, mock_gp, mock_knick, mock_info + ): + proj = _DummyProject() + mock_proj.instance.return_value = proj + shape_vg = _make_layer(name="Verfahrensgebiet", lid="S1") + mock_lade.return_value = shape_vg + mock_gem.return_value = [("1234", "Muster")] + flst_layer = _make_layer(name="Flurstücke", lid="F1") + mock_flst.return_value = (flst_layer, LoadStatus.FIRST) + mock_select.return_value = 5 + temp_enriched = _make_layer(name="Verfahrensgebiet", lid="T1") + mock_erstelle.return_value = (temp_enriched, 5, 0) + mock_vergleiche.return_value = (1000.0, 800.0, False) + mock_prog.return_value = _make_progress_mock() + mock_ask.return_value = "detailpruefung" + + knickpunkte_mock = [MagicMock() for _ in range(3)] + mock_extrahiere.return_value = knickpunkte_mock + + pta_layer = _make_layer(name="PunktortTA", lid="P1") + mock_gp.return_value = pta_layer + + ergebnis = _make_layer(name="Knickpunkte", lid="K1") + mock_knick.return_value = (ergebnis, 2, 1, 0) + + enriched, flst, bror, status = verfahrensgebiet_shape_komplett(None) + + mock_gp.assert_called_once() + mock_knick.assert_called_once() + self.assertIs(enriched, shape_vg) + self.assertEqual(status, LoadStatus.FIRST) + + @patch(_PATCH_SHOW_INFO) + @patch(_PATCH_KNICK) + @patch(_PATCH_GP_ALKIS, return_value=None) + @patch(_PATCH_EXTRAHIERE) + @patch(_PATCH_ASK_DIALOG) + @patch(_PATCH_PROGRESS) + @patch(_PATCH_VERGLEICHE) + @patch(_PATCH_ERSTELLE) + @patch(_PATCH_SELECT) + @patch(_PATCH_PROJECT) + @patch(_PATCH_GEM) + @patch(_PATCH_FLST) + @patch(_PATCH_LADE_ZIP) + def test_punktort_ta_none_gibt_shape_vg_zurueck_und_zeigt_fehler( + self, mock_lade, mock_flst, mock_gem, mock_proj, + mock_select, mock_erstelle, mock_vergleiche, mock_prog, + mock_ask, mock_extrahiere, mock_gp, mock_knick, mock_info + ): + proj = _DummyProject() + mock_proj.instance.return_value = proj + shape_vg = _make_layer(name="Verfahrensgebiet", lid="S1") + mock_lade.return_value = shape_vg + mock_gem.return_value = [] + flst_layer = _make_layer(name="Flurstücke", lid="F1") + mock_flst.return_value = (flst_layer, LoadStatus.FIRST) + mock_select.return_value = 5 + temp_enriched = _make_layer(name="Verfahrensgebiet", lid="T1") + mock_erstelle.return_value = (temp_enriched, 5, 0) + mock_vergleiche.return_value = (1000.0, 800.0, False) + mock_prog.return_value = _make_progress_mock() + mock_ask.return_value = "detailpruefung" + mock_extrahiere.return_value = [MagicMock()] + + with patch(f"{_MOD}.QMessageBox") as mock_mb: + enriched, flst, bror, status = verfahrensgebiet_shape_komplett(None) + + mock_knick.assert_not_called() + self.assertIs(enriched, shape_vg) + mock_mb.critical.assert_called_once() + + # ------------------------------------------------------------------ # + # Cleanup: TempVerfahrensgebiet + # ------------------------------------------------------------------ # + + @patch(_PATCH_PROGRESS) + @patch(_PATCH_GEM) + @patch(_PATCH_PROJECT) + @patch(_PATCH_LADE_ZIP) + def test_temp_layer_wird_stets_entfernt(self, mock_lade, mock_proj, mock_gem, mock_prog): + proj = _DummyProject() + temp_layer = _make_layer(name="TempVerfahrensgebiet", lid="TEMP") + proj._layers_by_name["TempVerfahrensgebiet"] = [temp_layer] + mock_proj.instance.return_value = proj + mock_lade.return_value = None # → sofortiger Abbruch nach Schritt 1 + mock_gem.return_value = [] + mock_prog.return_value = _make_progress_mock() + + verfahrensgebiet_shape_komplett(None) + + self.assertIn("TEMP", proj.removed) + + # ------------------------------------------------------------------ # + # Rückgabe-Konsistenz + # ------------------------------------------------------------------ # + + @patch(_PATCH_SHOW_INFO) + @patch(_PATCH_PROGRESS) + @patch(_PATCH_VERGLEICHE) + @patch(_PATCH_ERSTELLE) + @patch(_PATCH_SELECT) + @patch(_PATCH_PROJECT) + @patch(_PATCH_GEM) + @patch(_PATCH_FLST) + @patch(_PATCH_LADE_ZIP) + def test_bror_layer_ist_stets_none( + self, mock_lade, mock_flst, mock_gem, mock_proj, + mock_select, mock_erstelle, mock_vergleiche, mock_prog, mock_info + ): + """Dritter Rückgabewert ist im Shape-Workflow immer None (kein BROR).""" + mock_proj.instance.return_value = _DummyProject() + shape_vg = _make_layer(name="Verfahrensgebiet", lid="S1") + mock_lade.return_value = shape_vg + mock_gem.return_value = [] + flst_layer = _make_layer(name="Flurstücke", lid="F1") + mock_flst.return_value = (flst_layer, LoadStatus.FIRST) + mock_select.return_value = 5 + temp_enriched = _make_layer(lid="T1") + mock_erstelle.return_value = (temp_enriched, 5, 0) + mock_vergleiche.return_value = (1000.0, 1000.0, True) + mock_prog.return_value = _make_progress_mock() + + _, _, bror, _ = verfahrensgebiet_shape_komplett(None) + self.assertIsNone(bror) + + # ------------------------------------------------------------------ # + # BROR-Ladelogik + # ------------------------------------------------------------------ # + + @patch(_PATCH_SHOW_INFO) + @patch(_PATCH_PROGRESS) + @patch(_PATCH_VERGLEICHE) + @patch(_PATCH_SELECT) + @patch(_PATCH_PROJECT) + @patch(_PATCH_GEM) + @patch(_PATCH_FLST) + @patch(f"{_MOD}.verfahrensgebiet_alkis") + @patch(f"{_MOD}.get_variable", return_value="VF-2026-001") + @patch(_PATCH_LADE_ZIP) + def test_bror_geladen_und_zurueckgegeben_wenn_vkz_gesetzt( + self, mock_lade, mock_get_var, mock_vg_alkis, mock_flst, + mock_gem, mock_proj, mock_select, mock_vergleiche, mock_prog, mock_info + ): + """VKZ gesetzt → verfahrensgebiet_alkis aufgerufen, bror_layer zurückgegeben.""" + proj = _DummyProject() + mock_proj.instance.return_value = proj + shape_vg = _make_layer(name="Verfahrensgebiet", lid="S1") + mock_lade.return_value = shape_vg + mock_gem.return_value = [] + flst_layer = _make_layer(name="Flurstücke", lid="F1") + mock_flst.return_value = (flst_layer, LoadStatus.FIRST) + mock_select.return_value = 5 + bror = _make_layer(name="Verfahrensgebiet", lid="BROR") + mock_vg_alkis.return_value = (bror, LoadStatus.FIRST, []) + mock_vergleiche.return_value = (1000.0, 1000.0, True) + mock_prog.return_value = _make_progress_mock() + + enriched, flst, bror_ret, status = verfahrensgebiet_shape_komplett(None) + + mock_vg_alkis.assert_called_once() + self.assertIs(bror_ret, bror) + self.assertIs(enriched, shape_vg) + self.assertEqual(status, LoadStatus.FIRST) + + @patch(_PATCH_SHOW_INFO) + @patch(_PATCH_PROGRESS) + @patch(_PATCH_VERGLEICHE) + @patch(_PATCH_SELECT) + @patch(_PATCH_PROJECT) + @patch(_PATCH_GEM) + @patch(_PATCH_FLST) + @patch(f"{_MOD}.verfahrensgebiet_alkis") + @patch(f"{_MOD}.get_variable", return_value="VF-2026-001") + @patch(_PATCH_LADE_ZIP) + def test_bror_wird_vor_shape_vg_ins_projekt_eingefuegt( + self, mock_lade, mock_get_var, mock_vg_alkis, mock_flst, + mock_gem, mock_proj, mock_select, mock_vergleiche, mock_prog, mock_info + ): + """BROR wird VOR shape_vg ins Projekt eingefügt (kein Namenskonflikt).""" + proj = _DummyProject() + mock_proj.instance.return_value = proj + shape_vg = _make_layer(name="Verfahrensgebiet", lid="S1") + mock_lade.return_value = shape_vg + mock_gem.return_value = [] + flst_layer = _make_layer(name="Flurstücke", lid="F1") + mock_flst.return_value = (flst_layer, LoadStatus.FIRST) + mock_select.return_value = 5 + bror = _make_layer(name="Verfahrensgebiet", lid="BROR") + mock_vg_alkis.return_value = (bror, LoadStatus.FIRST, []) + mock_vergleiche.return_value = (1000.0, 1000.0, True) + mock_prog.return_value = _make_progress_mock() + + verfahrensgebiet_shape_komplett(None) + + # BROR muss in added sein + self.assertIn(bror, proj.added) + # BROR muss vor shape_vg hinzugefügt werden + bror_idx = proj.added.index(bror) + shape_idx = proj.added.index(shape_vg) + self.assertLess(bror_idx, shape_idx, "BROR muss vor shape_vg eingefügt werden") + + @patch(_PATCH_PROGRESS) + @patch(_PATCH_VERGLEICHE) + @patch(_PATCH_SELECT) + @patch(_PATCH_PROJECT) + @patch(_PATCH_GEM) + @patch(_PATCH_FLST) + @patch(f"{_MOD}.verfahrensgebiet_alkis") + @patch(f"{_MOD}.get_variable", return_value="VF-2026-001") + @patch(_PATCH_LADE_ZIP) + def test_bror_wird_bei_abbruch_aus_projekt_entfernt( + self, mock_lade, mock_get_var, mock_vg_alkis, mock_flst, + mock_gem, mock_proj, mock_select, mock_vergleiche, mock_prog + ): + """BROR und shape_vg werden bei Fehler aus dem Projekt entfernt.""" + proj = _DummyProject() + mock_proj.instance.return_value = proj + shape_vg = _make_layer(name="Verfahrensgebiet", lid="S1") + mock_lade.return_value = shape_vg + mock_gem.return_value = [] + bror = _make_layer(name="Verfahrensgebiet", lid="BROR") + mock_vg_alkis.return_value = (bror, LoadStatus.FIRST, []) + mock_flst.return_value = (None, LoadStatus.NONE) # → Abbruch + mock_prog.return_value = _make_progress_mock() + + _, _, _, status = verfahrensgebiet_shape_komplett(None) + + self.assertEqual(status, LoadStatus.NONE) + self.assertIn("S1", proj.removed) + self.assertIn("BROR", proj.removed) + + @patch(_PATCH_PROGRESS) + @patch(_PATCH_PROJECT) + @patch(_PATCH_GEM) + @patch(f"{_MOD}.verfahrensgebiet_alkis") + @patch(f"{_MOD}.get_variable", return_value="") + @patch(_PATCH_LADE_ZIP) + def test_verfahrensgebiet_alkis_nicht_aufgerufen_wenn_vkz_leer( + self, mock_lade, mock_get_var, mock_vg_alkis, mock_gem, mock_proj, mock_prog + ): + """Leere VKZ → verfahrensgebiet_alkis wird NICHT aufgerufen.""" + mock_proj.instance.return_value = _DummyProject() + mock_lade.return_value = _make_layer(name="Verfahrensgebiet") + mock_gem.return_value = [] + mock_prog.return_value = _make_progress_mock() + + with patch(_PATCH_FLST, return_value=(None, LoadStatus.NONE)): + verfahrensgebiet_shape_komplett(None) + + mock_vg_alkis.assert_not_called() + + @patch(_PATCH_SHOW_INFO) + @patch(_PATCH_PROGRESS) + @patch(_PATCH_VERGLEICHE) + @patch(_PATCH_SELECT) + @patch(_PATCH_PROJECT) + @patch(_PATCH_GEM) + @patch(_PATCH_FLST) + @patch(f"{_MOD}.verfahrensgebiet_alkis") + @patch(f"{_MOD}.get_variable", return_value="VF-2026-001") + @patch(_PATCH_LADE_ZIP) + def test_bror_als_erstes_argument_beim_flaechenvergleich( + self, mock_lade, mock_get_var, mock_vg_alkis, mock_flst, + mock_gem, mock_proj, mock_select, mock_vergleiche, mock_prog, mock_info + ): + """Mit BROR: _vergleiche_flaechen wird (bror, shape_vg, ...) aufgerufen.""" + proj = _DummyProject() + mock_proj.instance.return_value = proj + shape_vg = _make_layer(name="Verfahrensgebiet", lid="S1") + mock_lade.return_value = shape_vg + mock_gem.return_value = [] + flst_layer = _make_layer(name="Flurstücke", lid="F1") + mock_flst.return_value = (flst_layer, LoadStatus.FIRST) + mock_select.return_value = 5 + bror = _make_layer(name="Verfahrensgebiet", lid="BROR") + mock_vg_alkis.return_value = (bror, LoadStatus.FIRST, []) + mock_vergleiche.return_value = (1000.0, 1000.0, True) + mock_prog.return_value = _make_progress_mock() + + verfahrensgebiet_shape_komplett(None) + + # Erster Arg = bror, zweiter Arg = shape_vg + call_args = mock_vergleiche.call_args[0] + self.assertIs(call_args[0], bror) + self.assertIs(call_args[1], shape_vg) + + @patch(_PATCH_SHOW_INFO) + @patch(_PATCH_PROGRESS) + @patch(_PATCH_VERGLEICHE) + @patch(_PATCH_ERSTELLE) + @patch(_PATCH_SELECT) + @patch(_PATCH_PROJECT) + @patch(_PATCH_GEM) + @patch(_PATCH_FLST) + @patch(f"{_MOD}.verfahrensgebiet_alkis", return_value=(None, LoadStatus.NONE, [])) + @patch(f"{_MOD}.get_variable", return_value="VF-2026-001") + @patch(_PATCH_LADE_ZIP) + def test_fallback_auf_temp_enriched_wenn_bror_ladeversagen( + self, mock_lade, mock_get_var, mock_vg_alkis, mock_flst, + mock_gem, mock_proj, mock_select, mock_erstelle, mock_vergleiche, mock_prog, mock_info + ): + """BROR-Ladeversagen → Fallback: _vergleiche_flaechen mit shape_vg vs temp_enriched.""" + proj = _DummyProject() + mock_proj.instance.return_value = proj + shape_vg = _make_layer(name="Verfahrensgebiet", lid="S1") + mock_lade.return_value = shape_vg + mock_gem.return_value = [] + flst_layer = _make_layer(name="Flurstücke", lid="F1") + mock_flst.return_value = (flst_layer, LoadStatus.FIRST) + mock_select.return_value = 5 + temp_enriched = _make_layer(lid="T1") + mock_erstelle.return_value = (temp_enriched, 5, 0) + mock_vergleiche.return_value = (1000.0, 1000.0, True) + mock_prog.return_value = _make_progress_mock() + + _, _, bror_ret, _ = verfahrensgebiet_shape_komplett(None) + + # kein BROR zurückgegeben + self.assertIsNone(bror_ret) + # Fallback: erste Arg = shape_vg + call_args = mock_vergleiche.call_args[0] + self.assertIs(call_args[0], shape_vg) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_working_tab.py b/tests/test_working_tab.py index edc774c..9a9e1c3 100644 --- a/tests/test_working_tab.py +++ b/tests/test_working_tab.py @@ -566,5 +566,262 @@ class TestWorkingTab(unittest.TestCase): self.assertTrue(tab.btn_gem_alkis.isEnabled()) # finally +# --------------------------------------------------------------------------- +# handle_verf_shape +# --------------------------------------------------------------------------- + +class TestHandleVerfShape(unittest.TestCase): + + def _make_tab(self): + tab = WorkingTab.__new__(WorkingTab) + tab.btn_verf_alkis = _DummyButton() + tab.btn_verf_shape = _DummyButton() + tab.btn_flurst_alkis = _DummyButton() + tab.btn_gem_alkis = _DummyButton() + tab.btn_grenz_alkis = _DummyButton() + tab.btn_verf_aus_auswahl = _DummyButton() + tab.btn_verf_aus_liste = _DummyButton() + tab.verf_layer_combo = _DummyCombo() + tab.haken_verf = types.SimpleNamespace(setText=lambda _t: None, setStyleSheet=lambda _s: None) + tab.haken_flurst = types.SimpleNamespace(setText=lambda _t: None, setStyleSheet=lambda _s: None) + tab.haken_gem = types.SimpleNamespace(setText=lambda _t: None, setStyleSheet=lambda _s: None) + tab.haken_grenz = types.SimpleNamespace(setText=lambda _t: None, setStyleSheet=lambda _s: None) + tab.setze_haken = lambda *_args, **_kwargs: None + tab.pruefmanager = types.SimpleNamespace(verarbeite=lambda ergebnis: ergebnis) + return tab + + @patch("sn_verfahrensgebiet.ui.tabs.working_tab.set_verfahrensgebiet_layer_id") + @patch( + "sn_verfahrensgebiet.ui.tabs.working_tab.verfahrensgebiet_shape_komplett", + return_value=(None, None, None, LoadStatus.NONE), + ) + def test_status_none_clears_layer_variable(self, _mock_shape, mock_set_id): + """Status NONE → Layer-Variable wird auf '' gesetzt.""" + tab = self._make_tab() + + with patch("sn_verfahrensgebiet.ui.tabs.working_tab.get_variable", return_value="VF-123"), \ + patch("sn_verfahrensgebiet.ui.tabs.working_tab.ask_yes_no"): + tab.handle_verf_shape() + + mock_set_id.assert_called_once_with("") + + @patch("sn_verfahrensgebiet.ui.tabs.working_tab.set_verfahrensgebiet_layer_id") + @patch( + "sn_verfahrensgebiet.ui.tabs.working_tab.verfahrensgebiet_shape_komplett", + return_value=(None, None, None, LoadStatus.NONE), + ) + def test_button_always_re_enabled_on_failure(self, _mock_shape, _mock_set_id): + """Button muss nach Status NONE wieder freigegeben sein (finally).""" + tab = self._make_tab() + + with patch("sn_verfahrensgebiet.ui.tabs.working_tab.get_variable", return_value="VF-123"), \ + patch("sn_verfahrensgebiet.ui.tabs.working_tab.ask_yes_no"): + tab.handle_verf_shape() + + self.assertTrue(tab.btn_verf_shape.isEnabled()) + self.assertEqual(tab.btn_verf_shape.text(), "Aus Shape laden") + + @patch("sn_verfahrensgebiet.ui.tabs.working_tab.ask_yes_no", return_value=False) + @patch("sn_verfahrensgebiet.ui.tabs.working_tab.get_variable", return_value="") + @patch("sn_verfahrensgebiet.ui.tabs.working_tab.verfahrensgebiet_shape_komplett") + def test_verfahrensnummer_nicht_gesetzt_nein_bricht_ab( + self, mock_shape, mock_get_var, mock_ask + ): + """Verfahrensnummer leer + Nutzer wählt 'Nein' → Vorgang abgebrochen.""" + tab = self._make_tab() + + tab.handle_verf_shape() + + mock_ask.assert_called_once() + mock_shape.assert_not_called() + # Button darf nicht busy geblieben sein + self.assertTrue(tab.btn_verf_shape.isEnabled()) + + @patch("sn_verfahrensgebiet.ui.tabs.working_tab.set_verfahrensgebiet_layer_id") + @patch("sn_verfahrensgebiet.ui.tabs.working_tab.ask_yes_no", return_value=True) + @patch("sn_verfahrensgebiet.ui.tabs.working_tab.get_variable", return_value="") + @patch( + "sn_verfahrensgebiet.ui.tabs.working_tab.verfahrensgebiet_shape_komplett", + return_value=(None, None, None, LoadStatus.NONE), + ) + def test_verfahrensnummer_nicht_gesetzt_ja_fuehrt_fort( + self, mock_shape, mock_get_var, mock_ask, mock_set_id + ): + """Verfahrensnummer leer + Nutzer wählt 'Ja' → Pipeline wird aufgerufen.""" + tab = self._make_tab() + + tab.handle_verf_shape() + + mock_ask.assert_called_once() + mock_shape.assert_called_once_with(tab) + + @patch("sn_verfahrensgebiet.ui.tabs.working_tab.set_verfahrensgebiet_layer_id") + @patch("sn_verfahrensgebiet.ui.tabs.working_tab.ask_yes_no") + @patch("sn_verfahrensgebiet.ui.tabs.working_tab.get_variable", return_value="VF-2026-001") + @patch( + "sn_verfahrensgebiet.ui.tabs.working_tab.verfahrensgebiet_shape_komplett", + return_value=(None, None, None, LoadStatus.NONE), + ) + def test_verfahrensnummer_gesetzt_kein_rueckfrage_dialog( + self, mock_shape, mock_get_var, mock_ask, mock_set_id + ): + """Verfahrensnummer gesetzt → kein ask_yes_no-Dialog.""" + tab = self._make_tab() + + tab.handle_verf_shape() + + mock_ask.assert_not_called() + mock_shape.assert_called_once_with(tab) + + @patch("sn_verfahrensgebiet.ui.tabs.working_tab.get_variable", return_value="VF-123") + @patch("sn_verfahrensgebiet.ui.tabs.working_tab.success") + @patch("sn_verfahrensgebiet.ui.tabs.working_tab.apply_style") + @patch("sn_verfahrensgebiet.ui.tabs.working_tab.QgsProject.instance") + @patch("sn_verfahrensgebiet.ui.tabs.working_tab.verfahrensgebiet_shape_komplett") + @patch("sn_verfahrensgebiet.ui.tabs.working_tab.iface") + @patch("sn_verfahrensgebiet.ui.tabs.working_tab.set_verfahrensgebiet_layer_id") + def test_success_path_applies_style_and_sets_haken( + self, + mock_set_layer_id, + mock_iface, + mock_shape_komplett, + mock_project_instance, + mock_apply_style, + mock_success, + mock_get_var, + ): + """Erfolg → verfahrensgebiet.qml angewendet, haken_verf gesetzt, success aufgerufen.""" + tab = self._make_tab() + haken_calls = [] + tab.setze_haken = lambda lbl, v: haken_calls.append((lbl, v)) + + enriched = _DummyLayer(valid=True, layer_id="shape_vg") + flst = _DummyLayer(valid=True, layer_id="flst_ok") + mock_shape_komplett.return_value = (enriched, flst, None, LoadStatus.FIRST) + proj = _DummyProject() + proj.mapLayersByName = lambda _n: [] + proj.mapLayer = lambda _lid: None + mock_project_instance.return_value = proj + mock_iface.mapCanvas.return_value = _DummyCanvas() + + tab.handle_verf_shape() + + # Style auf enriched layer angewendet + apply_calls = [str(c) for c in mock_apply_style.call_args_list] + self.assertTrue( + any("verfahrensgebiet.qml" in c for c in apply_calls), + "verfahrensgebiet.qml muss angewendet werden", + ) + # Flurstücke-Style angewendet + self.assertTrue( + any("GIS_Flst_Beschriftung_ALKIS_NAS.qml" in c for c in apply_calls), + "Flurstücke-Style muss angewendet werden", + ) + # Haken gesetzt + self.assertTrue(any(v for _, v in haken_calls), "Mindestens ein Haken muss gesetzt werden") + # Layer-ID gespeichert + mock_set_layer_id.assert_called_with("shape_vg") + # Success-Meldung + mock_success.assert_called_once() + # Button wieder freigegeben + self.assertTrue(tab.btn_verf_shape.isEnabled()) + + @patch("sn_verfahrensgebiet.ui.tabs.working_tab.get_variable", return_value="VF-123") + @patch("sn_verfahrensgebiet.ui.tabs.working_tab.success") + @patch("sn_verfahrensgebiet.ui.tabs.working_tab.apply_style") + @patch("sn_verfahrensgebiet.ui.tabs.working_tab.QgsProject.instance") + @patch("sn_verfahrensgebiet.ui.tabs.working_tab.verfahrensgebiet_shape_komplett") + @patch("sn_verfahrensgebiet.ui.tabs.working_tab.iface") + @patch("sn_verfahrensgebiet.ui.tabs.working_tab.set_verfahrensgebiet_layer_id") + def test_bror_layer_erhaelt_qml_stil( + self, + mock_set_layer_id, + mock_iface, + mock_shape_komplett, + mock_project_instance, + mock_apply_style, + mock_success, + mock_get_var, + ): + """Wenn bror_layer zurückgegeben → BauRaumOderBodenordnungsrecht.qml angewendet.""" + tab = self._make_tab() + tab.setze_haken = lambda *a, **kw: None + + enriched = _DummyLayer(valid=True, layer_id="shape_vg") + bror = _DummyLayer(valid=True, layer_id="bror_id") + mock_shape_komplett.return_value = (enriched, None, bror, LoadStatus.FIRST) + + proj = _DummyProject() + proj.mapLayersByName = lambda _n: [] + proj.mapLayer = lambda lid: enriched if lid == "shape_vg" else None + mock_project_instance.return_value = proj + mock_iface.mapCanvas.return_value = _DummyCanvas() + + tab.handle_verf_shape() + + apply_calls = [str(c) for c in mock_apply_style.call_args_list] + self.assertTrue( + any("BauRaumOderBodenordnungsrecht.qml" in c for c in apply_calls), + "BauRaumOderBodenordnungsrecht.qml muss auf bror_layer angewendet werden", + ) + + @patch("sn_verfahrensgebiet.ui.tabs.working_tab.get_variable", return_value="VF-123") + @patch("sn_verfahrensgebiet.ui.tabs.working_tab.apply_style") + @patch("sn_verfahrensgebiet.ui.tabs.working_tab.QgsProject.instance") + @patch("sn_verfahrensgebiet.ui.tabs.working_tab.verfahrensgebiet_shape_komplett") + @patch("sn_verfahrensgebiet.ui.tabs.working_tab.iface") + @patch("sn_verfahrensgebiet.ui.tabs.working_tab.set_verfahrensgebiet_layer_id") + def test_enriched_layer_already_in_project_not_added_again( + self, + mock_set_layer_id, + mock_iface, + mock_shape_komplett, + mock_project_instance, + mock_apply_style, + mock_get_var, + ): + """Layer bereits im Projekt → kein erneutes addMapLayer für VG-Layer.""" + tab = self._make_tab() + tab.setze_haken = lambda *a, **kw: None + + enriched = _DummyLayer(valid=True, layer_id="shape_vg") + mock_shape_komplett.return_value = (enriched, None, None, LoadStatus.FIRST) + + proj = _DummyProject() + proj.mapLayersByName = lambda _n: [] + proj.mapLayer = lambda lid: enriched if lid == "shape_vg" else None + mock_project_instance.return_value = proj + mock_iface.mapCanvas.return_value = _DummyCanvas() + + tab.handle_verf_shape() + + # Enriched-Layer darf NICHT erneut via addMapLayer hinzugefügt worden sein + self.assertNotIn(enriched, proj.added) + + def test_busy_state_set_during_execution(self): + """Button wird während der Ausführung auf busy gesetzt.""" + tab = self._make_tab() + busy_states = [] + + original_set_busy = WorkingTab._set_busy + + def capture_busy(self_ref, button, busy, original_text="Aus Shape laden"): + busy_states.append(busy) + original_set_busy(self_ref, button, busy, original_text) + + with patch.object(WorkingTab, "_set_busy", capture_busy), \ + patch( + "sn_verfahrensgebiet.ui.tabs.working_tab.verfahrensgebiet_shape_komplett", + return_value=(None, None, None, LoadStatus.NONE), + ), \ + patch("sn_verfahrensgebiet.ui.tabs.working_tab.set_verfahrensgebiet_layer_id"), \ + patch("sn_verfahrensgebiet.ui.tabs.working_tab.get_variable", return_value="VF-123"): + tab.handle_verf_shape() + + self.assertIn(True, busy_states, "Button muss mindestens einmal auf busy gesetzt werden") + self.assertIn(False, busy_states, "Button muss am Ende wieder freigegeben werden") + + if __name__ == "__main__": unittest.main() + diff --git a/ui/tabs/working_tab.py b/ui/tabs/working_tab.py index 13e5952..813a10f 100644 --- a/ui/tabs/working_tab.py +++ b/ui/tabs/working_tab.py @@ -8,7 +8,8 @@ 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, create_progress_dialog +from sn_basis.functions.dialog_wrapper import show_info_dialog, create_progress_dialog, ask_yes_no +from sn_basis.functions.variable_wrapper import get_variable from sn_basis.modules.Pruefmanager import Pruefmanager from sn_basis.modules.pruef_ergebnis import pruef_ergebnis from qgis.utils import iface @@ -21,6 +22,7 @@ from sn_verfahrensgebiet.functions.flurstueck_alkis import ( FLST_LAYER_NAME, ) from sn_verfahrensgebiet.functions.verfahrensgebiet_alkis_komplett import verfahrensgebiet_alkis_komplett +from sn_verfahrensgebiet.functions.verfahrensgebiet_shape_komplett import verfahrensgebiet_shape_komplett 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, @@ -167,7 +169,7 @@ class WorkingTab(QWidget): 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) + self.btn_verf_shape.clicked.connect(self.handle_verf_shape) # self.btn_flurst_shape.clicked.connect(self.handle_flurst_shape) # Beim Init: aktuelle Verfahrensgebiet-Layer-ID laden und im Dropdown anzeigen @@ -220,6 +222,62 @@ class WorkingTab(QWidget): finally: self._set_busy(self.btn_verf_alkis, False, "Aus ALKIS laden") + def handle_verf_shape(self) -> None: + """Lädt ein Verfahrensgebiet aus einem ZIP-Archiv mit Shapefile-Set. + + Delegiert an :func:`~sn_verfahrensgebiet.functions.verfahrensgebiet_shape_komplett.verfahrensgebiet_shape_komplett`. + Nach Erfolg werden Shape-VG-Layer und Flurstücks-Layer ins Projekt eingefügt, + Styles angewendet und die UI-Elemente aktualisiert. + """ + # Vorprüfung: Verfahrensnummer muss gesetzt sein + vkz = get_variable("verfahrensnummer") or "" + if not str(vkz).strip(): + if not ask_yes_no( + "Verfahrensnummer nicht gesetzt", + "Verfahrensnummer nicht gesetzt. Trotzdem fortfahren?", + default=False, + parent=self, + ): + return + + self._set_busy(self.btn_verf_shape, True, "Aus Shape laden") + try: + enriched_layer, flst_layer, bror_layer, status = verfahrensgebiet_shape_komplett(self) + + if status == LoadStatus.NONE or not enriched_layer: + set_verfahrensgebiet_layer_id("") + return + + project = QgsProject.instance() + + # Flurstücks-Layer ins Projekt (sofern neu geladen) + if flst_layer and flst_layer.isValid(): + existing_flst = project.mapLayersByName(flst_layer.name()) + if not existing_flst: + project.addMapLayer(flst_layer) + apply_style(flst_layer, "GIS_Flst_Beschriftung_ALKIS_NAS.qml") + self.setze_haken(self.haken_flurst, True) + + # BROR-Layer: QML-Stil anwenden (sofern geladen) + if bror_layer and bror_layer.isValid(): + apply_style(bror_layer, "BauRaumOderBodenordnungsrecht.qml") + + # Shape-VG-Layer ins Projekt + 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() + self.setze_haken(self.haken_verf, True) + + self.verf_layer_combo.setLayer(enriched_layer) + self._on_verf_layer_changed(enriched_layer) + + success("Verfahrensgebiet", "Layer aus Shape geladen.", duration=5) + + finally: + self._set_busy(self.btn_verf_shape, False, "Aus Shape laden") + def _set_busy(self, button: QPushButton, busy: bool, original_text: str = "Aus ALKIS laden") -> None: """Setzt den Busy-State eines Buttons. From 6e0bd213de6bfc9767d1cac79cb637a2f0e02b2f Mon Sep 17 00:00:00 2001 From: daniel Date: Wed, 15 Jul 2026 15:39:21 +0200 Subject: [PATCH 7/7] Vergleich BROR-VG --- functions/verfahrensgebiet_alkis_komplett.py | 2 +- functions/verfahrensgebiet_shape_komplett.py | 120 ++++++++++++++++++- 2 files changed, 119 insertions(+), 3 deletions(-) diff --git a/functions/verfahrensgebiet_alkis_komplett.py b/functions/verfahrensgebiet_alkis_komplett.py index f069368..adca779 100644 --- a/functions/verfahrensgebiet_alkis_komplett.py +++ b/functions/verfahrensgebiet_alkis_komplett.py @@ -444,7 +444,7 @@ def verfahrensgebiet_alkis_komplett( "Abweichung erkannt", f"Abweichung zwischen BauRaumOderBodenordnungsrecht-Objekt und abgeleitetem " f"Verfahrensgebiet erkannt.\n\n" - f"BauRaumOderBodenordnungsrecht: {bror_flaeche:.2f} m²\n" + f"As: {bror_flaeche:.2f} m²\n" f"Verfahrensgebiet: {vg_flaeche:.2f} m²\n" f"Differenz: {diff:.2f} m²\n\n" f"Wie möchten Sie fortfahren?", diff --git a/functions/verfahrensgebiet_shape_komplett.py b/functions/verfahrensgebiet_shape_komplett.py index 31c1c51..db1bb1a 100644 --- a/functions/verfahrensgebiet_shape_komplett.py +++ b/functions/verfahrensgebiet_shape_komplett.py @@ -43,11 +43,11 @@ import os import zipfile from typing import Optional, Tuple -from sn_basis.functions import QgsProject, QgsVectorLayer, QgsFeature +from sn_basis.functions import QgsProject, QgsVectorLayer, QgsFeature, QgsField 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.qt_wrapper import QFileDialog, QVariant from sn_basis.functions.qgiscore_wrapper import QgsGeometry from sn_basis.functions.dialog_wrapper import ( create_progress_dialog, @@ -84,6 +84,117 @@ from sn_verfahrensgebiet.functions.verfahrensgebiet_alkis_komplett import ( # Interne Hilfsfunktionen # --------------------------------------------------------------------------- +def _wende_vergleich_stil_an(layer) -> None: + """Wendet einen kategorisierten Renderer auf den Vergleich-Layer an. + + Farben: + - ``identisch`` (Schnittmenge): Gr\u00fcn + - ``nur_bror`` (nur im BROR): Gelb + - ``nur_vg`` (nur im Verfahrensgebiet): Blau + + Schl\u00e4gt der Import der QGIS-Rendering-Klassen fehl (z.\u00a0B. im + Test-Modus), wird der Fehler still ignoriert. + """ + try: + from qgis.core import ( # type: ignore[import] + QgsCategorizedSymbolRenderer, + QgsRendererCategory, + QgsFillSymbol, + ) + + _FARBEN = [ + ("identisch", "#00aa00", 0.5), # gr\u00fcn + ("nur_bror", "#ffcc00", 0.5), # gelb + ("nur_vg", "#3366ff", 0.5), # blau + ] + + kategorien = [] + for wert, farbe, opazitaet in _FARBEN: + symbol = QgsFillSymbol.createSimple({ + "color": farbe, + "outline_color": "#555555", + "outline_width": "0.26", + }) + symbol.setOpacity(opazitaet) + kategorien.append(QgsRendererCategory(wert, symbol, wert)) + + renderer = QgsCategorizedSymbolRenderer("typ", kategorien) + layer.setRenderer(renderer) + except Exception: + pass + + +def _erstelle_vergleich_layer(vg_layer, bror_layer) -> Optional[object]: + """Erzeugt einen Vergleich-Layer aus geometrischen Differenzoperationen. + + F\u00fcr jeden Fl\u00e4chentyp wird ein Feature angelegt: + + - ``identisch``: Schnittmenge aus VG und BROR (gr\u00fcn). + - ``nur_bror``: Fl\u00e4che, die nur im BROR-Objekt liegt (gelb). + - ``nur_vg``: Fl\u00e4che, die nur im Verfahrensgebiet liegt (blau). + + Parameters + ---------- + vg_layer : + Verfahrensgebiet-Layer (Shape). + bror_layer : + BauRaumOderBodenordnungsrecht-Layer (ALKIS). + + Returns + ------- + Optional[QgsVectorLayer] + Memory-Layer ``Vergleich`` mit Stil oder ``None`` bei Fehler. + """ + try: + vg_geom = None + for feat in vg_layer.getFeatures(): + g = feat.geometry() + if g and not g.isEmpty(): + vg_geom = g + break + + bror_geom = None + for feat in bror_layer.getFeatures(): + g = feat.geometry() + if g and not g.isEmpty(): + bror_geom = g + break + + if not vg_geom or not bror_geom: + return None + + geom_identisch = vg_geom.intersection(bror_geom) + geom_nur_bror = bror_geom.difference(vg_geom) + geom_nur_vg = vg_geom.difference(bror_geom) + + crs = vg_layer.crs().authid() + mem_layer = QgsVectorLayer( + f"MultiPolygon?crs={crs}", "Vergleich", "memory" + ) + pr = mem_layer.dataProvider() + pr.addAttributes([QgsField("typ", QVariant.String)]) + mem_layer.updateFields() + + feats = [] + for geom, typ in [ + (geom_identisch, "identisch"), + (geom_nur_bror, "nur_bror"), + (geom_nur_vg, "nur_vg"), + ]: + if geom and not geom.isEmpty(): + f = QgsFeature() + f.setGeometry(geom) + f.setAttributes([typ]) + feats.append(f) + + pr.addFeatures(feats) + mem_layer.updateExtents() + _wende_vergleich_stil_an(mem_layer) + + return mem_layer if mem_layer.isValid() else None + except Exception: + return None + def _finde_shp(verzeichnis: str) -> Optional[str]: """Sucht rekursiv nach der ersten ``.shp``-Datei im Verzeichnis. @@ -443,6 +554,11 @@ def verfahrensgebiet_shape_komplett( dialog_ref_label = "BauRaumOderBodenordnungsrecht" dialog_vg_label = "Shape-Verfahrensgebiet " pruef_layer = bror_layer + + # Vergleich-Layer (grün/gelb/blau) ins Projekt einfügen + vergleich_layer = _erstelle_vergleich_layer(shape_vg_layer, bror_layer) + if vergleich_layer: + project.addMapLayer(vergleich_layer) else: # Kein BROR → temp. enriched Layer aus Flurstücken erzeugen temp_enriched, _, _ = erstelle_verfahrensgebiet_aus_auswahl(flst_layer)