qgis-jvf-dtm-plugin/geometry_processor.py at main · MapGeeks/qgis-jvf-dtm-plugin · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
"""
@brief Geometry processing

Classes:
 - GeometryProcessor

(C) 2024-2025 by MapGeeks
@author Petr Barandovski petr.barandovski@gmail.com
@author Linda Karlovska linda.karlovska@seznam.cz

This program is free software: you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.
"""

import logging
import lxml.etree as ET
from typing import Dict, Tuple, Optional, Generator
from osgeo import ogr

from qgis.core import QgsGeometry, QgsWkbTypes

logger = logging.getLogger(__name__)


class GeometryProcessor:
    """Třída pro efektivní zpracování geometrií"""

    def __init__(self):
        self._geom_cache: Dict[str, Tuple[QgsGeometry, str, str]] = {}
        self.ns = {"gml": "http://www.opengis.net/gml/3.2", "x": "*"}

    @staticmethod
    def _get_geometry_type(qgs_geom: QgsGeometry) -> Optional[str]:
        """
        Určí typ geometrie z QgsGeometry.

        Args:
            qgs_geom: QGIS geometrie

        Returns:
            Textový popis typu geometrie nebo None
        """
        type_mapping = {
            QgsWkbTypes.Point: "Point",
            QgsWkbTypes.PointZ: "Point",
            QgsWkbTypes.MultiPoint: "Point",
            QgsWkbTypes.MultiPointZ: "Point",
            QgsWkbTypes.LineString: "LineString",
            QgsWkbTypes.LineStringZ: "LineString",
            QgsWkbTypes.MultiLineString: "LineString",
            QgsWkbTypes.MultiLineStringZ: "LineString",
            QgsWkbTypes.Polygon: "Polygon",
            QgsWkbTypes.PolygonZ: "Polygon",
            QgsWkbTypes.MultiPolygon: "Polygon",
            QgsWkbTypes.MultiPolygonZ: "Polygon",
        }

        return type_mapping.get(qgs_geom.wkbType())

    def _parse_geometry(
        self, geom_elem: ET.Element
    ) -> Optional[Tuple[QgsGeometry, str, str]]:
        """
        Parsuje XML element na QGIS geometrii.

        Args:
            geom_elem: XML element obsahující geometrii

        Returns:
            Tuple obsahující (QgsGeometry, gml_id, typ_geometrie) nebo None
        """
        try:
            gml_id = geom_elem.get(f'{{{self.ns["gml"]}}}id', "")
            gml_str = ET.tostring(geom_elem, encoding="unicode", with_tail=False)
            geom = ogr.CreateGeometryFromGML(gml_str)

            if geom and geom.IsValid():
                qgs_geom = QgsGeometry.fromWkt(geom.ExportToWkt())
                if qgs_geom:
                    geom_type = self._get_geometry_type(qgs_geom)
                    if geom_type:
                        return qgs_geom, gml_id, geom_type
        except Exception as e:
            logger.error(f"Error processing geometry: {e}")
        return None

    def process_geometries(
        self, element: ET.Element
    ) -> Generator[Tuple[QgsGeometry, str, str], None, None]:
        """
        Generátor pro zpracování geometrií - umožňuje postupné zpracování bez načtení všech do paměti.

        Args:
            element: XML element obsahující geometrie

        Yields:
            Tuple obsahující (QgsGeometry, gml_id, typ_geometrie)
        """
        if element is None:
            return

        geom_elems = element.findall(".//*[@gml:id]", self.ns)
        for geom_elem in geom_elems:
            gml_id = geom_elem.get(f'{{{self.ns["gml"]}}}id')

            # Pokud je geometrie již v cache, vrátí se z ní
            if gml_id in self._geom_cache:
                yield self._geom_cache[gml_id]
                continue

            # Zpracování geometrie
            result = self._parse_geometry(geom_elem)
            if result:
                self._geom_cache[gml_id] = result
                yield result