379 lines
14 KiB
Python
379 lines
14 KiB
Python
"""
|
||
sn_plan41/modules/vln_api_logic.py – Fachlogik für die VLN-API-Integration in Tab A.
|
||
|
||
Kapselt:
|
||
- Anmeldung / Abmeldung (KartenApiClient aus vln_karten)
|
||
- Persistierung des API-Keys in QSettings
|
||
- Verfahrensliste laden (GET /tgen)
|
||
- Persistierung der VKZ-Auswahl in der Projektdatei
|
||
- Plan 41 laden (2 Layer + Stile aus sn_plan41/assets/)
|
||
- Aktiven Layer hochladen
|
||
|
||
Voraussetzung: Das Plugin `vln_karten` muss installiert sein.
|
||
Wenn nicht vorhanden, liefert :attr:`is_available` False; die UI
|
||
kann dann einen entsprechenden Hinweis anzeigen.
|
||
|
||
Sitzungsablauf bei AuthError:
|
||
- :meth:`handle_session_expired` aufrufen → API-Key löschen + logout
|
||
- In der UI danach :meth:`_update_vln_ui_state` aufrufen und
|
||
QMessageBox.warning anzeigen. Kein automatischer Login-Dialog.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from typing import Optional
|
||
|
||
from sn_basis.functions.qt_wrapper import QSettings
|
||
from sn_basis.functions.qgiscore_wrapper import QgsProject
|
||
from sn_basis.functions.sys_wrapper import get_plugin_root, join_path
|
||
from sn_basis.functions.ly_style_wrapper import apply_style_from_path
|
||
|
||
# Lokale API-Module — fallenübrig, sobald QGIS verfügbar ist.
|
||
# In der reinen Python-Testumgebung (kein qgis-Paket) schlagen die
|
||
# qgis-Imports in den Untermodulen fehl; daher der try/except-Guard.
|
||
try:
|
||
from sn_plan41.modules.vln_api_client import KartenApiClient, ApiError, AuthError
|
||
from sn_plan41.modules.vln_layer_manager import (
|
||
feature_collection_to_layers,
|
||
plugin_layers,
|
||
layers_to_feature_collection,
|
||
layer_api_path,
|
||
PROP_DATASET,
|
||
PROP_VERFAHREN,
|
||
)
|
||
VLN_KARTEN_AVAILABLE = True
|
||
except (ImportError, ModuleNotFoundError):
|
||
KartenApiClient = None # type: ignore[assignment,misc]
|
||
ApiError = Exception # type: ignore[assignment,misc]
|
||
AuthError = Exception # type: ignore[assignment,misc]
|
||
VLN_KARTEN_AVAILABLE = False
|
||
|
||
# Geometrietyp-Konstanten (QGIS intern: 1 = Linie, 2 = Fläche)
|
||
try:
|
||
from qgis.core import Qgis as _Qgis
|
||
_GEOM_LINE = _Qgis.GeometryType.Line
|
||
_GEOM_POLYGON = _Qgis.GeometryType.Polygon
|
||
except (ImportError, AttributeError):
|
||
try:
|
||
from qgis.core import QgsWkbTypes as _QgsWkbTypes
|
||
_GEOM_LINE = _QgsWkbTypes.LineGeometry
|
||
_GEOM_POLYGON = _QgsWkbTypes.PolygonGeometry
|
||
except (ImportError, AttributeError):
|
||
_GEOM_LINE = 1
|
||
_GEOM_POLYGON = 2
|
||
|
||
# Konstanten
|
||
SETTINGS_GROUP = "vln_karten"
|
||
PROJECT_SCOPE = "vln_karten"
|
||
PROJECT_KEY_VKZ = "/vkz"
|
||
|
||
P41_DATASET_KEY = "p41"
|
||
P41_LABEL = "Plan 41 (Wege- und Gewässerplan)"
|
||
P41_PATH_TEMPLATE = "/maps/p41/{vkz}"
|
||
P41_GEOMETRY = "MultiPolygon"
|
||
|
||
STYLE_FLAECHE = "QGIS_P41_API_Layer_flaeche.qml"
|
||
STYLE_LINIE = "QGIS_P41_API_Layer_Linie.qml"
|
||
|
||
|
||
class VlnApiLogic:
|
||
"""
|
||
Kapselt die VLN-API-Fachlogik für den Plan41-Tab.
|
||
|
||
Alle Methoden, die Netzwerkkommunikation erfordern, können
|
||
``ApiError`` oder ``AuthError`` werfen – die UI ist für die
|
||
Fehlerdarstellung zuständig.
|
||
"""
|
||
|
||
def __init__(self, pruefmanager=None) -> None:
|
||
self.pruefmanager = pruefmanager
|
||
self._client: Optional[object] = None # KartenApiClient | None
|
||
|
||
# API-Client aus gespeicherten Credentials initialisieren
|
||
if VLN_KARTEN_AVAILABLE:
|
||
api_key, mail = self.load_stored_credentials()
|
||
if api_key:
|
||
self._client = KartenApiClient(api_key=api_key, mail=mail)
|
||
|
||
# ------------------------------------------------------------------
|
||
# Verfügbarkeit
|
||
# ------------------------------------------------------------------
|
||
|
||
@staticmethod
|
||
def is_available() -> bool:
|
||
"""True, wenn die lokalen API-Module geladen werden konnten
|
||
(erfordert eine QGIS-Laufzeitumgebung)."""
|
||
return VLN_KARTEN_AVAILABLE
|
||
|
||
# ------------------------------------------------------------------
|
||
# Authentifizierungsstatus
|
||
# ------------------------------------------------------------------
|
||
|
||
@property
|
||
def is_authenticated(self) -> bool:
|
||
"""True, wenn ein gültiger API-Key vorhanden ist."""
|
||
if self._client is None:
|
||
return False
|
||
return bool(getattr(self._client, "is_authenticated", False))
|
||
|
||
@property
|
||
def mail(self) -> Optional[str]:
|
||
"""E-Mail-Adresse des angemeldeten Benutzers oder None."""
|
||
if self._client is None:
|
||
return None
|
||
return getattr(self._client, "mail", None)
|
||
|
||
# ------------------------------------------------------------------
|
||
# QSettings: API-Key persistieren
|
||
# ------------------------------------------------------------------
|
||
|
||
def load_stored_credentials(self) -> tuple:
|
||
"""Liest API-Key und Mail aus QSettings.
|
||
Gibt ``(api_key, mail)`` zurück – leere Strings wenn nicht vorhanden.
|
||
"""
|
||
settings = QSettings()
|
||
settings.beginGroup(SETTINGS_GROUP)
|
||
api_key = settings.value("api_key", "")
|
||
mail = settings.value("mail", "")
|
||
settings.endGroup()
|
||
return api_key or None, mail or None
|
||
|
||
def save_api_key(self, api_key: str, mail: str) -> None:
|
||
"""Speichert API-Key und Mail in QSettings."""
|
||
settings = QSettings()
|
||
settings.beginGroup(SETTINGS_GROUP)
|
||
settings.setValue("api_key", api_key)
|
||
settings.setValue("mail", mail or "")
|
||
settings.endGroup()
|
||
|
||
def clear_api_key(self) -> None:
|
||
"""Entfernt API-Key aus QSettings."""
|
||
settings = QSettings()
|
||
settings.beginGroup(SETTINGS_GROUP)
|
||
settings.remove("api_key")
|
||
settings.endGroup()
|
||
|
||
# ------------------------------------------------------------------
|
||
# Login / Logout
|
||
# ------------------------------------------------------------------
|
||
|
||
def login(self, mail: str, password: str) -> None:
|
||
"""
|
||
Meldet den Benutzer an und speichert den API-Key.
|
||
|
||
:raises ApiError: Bei ungültigen Zugangsdaten oder Netzwerkproblem.
|
||
:raises RuntimeError: Wenn QGIS nicht verfügbar ist (kein qgis-Paket).
|
||
"""
|
||
if not VLN_KARTEN_AVAILABLE:
|
||
raise RuntimeError(
|
||
"VLN-API-Module nicht verfügbar (QGIS-Laufzeitumgebung benötigt)."
|
||
)
|
||
client = KartenApiClient()
|
||
client.login(mail, password) # wirft ApiError bei Fehler
|
||
self._client = client
|
||
self.save_api_key(client.api_key, mail)
|
||
|
||
def logout(self) -> None:
|
||
"""Meldet den Benutzer ab (lokal, kein API-Aufruf)."""
|
||
if self._client is not None:
|
||
getattr(self._client, "logout", lambda: None)()
|
||
self._client = None
|
||
|
||
def handle_session_expired(self) -> None:
|
||
"""
|
||
Behandelt einen abgelaufenen API-Key:
|
||
entfernt ihn aus QSettings und setzt den Client zurück.
|
||
"""
|
||
self.clear_api_key()
|
||
self.logout()
|
||
|
||
# ------------------------------------------------------------------
|
||
# Verfahrensliste
|
||
# ------------------------------------------------------------------
|
||
|
||
def get_verfahren(self) -> list:
|
||
"""
|
||
Lädt alle Verfahren (TGs) vom Server.
|
||
|
||
:returns: Sortierte Liste von ``{"vkz": ..., "name": ...}``-Dicts.
|
||
:raises AuthError: Wenn nicht angemeldet oder Session abgelaufen.
|
||
:raises ApiError: Bei sonstigem Netzwerkfehler.
|
||
"""
|
||
if self._client is None:
|
||
raise AuthError("Nicht angemeldet.")
|
||
return self._client.get_verfahren()
|
||
|
||
# ------------------------------------------------------------------
|
||
# VKZ-Persistenz in Projektdatei
|
||
# ------------------------------------------------------------------
|
||
|
||
def load_stored_vkz(self) -> Optional[str]:
|
||
"""Liest die zuletzt gewählte VKZ aus der Projektdatei."""
|
||
try:
|
||
vkz, _ok = QgsProject.instance().readEntry(
|
||
PROJECT_SCOPE, PROJECT_KEY_VKZ, ""
|
||
)
|
||
return vkz if vkz else None
|
||
except Exception:
|
||
return None
|
||
|
||
def save_vkz(self, vkz: Optional[str]) -> None:
|
||
"""Schreibt die gewählte VKZ in die Projektdatei."""
|
||
if not vkz:
|
||
return
|
||
try:
|
||
QgsProject.instance().writeEntry(PROJECT_SCOPE, PROJECT_KEY_VKZ, vkz)
|
||
except Exception:
|
||
pass
|
||
|
||
# ------------------------------------------------------------------
|
||
# Plan 41 laden
|
||
# ------------------------------------------------------------------
|
||
|
||
def get_p41_layers(self, vkz: str) -> list:
|
||
"""Gibt alle bereits geladenen P41-Layer für diese VKZ zurück."""
|
||
return plugin_layers(P41_DATASET_KEY, vkz)
|
||
|
||
def remove_p41_layers(self, vkz: str) -> None:
|
||
"""Entfernt alle P41-Layer dieser VKZ aus dem Projekt."""
|
||
existing = plugin_layers(P41_DATASET_KEY, vkz)
|
||
if existing:
|
||
QgsProject.instance().removeMapLayers([l.id() for l in existing])
|
||
|
||
def load_p41(self, vkz: str) -> list:
|
||
"""
|
||
Lädt den Plan-41-Datensatz für eine VKZ vom Server und legt
|
||
Memory-Layer im Projekt an. Auf die entstandenen Layer werden
|
||
die passenden QML-Stile aus ``sn_plan41/assets/`` angewendet.
|
||
|
||
:returns: Liste der erzeugten ``QgsVectorLayer``.
|
||
:raises AuthError: Wenn nicht angemeldet oder Session abgelaufen.
|
||
:raises ApiError: Bei sonstigem Netzwerkfehler.
|
||
:raises RuntimeError: Wenn QGIS nicht verfügbar ist.
|
||
"""
|
||
if not VLN_KARTEN_AVAILABLE:
|
||
raise RuntimeError(
|
||
"VLN-API-Module nicht verfügbar (QGIS-Laufzeitumgebung benötigt)."
|
||
)
|
||
if self._client is None:
|
||
raise AuthError("Nicht angemeldet.")
|
||
|
||
path = P41_PATH_TEMPLATE.format(vkz=vkz)
|
||
feature_collection = self._client.load_layer_complete(
|
||
P41_DATASET_KEY, vkz
|
||
)
|
||
base_name = "%s %s" % (P41_LABEL, vkz)
|
||
layers = feature_collection_to_layers(
|
||
feature_collection,
|
||
base_name,
|
||
P41_GEOMETRY,
|
||
P41_DATASET_KEY,
|
||
vkz,
|
||
path,
|
||
)
|
||
self._apply_p41_styles(layers)
|
||
return layers
|
||
|
||
def _apply_p41_styles(self, layers: list) -> None:
|
||
"""Wendet den passenden QML-Stil auf jeden Layer an."""
|
||
assets_dir = join_path(get_plugin_root(), "sn_plan41", "assets")
|
||
for layer in layers:
|
||
try:
|
||
gtype = layer.geometryType()
|
||
except Exception:
|
||
continue
|
||
|
||
if gtype == _GEOM_POLYGON:
|
||
style_file = STYLE_FLAECHE
|
||
elif gtype == _GEOM_LINE:
|
||
style_file = STYLE_LINIE
|
||
else:
|
||
continue # Punkt- und None-Layer erhalten keinen Stil
|
||
|
||
style_path = join_path(assets_dir, style_file)
|
||
apply_style_from_path(layer, style_path)
|
||
|
||
# ------------------------------------------------------------------
|
||
# Upload
|
||
# ------------------------------------------------------------------
|
||
|
||
def upload_active_layer(self, active_layer) -> tuple:
|
||
"""
|
||
Lädt alle Teil-Layer des aktiven Layers zum Server hoch.
|
||
|
||
:param active_layer: Aktiver ``QgsVectorLayer`` (aus ``iface.activeLayer()``).
|
||
:returns: ``(success: bool, message: str)``
|
||
:raises AuthError: Wenn Session abgelaufen.
|
||
:raises ApiError: Bei Netzwerkfehler.
|
||
"""
|
||
if not VLN_KARTEN_AVAILABLE:
|
||
return False, (
|
||
"VLN-API-Module nicht verfügbar (QGIS-Laufzeitumgebung benötigt)."
|
||
)
|
||
|
||
if self._client is None:
|
||
raise AuthError("Nicht angemeldet.")
|
||
|
||
path = layer_api_path(active_layer) if active_layer else None
|
||
if not path:
|
||
return (
|
||
False,
|
||
"Der aktive Layer wurde nicht über 'vln_karten' geladen. "
|
||
"Bitte einen Plan41-Layer aus dem Projekt auswählen.",
|
||
)
|
||
|
||
dataset_key = active_layer.customProperty(PROP_DATASET)
|
||
vkz = active_layer.customProperty(PROP_VERFAHREN)
|
||
siblings = plugin_layers(dataset_key, vkz)
|
||
|
||
for layer in siblings:
|
||
if layer.isEditable() and not layer.commitChanges():
|
||
return (
|
||
False,
|
||
"Die Bearbeitungssitzung von '%s' konnte nicht "
|
||
"gespeichert werden." % layer.name(),
|
||
)
|
||
|
||
feature_collection = layers_to_feature_collection(siblings)
|
||
self._client.save_feature_collection(path, feature_collection)
|
||
|
||
total = sum(layer.featureCount() for layer in siblings)
|
||
return (
|
||
True,
|
||
"%d Objekte aus %d Layer(n) erfolgreich hochgeladen."
|
||
% (total, len(siblings)),
|
||
)
|
||
|
||
def upload_summary(self, active_layer) -> Optional[dict]:
|
||
"""
|
||
Gibt eine Zusammenfassung der zu hochladenden Layer zurück
|
||
(für den Bestätigungs-Dialog in der UI), ohne den Upload
|
||
tatsächlich auszuführen.
|
||
|
||
:returns: Dict mit ``dataset_label``, ``vkz``, ``siblings``
|
||
(Layer-Liste), ``total`` (Objektanzahl), ``listing`` (str)
|
||
oder ``None`` wenn kein gültiger Layer.
|
||
"""
|
||
if not VLN_KARTEN_AVAILABLE or active_layer is None:
|
||
return None
|
||
path = layer_api_path(active_layer)
|
||
if not path:
|
||
return None
|
||
|
||
dataset_key = active_layer.customProperty(PROP_DATASET)
|
||
vkz = active_layer.customProperty(PROP_VERFAHREN)
|
||
siblings = plugin_layers(dataset_key, vkz)
|
||
|
||
total = sum(l.featureCount() for l in siblings)
|
||
listing = "\n".join(
|
||
"• %s (%d Objekte)" % (l.name(), l.featureCount())
|
||
for l in siblings
|
||
)
|
||
return {
|
||
"dataset_label": P41_LABEL if dataset_key == P41_DATASET_KEY else dataset_key,
|
||
"vkz": vkz,
|
||
"siblings": siblings,
|
||
"total": total,
|
||
"listing": listing,
|
||
}
|