""" sn_basis/functions/qgiscore_wrapper.py – zentrale QGIS-Core-Abstraktion """ from typing import Type, Any, Optional from sn_basis.functions.qt_wrapper import ( QUrl, QEventLoop, QNetworkRequest, ) # --------------------------------------------------------- # QGIS-Symbole (werden dynamisch gesetzt) # --------------------------------------------------------- QgsProject: Type[Any] QgsVectorLayer: Type[Any] QgsRasterLayer: Type[Any] QgsNetworkAccessManager: Type[Any] Qgis: Type[Any] QgsMapLayerProxyModel: Type[Any] QgsVectorFileWriter: Type[Any] # neu: Schreib-API QgsFeature: Type[Any] QgsField: Type[Any] QgsGeometry: Type[Any] QgsFeatureRequest: Type[Any] QgsCoordinateTransform: Type[Any] QgsCoordinateReferenceSystem: Type[Any] QgsPrintLayout: Type[Any] QgsLayoutItemMap: Type[Any] QgsLayoutItemLabel: Type[Any] QgsLayoutPoint: Type[Any] QgsLayoutSize: Type[Any] QgsUnitTypes: Type[Any] QgsLayoutItem: Type[Any] QGIS_AVAILABLE = False # --------------------------------------------------------- # Versuch: QGIS-Core importieren # --------------------------------------------------------- try: from qgis.core import ( QgsProject as _QgsProject, QgsVectorLayer as _QgsVectorLayer, QgsRasterLayer as _QgsRasterLayer, QgsNetworkAccessManager as _QgsNetworkAccessManager, Qgis as _Qgis, QgsMapLayerProxyModel as _QgsMaplLayerProxyModel, QgsVectorFileWriter as _QgsVectorFileWriter, QgsFeature as _QgsFeature, QgsField as _QgsField, QgsGeometry as _QgsGeometry, QgsFeatureRequest as _QgsFeatureRequest, QgsCoordinateTransform as _QgsCoordinateTransform, QgsCoordinateReferenceSystem as _QgsCoordinateReferenceSystem, QgsPrintLayout as _QgsPrintLayout, QgsLayoutItemMap as _QgsLayoutItemMap, QgsLayoutItemLabel as _QgsLayoutItemLabel, QgsLayoutPoint as _QgsLayoutPoint, QgsLayoutSize as _QgsLayoutSize, QgsUnitTypes as _QgsUnitTypes, QgsLayoutItem as _QgsLayoutItem, ) QgsProject = _QgsProject QgsVectorLayer = _QgsVectorLayer QgsRasterLayer = _QgsRasterLayer QgsNetworkAccessManager = _QgsNetworkAccessManager Qgis = _Qgis QgsMapLayerProxyModel = _QgsMaplLayerProxyModel QgsVectorFileWriter = _QgsVectorFileWriter QgsFeature = _QgsFeature QgsField = _QgsField QgsGeometry = _QgsGeometry QgsFeatureRequest = _QgsFeatureRequest QgsCoordinateTransform = _QgsCoordinateTransform QgsCoordinateReferenceSystem = _QgsCoordinateReferenceSystem QgsPrintLayout = _QgsPrintLayout QgsLayoutItemMap = _QgsLayoutItemMap QgsLayoutItemLabel = _QgsLayoutItemLabel QgsLayoutPoint = _QgsLayoutPoint QgsLayoutSize = _QgsLayoutSize QgsUnitTypes = _QgsUnitTypes QgsLayoutItem = _QgsLayoutItem QGIS_AVAILABLE = True # --------------------------------------------------------- # Mock-Modus # --------------------------------------------------------- except Exception: QGIS_AVAILABLE = False class _MockLayoutManager: def layoutByName(self, name: str): return None def addLayout(self, layout: Any) -> bool: return True class _MockQgsProject: def __init__(self): self._variables = {} self._layout_manager = _MockLayoutManager() @staticmethod def instance() -> "_MockQgsProject": return _MockQgsProject() def read(self) -> bool: return True def layoutManager(self): return self._layout_manager QgsProject = _MockQgsProject class _MockQgsVectorLayer: def __init__(self, *args, **kwargs): self._valid = True def isValid(self) -> bool: return self._valid def loadNamedStyle(self, path: str): return True, "" def triggerRepaint(self) -> None: pass def dataProvider(self): return None QgsVectorLayer = _MockQgsVectorLayer class _MockQgsNetworkAccessManager: @staticmethod def instance(): return _MockQgsNetworkAccessManager() def head(self, request: Any): return None class _MockQgsRasterLayer: """ Minimaler Mock für QgsRasterLayer, ausreichend für Tests und um im Datenabruf ein Raster-Layer-Objekt im pruef_ergebnis kontext mitzugeben. """ def __init__(self, source: str, name: str = "Raster", provider: str = "wms"): self.source = source self._name = name self.provider = provider self._valid = True def isValid(self) -> bool: return self._valid def name(self) -> str: return self._name def dataProvider(self): return None QgsRasterLayer = _MockQgsRasterLayer class _MockQgsPrintLayout: def __init__(self, project: Any): self.project = project self._name = "" self._page = _MockQgsLayoutPage() def initializeDefaults(self) -> None: pass def setName(self, name: str) -> None: self._name = name def pageCollection(self): return self def page(self, index: int): return self._page def addLayoutItem(self, item: Any) -> None: pass class _MockQgsLayoutPage: def setPageSize(self, size: Any) -> None: self.size = size class _MockQgsLayoutItem: class ReferencePoint: LowerLeft = 0 class _MockQgsLayoutItemMap: def __init__(self, layout: Any): self.layout = layout def setId(self, item_id: str) -> None: pass def setExtent(self, extent: Any) -> None: pass def setScale(self, scale: float) -> None: pass def attemptMove(self, point: Any) -> None: pass def attemptResize(self, size: Any) -> None: pass def setFollowVisibilityPreset(self, active: bool) -> None: pass def setFollowVisibilityPresetName(self, name: str) -> None: pass class _MockQgsLayoutItemLabel: ModeHtml = 1 def __init__(self, layout: Any): self.layout = layout def setId(self, item_id: str) -> None: pass def setText(self, text: str) -> None: pass def setMode(self, mode: Any) -> None: pass def setFont(self, font: Any) -> None: pass def setReferencePoint(self, point: Any) -> None: pass def attemptMove(self, point: Any) -> None: pass def attemptResize(self, size: Any) -> None: pass class _MockQgsLayoutPoint: def __init__(self, x: float, y: float, unit: Any): self.x = x self.y = y self.unit = unit class _MockQgsLayoutSize: def __init__(self, width: float, height: float, unit: Any): self.width = width self.height = height self.unit = unit class _MockQgsUnitTypes: LayoutMillimeters = 0 QgsPrintLayout = _MockQgsPrintLayout QgsLayoutItemMap = _MockQgsLayoutItemMap QgsLayoutItemLabel = _MockQgsLayoutItemLabel QgsLayoutPoint = _MockQgsLayoutPoint QgsLayoutSize = _MockQgsLayoutSize QgsUnitTypes = _MockQgsUnitTypes QgsLayoutItem = _MockQgsLayoutItem class _MockQgsFeatureRequest: def __init__(self): self._filter_rect = None def setFilterRect(self, rect): self._filter_rect = rect return self QgsFeatureRequest = _MockQgsFeatureRequest class _MockQgsCoordinateTransform: def __init__(self, *args, **kwargs): pass def transformBoundingBox(self, rect): return rect class _MockQgsCoordinateReferenceSystem: def __init__(self, *args, **kwargs): pass QgsCoordinateTransform = _MockQgsCoordinateTransform QgsCoordinateReferenceSystem = _MockQgsCoordinateReferenceSystem QgsNetworkAccessManager = _MockQgsNetworkAccessManager class _MockQgis: class MessageLevel: Success = 0 Info = 1 Warning = 2 Critical = 3 Qgis = _MockQgis class _MockQgsMapLayerProxyModel: # Layer-Typen (entsprechen QGIS-Konstanten) NoLayer = 0 VectorLayer = 1 RasterLayer = 2 PluginLayer = 3 MeshLayer = 4 VectorTileLayer = 5 PointCloudLayer = 6 def __init__(self, *args, **kwargs): pass QgsMapLayerProxyModel = _MockQgsMapLayerProxyModel # --------------------------------------------------------- # Mock für QgsVectorFileWriter # --------------------------------------------------------- class _MockSaveVectorOptions: """ Minimaler Ersatz für QgsVectorFileWriter.SaveVectorOptions. Felder werden als einfache Attribute bereitgestellt. """ def __init__(self): self.driverName: str = "GPKG" self.layerName: Optional[str] = None self.fileEncoding: str = "UTF-8" # Action-Konstanten werden symbolisch verwendet self.actionOnExistingFile: Optional[int] = None class _MockQgsVectorFileWriter: """ Minimaler Mock für QgsVectorFileWriter mit der benötigten API: - SaveVectorOptions (als Klasse) - writeAsVectorFormatV3(layer, path, transformContext, options) -> error_code - NoError (Konstante) - CreateOrOverwriteFile / CreateOrOverwriteLayer (Konstanten) """ # Fehlerkonstanten (0 = NoError) NoError = 0 # Action-Konstanten (Werte nur symbolisch) CreateOrOverwriteFile = 1 CreateOrOverwriteLayer = 2 # SaveVectorOptions-Klasse SaveVectorOptions = _MockSaveVectorOptions @staticmethod def writeAsVectorFormatV3(layer: Any, path: str, transform_context: Any, options: Any) -> int: """ Mock-Schreibfunktion. Verhalten im Mock: - Wenn 'layer' None oder options.layerName fehlt, geben wir NoError zurück, aber schreiben nichts (Tests erwarten nur Rückgabecode). - Diese Implementierung versucht nicht, echte Dateien zu schreiben. - Rückgabewert: 0 (NoError) bei Erfolg, sonst eine positive Fehlernummer. """ try: # Sehr einfache Validierung: wenn path leer -> Fehler if not path: return 999 # Simuliere Erfolg return _MockQgsVectorFileWriter.NoError except Exception: return 999 # generischer Fehlercode QgsVectorFileWriter = _MockQgsVectorFileWriter # --------------------------------------------------------- # Netzwerk # --------------------------------------------------------- class NetworkReply: """ Minimaler Wrapper für Netzwerkantworten. """ def __init__(self, error: int): self.error = error def network_head(url: str) -> NetworkReply | None: """ Führt einen HTTP-HEAD-Request aus. Rückgabe: - NetworkReply(error=0) → erreichbar - NetworkReply(error!=0) → nicht erreichbar - None → Netzwerk nicht verfügbar / Fehler beim Request """ if not QGIS_AVAILABLE: return None if QUrl is None or QNetworkRequest is None: return None try: manager = QgsNetworkAccessManager.instance() request = QNetworkRequest(QUrl(url)) reply = manager.head(request) # synchron warten (kurz) if QEventLoop is not None: loop = QEventLoop() reply.finished.connect(loop.quit) loop.exec() return NetworkReply(error=reply.error()) except Exception: return None # --------------------------------------------------------- # Layer-Geometrie / Extent # --------------------------------------------------------- def get_layer_extent(layer: Any) -> Any: """ Gibt die Ausdehnung (Extent) eines Layers zurück. Diese Funktion kapselt den Zugriff auf ``layer.extent()`` und dient als zentrale Abstraktion für alle Stellen, die die Bounding Box eines Layers benötigen (z.B. für räumliche Filter im Datenabruf). Verhalten --------- - Wenn QGIS verfügbar ist und der Layer eine ``extent()``-Methode besitzt, wird deren Rückgabewert zurückgegeben. - Wenn QGIS nicht verfügbar ist oder der Layer keine ``extent()``-Methode hat, wird ``None`` zurückgegeben. """ if not QGIS_AVAILABLE or layer is None: return None extent_func = getattr(layer, "extent", None) if callable(extent_func): try: return extent_func() except Exception: return None return None # --------------------------------------------------------- # Buffer-Layer erzeugen # --------------------------------------------------------- def create_buffer_layer( source_layer: Any, distance_m: float, layer_name: str = "BufferLayer" ) -> Optional[Any]: """ Erzeugt einen Pufferlayer um alle Features eines Quelllayers. Diese Funktion dient als zentrale Abstraktion für die Erzeugung eines Pufferlayers in QGIS. Sie wird z.B. im Datenabruf verwendet, wenn der Raumfilter ``"Pufferlayer"`` aktiv ist. Verhalten --------- - Wenn QGIS verfügbar ist und der ``source_layer`` gültig ist, wird ein temporärer Vektorlayer erzeugt, der die gepufferten Geometrien enthält. - Der Puffer wird in Metern angegeben. - Der zurückgegebene Layer ist **nicht gespeichert**, sondern ein temporärer Speicherlayer, der anschließend über den UI‑Wrapper ins Projekt geladen werden kann. - Wenn QGIS nicht verfügbar ist oder ein Fehler auftritt, wird ``None`` zurückgegeben. """ if not QGIS_AVAILABLE: return None if source_layer is None or not hasattr(source_layer, "getFeatures"): return None try: # Geometrien puffern buffered_geoms = [] for feat in source_layer.getFeatures(): geom = feat.geometry() if geom is None: continue buf = geom.buffer(distance_m, 8) if buf is not None: buffered_geoms.append(buf) if not buffered_geoms: return None # Neuen Memory-Layer erzeugen crs = source_layer.crs().authid() if hasattr(source_layer, "crs") else "EPSG:4326" mem_layer = QgsVectorLayer(f"Polygon?crs={crs}", layer_name, "memory") prov = mem_layer.dataProvider() prov.addAttributes([]) mem_layer.updateFields() # Features hinzufügen from qgis.core import QgsFeature for geom in buffered_geoms: f = QgsFeature() f.setGeometry(geom) prov.addFeature(f) mem_layer.updateExtents() return mem_layer except Exception: return None #Hilfsfunktion, keine qgiscore-Entsprechung def layer_exists_in_gpkg(gpkg_path: str, layer_name: str) -> bool: """ Prüft, ob ein Layer mit dem Namen `layer_name` in `gpkg_path` existiert. - bevorzugt: SQLite-Abfrage auf gpkg_contents - fallback: kurzer Versuch, mit QgsVectorLayer zu laden (wenn QGIS verfügbar) """ import os, sqlite3 if not gpkg_path or not layer_name or not os.path.exists(gpkg_path): return False # 1) SQLite-Check (schnell) try: conn = sqlite3.connect(gpkg_path) cur = conn.cursor() cur.execute("SELECT COUNT(1) FROM gpkg_contents WHERE table_name = ?", (layer_name,)) row = cur.fetchone() conn.close() if row and row[0] > 0: return True except Exception: # falls sqlite fehlschlägt, weiter zum QGIS-Fallback pass # 2) QGIS-Fallback: versuche kurz, den Layer zu laden try: if getattr(QgsVectorLayer, "__call__", None) and QGIS_AVAILABLE: uri = f"{gpkg_path}|layername={layer_name}" layer = QgsVectorLayer(uri, layer_name, "ogr") return bool(layer and getattr(layer, "isValid", lambda: False)()) except Exception: pass return False