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__":