forked from AG_QGIS/Plugin_SN_Basis
385 lines
11 KiB
Python
385 lines
11 KiB
Python
"""
|
||
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
|
||
|
||
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,
|
||
)
|
||
|
||
QgsProject = _QgsProject
|
||
QgsVectorLayer = _QgsVectorLayer
|
||
QgsRasterLayer = _QgsRasterLayer
|
||
QgsNetworkAccessManager = _QgsNetworkAccessManager
|
||
Qgis = _Qgis
|
||
QgsMapLayerProxyModel = _QgsMaplLayerProxyModel
|
||
QgsVectorFileWriter = _QgsVectorFileWriter
|
||
QgsFeature = _QgsFeature
|
||
QgsField = _QgsField
|
||
QgsGeometry = _QgsGeometry
|
||
|
||
QGIS_AVAILABLE = True
|
||
|
||
# ---------------------------------------------------------
|
||
# Mock-Modus
|
||
# ---------------------------------------------------------
|
||
|
||
except Exception:
|
||
QGIS_AVAILABLE = False
|
||
|
||
class _MockQgsProject:
|
||
def __init__(self):
|
||
self._variables = {}
|
||
|
||
@staticmethod
|
||
def instance() -> "_MockQgsProject":
|
||
return _MockQgsProject()
|
||
|
||
def read(self) -> bool:
|
||
return True
|
||
|
||
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
|
||
|
||
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
|