VLN_SN/API_Karte_QGISDemo
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

QGIS-Plugin VLN Karten: Verfahrens-Layer der Karten-API laden und hochladen

Browse Source 
Lädt Verfahrensgebiet, Plan 41, Karte alter Stand (KAS) und Wertermittlung
(WE) je VKZ vollständig aus KARTE_OBJEKT (Listen-Endpunkt mit Paging) und
schreibt sie per PUT zurück. Einmaliger Login (mail/password -> userauth),
API-Key persistiert in QSettings; Verfahrens-Auswahl in der Toolbar wird
je QGIS-Projekt gemerkt. Gemischte Geometrietypen werden beim Laden in
Punkte-/Linien-/Flächen-Layer gesplittet und beim Hochladen wieder
vereint. Qt5/Qt6-kompatibel (QGIS 3.22+ und QGIS 4).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
erik
2026-06-12 13:29:09 +02:00
parent b222b33de7
commit 091ef281b0
8 changed files with 1075 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+3
View File
@@ -0,0 +1,3 @@
__pycache__/
*.pyc
.DS_Store
README.md
+144 -1
View File
@@ -1,2 +1,145 @@
# API_Karte_QGISDemo # VLN Karten — QGIS-Plugin für die Karten-API des VLN Managers
QGIS-Plugin, das die Karten-Layer des [VLN Managers](https://www.vlnsachsen.de)
direkt in QGIS bearbeitbar macht. Es liest und schreibt dieselben
Datenbank-Daten (`KARTE_OBJEKT`), die auch das Web-GIS
(karte.flurneuordnung-sachsen.de) über die `/maps/*`-Endpunkte nutzt.
## Funktionen
- **Einmaliger Login** mit dem VLN-Manager-Konto (E-Mail/Passwort).
Der zurückgelieferte API-Key wird in den QGIS-Einstellungen gespeichert
und bei jedem QGIS-Start automatisch wiederverwendet; erst bei
HTTP 401/403 erscheint der Anmeldedialog erneut.
- **Verfahrens-Auswahl in der Toolbar** — alle Teilnehmergemeinschaften
aus `GET /tgen` als „VKZ — Name“. Die Auswahl wird in der Projektdatei
gespeichert und beim Öffnen des Projekts wiederhergestellt.
- **Vier Layer je Verfahren laden:**
| Button | API-Layer | DB-Art |
|---|---|---|
| Verfahrensgebiet | `umringe` | `UMRING` |
| Plan 41 (Wege- und Gewässerplan) | `p41` | `P41` |
| Karte alter Stand | `kas` | `KAS` |
| Wertermittlung | `we` | `WE` |
Geladen wird **vollständig** über den Listen-Endpunkt mit
limit/offset-Paging — auch bei mehr als 2000 Objekten.
- **Gemischte Geometrietypen:** Die API liefert Punkte, Linien und
Flächen in einer FeatureCollection. Beim Laden wird nach
Geometrie-Familie in bis zu drei Memory-Layer gesplittet
(Single-Typen werden zu Multi-Typen befördert); bei nur einem Typ
entsteht ein einzelner Layer.
- **Hochladen:** „Aktiven Layer hochladen“ vereint alle Teil-Layer
desselben Datensatzes (Punkte + Linien + Flächen derselben VKZ) wieder
zu einer FeatureCollection und schreibt sie per `PUT` zurück.
Vorher: Commit offener Bearbeitungssitzungen und Sicherheitsabfrage
mit Auflistung der beteiligten Layer.
- **Schutz vor Duplikaten:** Erneutes Laden eines bereits geladenen
Datensatzes ersetzt die vorhandenen Layer nach Rückfrage, statt sie zu
stapeln. Liefert der Server 0 Objekte, wird das deutlich gemeldet und
ein leerer Layer zum Digitalisieren angelegt.
## Voraussetzungen
- QGIS **3.22 oder neuer**, einschließlich **QGIS 4** (Qt6/PyQt6 —
der Code verwendet durchgehend scoped Enums und `exec()`).
- Ein VLN-Manager-Konto mit Zugriff auf die Karten-API.
## Installation
Repository klonen und den Plugin-Ordner in das QGIS-Profil verlinken
(alternativ kopieren), dann QGIS neu starten:
```bash
git clone https://entwicklung.flurneuordnung-sachsen.de/VLN_SN/API_Karte_QGISDemo.git
cd API_Karte_QGISDemo
# macOS, QGIS 4 (bei QGIS 3: "QGIS4" durch "QGIS3" ersetzen):
ln -s "$PWD/vln_karten" \
~/Library/Application\ Support/QGIS/QGIS4/profiles/default/python/plugins/vln_karten
# Linux:
# ~/.local/share/QGIS/QGIS4/profiles/default/python/plugins/
# Windows:
# %APPDATA%\QGIS\QGIS4\profiles\default\python\plugins\
```
Anschließend in QGIS unter *Erweiterungen → Erweiterungen verwalten*
„VLN Karten“ aktivieren (experimentelle Erweiterungen zulassen).
## Bedienung
1. **Anmelden …** in der Toolbar „VLN Karten“ (auch unter
*Web → VLN Karten*): E-Mail und Passwort des VLN-Manager-Kontos.
Das Passwort wird nicht gespeichert, nur der API-Key.
2. **Verfahren wählen** in der Auswahlliste.
3. **Layer laden**, in QGIS normal editieren (Memory-Layer,
EPSG:25833), **hochladen**.
> ⚠️ Der `PUT` der API ersetzt den kompletten Layer-Bestand der
> jeweiligen VKZ (versionierter Snapshot in `KARTE_SPEICHERSTAND`).
> Deshalb lädt das Plugin immer alle Teil-Layer gemeinsam hoch — nur
> einen Teil zu senden, würde die übrigen Geometrietypen serverseitig
> löschen.
## API-Vertrag
Basis-URL: `https://api.flurneuordnung-sachsen.de/v2`
(fest hinterlegt als `DEFAULT_BASE_URL` in
[vln_karten/api_client.py](vln_karten/api_client.py))
```
POST /person/login
Body: {"mail": "...", "password": "..."}
Antwort: {"data": {"userauth": "<url_token>", "id": ...}}
userauth = API-Key, danach Header "X-API-Key: <userauth>"
GET /tgen Verfahren/TGs (Auth nötig)
GET /maps/<layer>?vkz=&limit=&offset= Layer lesen (kas/we: Auth nötig)
Antwort: GeoJSON FeatureCollection in EPSG:25833 (ETRS89/UTM33)
PUT /maps/<layer>/{vkz} Layer schreiben (Auth nötig)
Body: GeoJSON (ersetzt den Layer dieser VKZ, neuer Snapshot)
Antwort: {"data": {"vkz","layer","art","speicher_id"}, "status":"ok"}
Fehlerformat: RFC 7807 (application/problem+json)
```
Ein weiterer Layer der API (`st` = Servicetermin) wäre ein zusätzlicher
Eintrag im `DATASETS`-Dict in [vln_karten/plugin.py](vln_karten/plugin.py).
## Aufbau
| Datei | Zweck |
|---|---|
| [vln_karten/plugin.py](vln_karten/plugin.py) | Toolbar, Aktionen, Verfahrens-Auswahl, `DATASETS`-Registry |
| [vln_karten/api_client.py](vln_karten/api_client.py) | HTTP-Client (Login, Paging-Loader, PUT) über `QgsBlockingNetworkRequest` |
| [vln_karten/layer_manager.py](vln_karten/layer_manager.py) | GeoJSON ↔ Memory-Layer, Geometrie-Splitting, Layer-Zusammenführung |
| [vln_karten/login_dialog.py](vln_karten/login_dialog.py) | Anmeldedialog |
| [vln_karten/metadata.txt](vln_karten/metadata.txt) | QGIS-Plugin-Metadaten |
## Technische Hinweise
- **CRS:** Die API liefert und erwartet Koordinaten in **EPSG:25833**
(abweichend von RFC 7946). Import/Export laufen ohne
WGS84-Transformation (`GEOJSON_CRS` in `layer_manager.py`).
- **Geometrien** gehen beim Hochladen als Multi-Typen an die API
(`ST_GeomFromGeoJSON` akzeptiert beides). Features ohne Geometrie
landen in einem Tabellen-Layer „ohne Geometrie“.
- **API-Key-Ablage:** unverschlüsselt in den QGIS-Einstellungen
(QSettings, Gruppe `vln_karten`). Wer das härten will, verlagert ihn
in den QGIS-Authentifizierungsmanager (`QgsApplication.authManager()`).
- **Requests** laufen synchron (blockierend) — für sehr große
Datensätze wäre `QgsNetworkAccessManager` mit Tasks der nächste Schritt.
- Die gewählte VKZ liegt in der Projektdatei (`writeEntry`-Scope
`vln_karten`) — verschiedene Projekte können verschiedene Verfahren
vorausgewählt haben.
## Entwicklung
Für schnelles Iterieren empfiehlt sich das Plugin **Plugin Reloader**
aus dem offiziellen QGIS-Repository — Codeänderungen wirken dann ohne
QGIS-Neustart. Der Plugin-Ordner kann dafür per Symlink direkt aus dem
Git-Arbeitsverzeichnis eingebunden bleiben (siehe Installation).
vln_karten/__init__.py
+3
View File
@@ -0,0 +1,3 @@
def classFactory(iface):
from .plugin import VlnKartenPlugin
return VlnKartenPlugin(iface)
vln_karten/api_client.py
+255
View File
@@ -0,0 +1,255 @@
"""HTTP-Client für die VLN-Manager-API (Karten-Plugin map.php + Stammdaten).
API-Vertrag laut Doku im Wissensdatenbank-Vault
("API.md", "API – Karten (map.php).md", "API – Schnittstellen (Referenz).md",
Stand 2026-06-08):
Basis-URL: https://api.flurneuordnung-sachsen.de/v2
POST /person/login
Body: {"mail": "...", "password": "..."}
Antwort: {"data": {"userauth": "<url_token>", "id": ...}}
Das userauth-Token ist der API-Key (= PERSON.url_token).
GET /tgen -> Teilnehmergemeinschaften (Verfahren)
GET /maps/<layer>/{vkz} -> GeoJSON FeatureCollection (EPSG:25833)
PUT /maps/<layer>/{vkz} <- GeoJSON (ersetzt den Layer dieser VKZ)
Antwort: {"data": {"vkz", "layer", "art", "speicher_id"}, "status": "ok"}
Layer: umringe, p41, st, kas, we
Auth nach Login über den Header "X-API-Key".
Fehlerformat: RFC 7807 (application/problem+json).
Es wird QgsBlockingNetworkRequest verwendet, damit Proxy- und
Zertifikatseinstellungen aus QGIS automatisch greifen.
"""
import json
from qgis.PyQt.QtCore import QUrl
from qgis.PyQt.QtNetwork import QNetworkRequest
from qgis.core import QgsBlockingNetworkRequest
# Fest verdrahtete Produktiv-API des VLN Managers.
DEFAULT_BASE_URL = "https://api.flurneuordnung-sachsen.de/v2"
class ApiError(Exception):
"""Fehler bei der Kommunikation mit der Karten-API."""
class AuthError(ApiError):
"""API-Key fehlt, ist abgelaufen oder hat keine Berechtigung (401/403)."""
class KartenApiClient:
def __init__(self, base_url=DEFAULT_BASE_URL, api_key=None, mail=None):
self.base_url = base_url.rstrip("/")
self.api_key = api_key
self.mail = mail
@property
def is_authenticated(self):
return bool(self.api_key)
def login(self, mail, password):
"""Meldet den Benutzer an. Die API liefert das userauth-Token
(= API-Key), das für alle weiteren Requests verwendet wird."""
data = self._request(
"POST",
"/person/login",
payload={"mail": mail, "password": password},
with_auth=False,
)
payload = data.get("data") if isinstance(data, dict) else None
if not isinstance(payload, dict):
payload = data if isinstance(data, dict) else {}
api_key = payload.get("userauth")
if not api_key:
raise ApiError(
"Login-Antwort enthielt kein userauth-Token. "
"Erhaltene Antwort: %s" % json.dumps(data)[:300]
)
self.api_key = api_key
self.mail = mail
return api_key
def logout(self):
self.api_key = None
self.mail = None
# ------------------------------------------------------------------
# Verfahren (Teilnehmergemeinschaften)
# ------------------------------------------------------------------
def get_verfahren(self):
"""Liste der Verfahren (TGs) als [{"vkz": ..., "name": ...}, ...],
sortiert nach VKZ. Quelle: GET /tgen."""
data = self._request("GET", "/tgen")
rows = None
if isinstance(data, list):
rows = data
elif isinstance(data, dict):
for key in ("data", "rows"):
value = data.get(key)
if isinstance(value, list):
rows = value
break
if isinstance(value, dict) and isinstance(value.get("rows"), list):
rows = value["rows"]
break
if rows is None:
raise ApiError(
"Antwort von /tgen hat ein unerwartetes Format: %s"
% json.dumps(data)[:200]
)
verfahren = []
for row in rows:
if not isinstance(row, dict):
continue
vkz = row.get("vkz") or row.get("VKZ")
name = (
row.get("kurzname")
or row.get("Kurzname")
or row.get("name")
or row.get("Name")
or ""
)
if vkz:
verfahren.append({"vkz": str(vkz), "name": str(name)})
verfahren.sort(key=lambda v: v["vkz"])
return verfahren
# ------------------------------------------------------------------
# Karten-Layer (die konkreten Pfade stehen in plugin.DATASETS)
# ------------------------------------------------------------------
def load_layer_complete(self, layer_key, vkz, page_size=2000):
"""Lädt ALLE Objekte eines Layers für eine VKZ aus KARTE_OBJEKT.
Nutzt wie das Web-GIS (0_map, db_p41) den Listen-Endpunkt
/maps/<layer>?vkz=…, aber mit limit/offset-Paging, bis keine
weitere Seite mehr kommt — so ist der Abruf auch bei mehr als
2000 Objekten garantiert vollständig."""
from urllib.parse import quote
features = []
offset = 0
while True:
data = self._request(
"GET",
"/maps/%s?vkz=%s&limit=%d&offset=%d"
% (quote(str(layer_key)), quote(str(vkz)), page_size, offset),
)
if not isinstance(data, dict) or data.get("type") != "FeatureCollection":
raise ApiError(
"Antwort von /maps/%s ist keine GeoJSON FeatureCollection."
% layer_key
)
page = data.get("features", [])
features.extend(page)
if len(page) < page_size:
break
offset += page_size
return {"type": "FeatureCollection", "features": features}
def load_feature_collection(self, path):
"""Lädt einen Layer als GeoJSON FeatureCollection (dict)."""
data = self._request("GET", path)
# Die Karten-Endpunkte liefern die FeatureCollection direkt,
# andere Routen verpacken sie in {"data": ...}.
if isinstance(data, dict):
if data.get("type") == "FeatureCollection":
return data
inner = data.get("data")
if isinstance(inner, dict) and inner.get("type") == "FeatureCollection":
return inner
raise ApiError(
"Antwort von %s ist keine GeoJSON FeatureCollection." % path
)
def save_feature_collection(self, path, feature_collection):
"""Schreibt eine GeoJSON FeatureCollection zurück an die API
(PUT ersetzt den kompletten Layer der jeweiligen VKZ)."""
return self._request("PUT", path, payload=feature_collection)
# ------------------------------------------------------------------
# Intern
# ------------------------------------------------------------------
def _request(self, method, path, payload=None, with_auth=True):
if with_auth and not self.is_authenticated:
raise AuthError("Nicht angemeldet — bitte zuerst einloggen.")
request = QNetworkRequest(QUrl(self.base_url + path))
request.setHeader(
QNetworkRequest.KnownHeaders.ContentTypeHeader, "application/json"
)
request.setRawHeader(b"Accept", b"application/json")
if with_auth:
request.setRawHeader(b"X-API-Key", self.api_key.encode("utf-8"))
body = b""
if payload is not None:
body = json.dumps(payload).encode("utf-8")
blocking = QgsBlockingNetworkRequest()
if method == "GET":
error = blocking.get(request)
elif method == "POST":
error = blocking.post(request, body)
elif method == "PUT":
error = blocking.put(request, body)
else:
raise ApiError("Nicht unterstützte HTTP-Methode: %s" % method)
reply = blocking.reply()
content = bytes(reply.content())
status = reply.attribute(
QNetworkRequest.Attribute.HttpStatusCodeAttribute
)
status = int(status) if status is not None else None
if status in (401, 403):
raise AuthError(
"Keine Berechtigung für %s %s (HTTP %s)%s"
% (method, path, status, self._error_detail(content))
)
if error != QgsBlockingNetworkRequest.NoError:
raise ApiError(
"%s %s fehlgeschlagen: %s%s"
% (method, path, blocking.errorMessage(), self._error_detail(content))
)
if status is not None and status >= 400:
raise ApiError(
"%s %s lieferte HTTP %s%s"
% (method, path, status, self._error_detail(content))
)
if not content:
return None
try:
return json.loads(content.decode("utf-8"))
except ValueError:
raise ApiError(
"Antwort von %s ist kein gültiges JSON." % path
)
@staticmethod
def _error_detail(content):
"""Liest title/detail aus einer RFC-7807-Fehlerantwort."""
if not content:
return ""
try:
problem = json.loads(content.decode("utf-8"))
parts = [
str(problem[k]) for k in ("title", "detail") if problem.get(k)
]
if parts:
return " — " + ": ".join(parts)
except (ValueError, AttributeError):
pass
return " — " + content[:300].decode("utf-8", "replace")
vln_karten/layer_manager.py
+178
View File
@@ -0,0 +1,178 @@
"""Konvertierung GeoJSON <-> QGIS-Layer.
Die API liefert je Layer (z.B. Plan 41) gemischte Geometrietypen in einer
FeatureCollection: Punkte, Linien, Polygone sowie deren Multi-Varianten.
QGIS-Memory-Layer können nur einen Geometrietyp halten — beim Laden wird
deshalb nach Geometrie-Familie (Punkte/Linien/Flächen) in bis zu drei
Layer gesplittet; Single-Geometrien werden zu Multi befördert.
Beim Hochladen müssen alle Teil-Layer desselben Datensatzes wieder zu
EINER FeatureCollection vereint werden, weil der PUT der API den
kompletten Layer-Bestand der VKZ ersetzt (siehe layers_to_feature_collection
und plugin_layers).
Welcher Datensatz/welche VKZ zu einem Layer gehört, steht als
Custom-Property am Layer.
"""
import json
from qgis.core import (
QgsJsonExporter,
QgsJsonUtils,
QgsProject,
QgsVectorLayer,
)
try: # QGIS >= 3.30 und QGIS 4
from qgis.core import Qgis
_GEOM_POINT = Qgis.GeometryType.Point
_GEOM_LINE = Qgis.GeometryType.Line
_GEOM_POLYGON = Qgis.GeometryType.Polygon
except (ImportError, AttributeError): # ältere QGIS-3-Versionen
from qgis.core import QgsWkbTypes
_GEOM_POINT = QgsWkbTypes.PointGeometry
_GEOM_LINE = QgsWkbTypes.LineGeometry
_GEOM_POLYGON = QgsWkbTypes.PolygonGeometry
# Die VLN-API liefert Koordinaten unverändert aus der DB (ST_AsGeoJSON):
# ETRS89 / UTM Zone 33 — verifiziert per GET /maps/umringe (Werte wie
# [406924, 5658760] liegen im Raum Dresden, nicht in Grad).
GEOJSON_CRS = "EPSG:25833"
PROP_DATASET = "vln_karten/dataset"
PROP_VERFAHREN = "vln_karten/verfahren"
PROP_PATH = "vln_karten/api_path"
# Geometrie-Familie -> (Memory-Provider-Typ, Namenszusatz).
# Reihenfolge = Lade-Reihenfolge: Flächen zuerst, damit Punkte im
# Layerbaum oben landen und nicht verdeckt werden.
_FAMILIES = (
("polygon", "MultiPolygon", "Flächen"),
("line", "MultiLineString", "Linien"),
("point", "MultiPoint", "Punkte"),
("none", "None", "ohne Geometrie"),
)
def _family(feature):
"""Geometrie-Familie eines Features. Unbekannte Typen (z.B.
GeometryCollection) landen bei 'none' — Attribute bleiben erhalten,
die Geometrie ginge beim Hochladen verloren."""
if not feature.hasGeometry() or feature.geometry().isNull():
return "none"
gtype = feature.geometry().type()
if gtype == _GEOM_POINT:
return "point"
if gtype == _GEOM_LINE:
return "line"
if gtype == _GEOM_POLYGON:
return "polygon"
return "none"
def memory_layer_from_features(
uri_type, name, fields, features, dataset_key, verfahren_nr, api_path, promote=True
):
"""Baut einen Memory-Layer aus fertigen QgsFeatures, setzt die
Plugin-Custom-Properties und hängt ihn ins Projekt."""
uri = uri_type if uri_type == "None" else "%s?crs=%s" % (uri_type, GEOJSON_CRS)
layer = QgsVectorLayer(uri, name, "memory")
if not layer.isValid():
raise RuntimeError("Memory-Layer '%s' konnte nicht erzeugt werden." % name)
provider = layer.dataProvider()
provider.addAttributes(fields.toList())
layer.updateFields()
prepared = []
for feature in features:
geometry = feature.geometry()
if promote and not geometry.isNull() and not geometry.isMultipart():
geometry.convertToMultiType()
feature.setGeometry(geometry)
prepared.append(feature)
provider.addFeatures(prepared)
layer.updateExtents()
layer.setCustomProperty(PROP_DATASET, dataset_key)
layer.setCustomProperty(PROP_VERFAHREN, verfahren_nr)
layer.setCustomProperty(PROP_PATH, api_path)
return layer
def feature_collection_to_layers(
feature_collection, base_name, default_geometry, dataset_key, verfahren_nr, api_path
):
"""Erzeugt aus einer GeoJSON FeatureCollection je vorkommender
Geometrie-Familie einen Memory-Layer und hängt sie ins Projekt.
Liefert die Liste der erzeugten Layer."""
text = json.dumps(feature_collection)
fields = QgsJsonUtils.stringToFields(text)
features = QgsJsonUtils.stringToFeatureList(text, fields)
groups = {}
for feature in features:
groups.setdefault(_family(feature), []).append(feature)
layers = []
multiple = len(groups) > 1
def make_layer(uri_type, name, group_features, promote):
layers.append(
memory_layer_from_features(
uri_type, name, fields, group_features,
dataset_key, verfahren_nr, api_path, promote=promote,
)
)
if not groups:
# Leerer Datensatz: ein leerer Layer im Default-Typ, damit der
# Nutzer digitalisieren und hochladen kann.
make_layer(default_geometry, base_name, [], promote=False)
else:
for family, uri_type, suffix in _FAMILIES:
if family not in groups:
continue
name = "%s — %s" % (base_name, suffix) if multiple else base_name
make_layer(uri_type, name, groups[family], promote=(family != "none"))
QgsProject.instance().addMapLayers(layers)
return layers
def plugin_layers(dataset_key, verfahren_nr):
"""Alle Layer im Projekt, die zu diesem Datensatz und dieser VKZ
gehören (z.B. die Punkte-/Linien-/Flächen-Teil-Layer von Plan 41)."""
result = []
for layer in QgsProject.instance().mapLayers().values():
if (
isinstance(layer, QgsVectorLayer)
and layer.customProperty(PROP_DATASET) == dataset_key
and str(layer.customProperty(PROP_VERFAHREN)) == str(verfahren_nr)
):
result.append(layer)
return result
def layers_to_feature_collection(layers):
"""Vereint die Features mehrerer Layer zu EINER GeoJSON
FeatureCollection (für den PUT, der den Serverbestand ersetzt).
Keine Transformation nach WGS84: Die API erwartet die Koordinaten im
selben CRS, in dem sie sie liefert (EPSG:25833)."""
merged = []
for layer in layers:
exporter = QgsJsonExporter(layer)
exporter.setSourceCrs(layer.crs())
exporter.setTransformGeometries(False)
collection = json.loads(
exporter.exportFeatures(list(layer.getFeatures()))
)
merged.extend(collection.get("features", []))
return {"type": "FeatureCollection", "features": merged}
def layer_api_path(layer):
"""API-Pfad, unter dem dieser Layer geladen wurde — oder None,
wenn der Layer nicht von diesem Plugin stammt."""
if not isinstance(layer, QgsVectorLayer):
return None
return layer.customProperty(PROP_PATH) or None
vln_karten/login_dialog.py
+59
View File
@@ -0,0 +1,59 @@
"""Anmeldedialog: E-Mail und Passwort.
Die Server-URL ist fest im Client hinterlegt (api_client.DEFAULT_BASE_URL).
Die E-Mail-Adresse wird in den QGIS-Einstellungen gemerkt, das Passwort
bewusst nicht.
"""
from qgis.PyQt.QtCore import QSettings
from qgis.PyQt.QtWidgets import (
QDialog,
QDialogButtonBox,
QFormLayout,
QLineEdit,
)
SETTINGS_GROUP = "vln_karten"
class LoginDialog(QDialog):
def __init__(self, parent=None):
super().__init__(parent)
self.setWindowTitle("VLN Karten — Anmeldung")
self.setMinimumWidth(380)
settings = QSettings()
settings.beginGroup(SETTINGS_GROUP)
self.mail_edit = QLineEdit(settings.value("mail", ""))
settings.endGroup()
self.mail_edit.setPlaceholderText("vorname.name@beispiel.de")
self.password_edit = QLineEdit()
self.password_edit.setEchoMode(QLineEdit.EchoMode.Password)
layout = QFormLayout(self)
layout.addRow("E-Mail:", self.mail_edit)
layout.addRow("Passwort:", self.password_edit)
buttons = QDialogButtonBox(
QDialogButtonBox.StandardButton.Ok
| QDialogButtonBox.StandardButton.Cancel,
parent=self,
)
buttons.button(QDialogButtonBox.StandardButton.Ok).setText("Anmelden")
buttons.accepted.connect(self.accept)
buttons.rejected.connect(self.reject)
layout.addRow(buttons)
def accept(self):
settings = QSettings()
settings.beginGroup(SETTINGS_GROUP)
settings.setValue("mail", self.mail())
settings.endGroup()
super().accept()
def mail(self):
return self.mail_edit.text().strip()
def password(self):
return self.password_edit.text()
vln_karten/metadata.txt
+17
View File
@@ -0,0 +1,17 @@
[general]
name=VLN Karten
qgisMinimumVersion=3.22
qgisMaximumVersion=4.99
supportsQt6=True
description=Beispiel-Plugin: Verfahrensumringe, Wege- und Gewässerplan und Karte alter Stamm über die VLN Karten-API laden und zurückschreiben.
about=Demonstriert Login (Benutzername/Passwort -> API-Key), Laden von GeoJSON-Daten in editierbare Memory-Layer und Zurückschreiben der Änderungen an die API.
version=0.1.0
author=Erik
email=phpwelt@gmail.com
tracker=https://entwicklung.flurneuordnung-sachsen.de/VLN_SN/API_Karte_QGISDemo/issues
repository=https://entwicklung.flurneuordnung-sachsen.de/VLN_SN/API_Karte_QGISDemo
tags=vln,flurbereinigung,api,geojson
category=Web
icon=
experimental=True
deprecated=False
vln_karten/plugin.py
+416
View File
@@ -0,0 +1,416 @@
"""Hauptklasse des Plugins.
Ablauf:
- Der API-Key wird nach dem ersten Login in den QGIS-Einstellungen
gespeichert und bei jedem QGIS-Start wiederverwendet (kein erneuter
Login nötig, solange der Key gültig ist).
- Eine Toolbar-Auswahlliste zeigt alle Verfahren (TGs) aus GET /tgen.
Das gewählte Verfahren wird in der Projektdatei gemerkt und beim
Öffnen des Projekts wiederhergestellt.
- Buttons laden Verfahrensgebiet / Plan 41 / Karte alter Stand für das
gewählte Verfahren bzw. laden den aktiven Layer zur API hoch.
"""
from functools import partial
from qgis.PyQt.QtCore import QSettings, QTimer
from qgis.PyQt.QtWidgets import QAction, QComboBox, QDialog, QMessageBox
from qgis.core import Qgis, QgsProject
from .api_client import ApiError, AuthError, KartenApiClient
from .layer_manager import (
PROP_DATASET,
PROP_VERFAHREN,
feature_collection_to_layers,
layer_api_path,
layers_to_feature_collection,
plugin_layers,
)
from .login_dialog import LoginDialog, SETTINGS_GROUP
MENU_TITLE = "&VLN Karten"
# Schlüssel für die VKZ-Ablage in der Projektdatei (*.qgz).
PROJECT_SCOPE = "vln_karten"
PROJECT_KEY_VKZ = "/vkz"
# Karten-Layer der VLN-Manager-API (map.php). Weitere Layer (st, we)
# wären je ein Eintrag mehr. {vkz} = Verfahrenskennzeichen der TG.
DATASETS = {
"umringe": {
"label": "Verfahrensgebiet",
"path": "/maps/umringe/{vkz}",
"geometry": "MultiPolygon",
},
"p41": {
"label": "Plan 41 (Wege- und Gewässerplan)",
"path": "/maps/p41/{vkz}",
"geometry": "MultiPolygon",
},
# DB-Art KAS: enthält, was im Web-GIS über "in Datenbank speichern"
# abgelegt wurde — kann je VKZ leer sein, solange dort noch nichts
# gespeichert wurde.
"kas": {
"label": "Karte alter Stand (KAS)",
"path": "/maps/kas/{vkz}",
"geometry": "MultiPolygon",
},
# DB-Art WE: Wertermittlung (Web-GIS-Modul we.js).
"we": {
"label": "Wertermittlung (WE)",
"path": "/maps/we/{vkz}",
"geometry": "MultiPolygon",
},
}
class VlnKartenPlugin:
def __init__(self, iface):
self.iface = iface
self.client = KartenApiClient()
self.actions = []
self.toolbar = None
self.combo = None
# Unterdrückt das Mitschreiben in die Projektdatei, während die
# Auswahl programmatisch gesetzt wird.
self._restoring = False
# ------------------------------------------------------------------
# Plugin-Lebenszyklus
# ------------------------------------------------------------------
def initGui(self):
self.toolbar = self.iface.addToolBar("VLN Karten")
self.toolbar.setObjectName("VlnKartenToolbar")
self.login_action = self._add_action("Anmelden …", self.run_login)
self.combo = QComboBox()
self.combo.setMinimumWidth(280)
self.combo.setToolTip("Verfahren (Teilnehmergemeinschaft) auswählen")
self.combo.currentIndexChanged.connect(self.on_verfahren_changed)
self.toolbar.addWidget(self.combo)
self.dataset_actions = []
for key, dataset in DATASETS.items():
self.dataset_actions.append(
self._add_action(
"%s laden" % dataset["label"],
partial(self.run_load_dataset, key),
)
)
self.dataset_actions.append(
self._add_action("Aktiven Layer hochladen", self.run_upload_active_layer)
)
self.iface.projectRead.connect(self.restore_project_selection)
self.iface.newProjectCreated.connect(self.restore_project_selection)
self._set_logged_out_ui()
# Gespeicherten API-Key aus früherer Sitzung wiederverwenden;
# die Verfahrensliste erst nach dem QGIS-Start abrufen.
settings = QSettings()
settings.beginGroup(SETTINGS_GROUP)
api_key = settings.value("api_key", "")
mail = settings.value("mail", "")
settings.endGroup()
if api_key:
self.client = KartenApiClient(api_key=api_key, mail=mail)
QTimer.singleShot(0, self.populate_verfahren)
def unload(self):
for signal in (self.iface.projectRead, self.iface.newProjectCreated):
try:
signal.disconnect(self.restore_project_selection)
except (TypeError, RuntimeError):
pass
for action in self.actions:
self.iface.removePluginWebMenu(MENU_TITLE, action)
self.actions = []
if self.toolbar is not None:
self.toolbar.deleteLater()
self.toolbar = None
self.combo = None
def _add_action(self, text, callback):
action = QAction(text, self.iface.mainWindow())
action.triggered.connect(callback)
self.iface.addPluginToWebMenu(MENU_TITLE, action)
self.toolbar.addAction(action)
self.actions.append(action)
return action
# ------------------------------------------------------------------
# Anmeldung & Verfahrensliste
# ------------------------------------------------------------------
def run_login(self):
dialog = LoginDialog(self.iface.mainWindow())
if dialog.exec() != QDialog.DialogCode.Accepted:
return
client = KartenApiClient()
try:
client.login(dialog.mail(), dialog.password())
except ApiError as exc:
self._show_error("Anmeldung fehlgeschlagen", str(exc))
return
self.client = client
settings = QSettings()
settings.beginGroup(SETTINGS_GROUP)
settings.setValue("api_key", client.api_key)
settings.endGroup()
self.iface.messageBar().pushMessage(
"VLN Karten",
"Anmeldung erfolgreich — die Zugangsdaten werden für künftige "
"QGIS-Sitzungen gemerkt.",
level=Qgis.Success,
duration=5,
)
self.populate_verfahren()
def populate_verfahren(self):
"""Füllt die Auswahlliste mit allen Verfahren aus GET /tgen."""
try:
verfahren = self.client.get_verfahren()
except AuthError:
self._handle_session_expired()
return
except ApiError as exc:
self.iface.messageBar().pushMessage(
"VLN Karten",
"Verfahrensliste konnte nicht geladen werden: %s" % exc,
level=Qgis.Warning,
duration=10,
)
return
self._restoring = True
self.combo.clear()
self.combo.addItem("— Verfahren wählen —", None)
for v in verfahren:
label = "%s — %s" % (v["vkz"], v["name"]) if v["name"] else v["vkz"]
self.combo.addItem(label, v["vkz"])
self.combo.setEnabled(True)
self._restoring = False
self.login_action.setText(
"Angemeldet: %s" % (self.client.mail or "API-Key")
)
for action in self.dataset_actions:
action.setEnabled(True)
self.restore_project_selection()
def _handle_session_expired(self):
settings = QSettings()
settings.beginGroup(SETTINGS_GROUP)
settings.remove("api_key")
settings.endGroup()
self.client.logout()
self._set_logged_out_ui()
self._show_error(
"Sitzung abgelaufen",
"Der gespeicherte API-Key ist nicht mehr gültig — "
"bitte neu anmelden.",
)
self.run_login()
def _set_logged_out_ui(self):
self.login_action.setText("Anmelden …")
self._restoring = True
self.combo.clear()
self.combo.addItem("— bitte anmelden —", None)
self.combo.setEnabled(False)
self._restoring = False
for action in self.dataset_actions:
action.setEnabled(False)
# ------------------------------------------------------------------
# Verfahrens-Auswahl <-> Projektdatei
# ------------------------------------------------------------------
def on_verfahren_changed(self, index):
if self._restoring or self.combo is None:
return
vkz = self.combo.currentData()
if vkz:
QgsProject.instance().writeEntry(
PROJECT_SCOPE, PROJECT_KEY_VKZ, vkz
)
def restore_project_selection(self):
if self.combo is None or self.combo.count() == 0:
return
vkz, _ok = QgsProject.instance().readEntry(
PROJECT_SCOPE, PROJECT_KEY_VKZ, ""
)
self._restoring = True
index = self.combo.findData(vkz) if vkz else -1
self.combo.setCurrentIndex(index if index >= 0 else 0)
self._restoring = False
def _current_vkz(self):
return self.combo.currentData() if self.combo is not None else None
# ------------------------------------------------------------------
# Laden & Hochladen
# ------------------------------------------------------------------
def run_load_dataset(self, dataset_key):
if not self._ensure_logged_in():
return
vkz = self._current_vkz()
if not vkz:
self._show_error(
"Kein Verfahren gewählt",
"Bitte zuerst in der Auswahlliste ein Verfahren wählen.",
)
return
dataset = DATASETS[dataset_key]
if not self._confirm_replace_existing(dataset_key, vkz, dataset["label"]):
return
path = dataset["path"].format(vkz=vkz)
try:
# Vollständig aus KARTE_OBJEKT: Listen-Endpunkt mit Paging
# (wie das Web-GIS, aber ohne 2000er-Limit).
feature_collection = self.client.load_layer_complete(dataset_key, vkz)
layers = feature_collection_to_layers(
feature_collection,
"%s %s" % (dataset["label"], vkz),
dataset["geometry"],
dataset_key,
vkz,
path,
)
except AuthError:
self._handle_session_expired()
return
except (ApiError, RuntimeError) as exc:
self._show_error("Laden fehlgeschlagen", str(exc))
return
total = sum(layer.featureCount() for layer in layers)
if total == 0:
self.iface.messageBar().pushMessage(
"VLN Karten",
"Der Server hat für VKZ %s keine Objekte im Layer "
"'%s' — es wurde ein leerer Layer zum Digitalisieren "
"angelegt." % (vkz, dataset["label"]),
level=Qgis.Warning,
duration=10,
)
return
detail = ", ".join(
"%s: %d" % (layer.name(), layer.featureCount()) for layer in layers
)
self.iface.messageBar().pushMessage(
"VLN Karten",
"%d Layer geladen (%s)." % (len(layers), detail),
level=Qgis.Success,
duration=5,
)
def _confirm_replace_existing(self, dataset_key, vkz, label):
"""Ersetzt bereits geladene Layer dieses Datensatzes nach
Rückfrage — doppelte Layer-Sätze würden beim Hochladen alle
Objekte dupliziert an den Server schicken."""
existing = plugin_layers(dataset_key, vkz)
if not existing:
return True
answer = QMessageBox.question(
self.iface.mainWindow(),
"Layer ersetzen",
"%s für VKZ %s ist bereits im Projekt geladen (%s).\n"
"Beim Neuladen werden diese Layer ersetzt — nicht "
"hochgeladene Änderungen gehen verloren. Fortfahren?"
% (label, vkz, ", ".join(layer.name() for layer in existing)),
)
if answer != QMessageBox.StandardButton.Yes:
return False
QgsProject.instance().removeMapLayers([layer.id() for layer in existing])
return True
def run_upload_active_layer(self):
if not self._ensure_logged_in():
return
active = self.iface.activeLayer()
path = layer_api_path(active) if active else None
if not path:
self._show_error(
"Hochladen nicht möglich",
"Der aktive Layer wurde nicht über dieses Plugin geladen. "
"Bitte einen über 'VLN Karten' geladenen Layer auswählen.",
)
return
# Der PUT ersetzt den kompletten Layer-Bestand der VKZ auf dem
# Server. Deshalb werden immer alle Teil-Layer dieses Datensatzes
# (Punkte/Linien/Flächen) gemeinsam hochgeladen — nur den aktiven
# zu senden, würde die anderen Geometrietypen serverseitig löschen.
dataset_key = active.customProperty(PROP_DATASET)
vkz = active.customProperty(PROP_VERFAHREN)
siblings = plugin_layers(dataset_key, vkz)
for layer in siblings:
if layer.isEditable() and not layer.commitChanges():
self._show_error(
"Hochladen nicht möglich",
"Die offene Bearbeitungssitzung von '%s' konnte nicht "
"gespeichert werden." % layer.name(),
)
return
dataset = DATASETS.get(dataset_key, {})
total = sum(layer.featureCount() for layer in siblings)
listing = "\n".join(
"• %s (%d Objekte)" % (layer.name(), layer.featureCount())
for layer in siblings
)
answer = QMessageBox.question(
self.iface.mainWindow(),
"Zum Server hochladen",
"%s für VKZ %s übertragen?\n\n%s\n\n"
"Die Layer werden zusammengeführt (%d Objekte); der bisherige "
"Bestand dieser VKZ auf dem Server wird ersetzt."
% (dataset.get("label", dataset_key), vkz, listing, total),
)
if answer != QMessageBox.StandardButton.Yes:
return
try:
feature_collection = layers_to_feature_collection(siblings)
self.client.save_feature_collection(path, feature_collection)
except AuthError:
self._handle_session_expired()
return
except ApiError as exc:
self._show_error("Hochladen fehlgeschlagen", str(exc))
return
self.iface.messageBar().pushMessage(
"VLN Karten",
"%d Objekte aus %d Layer(n) erfolgreich hochgeladen."
% (total, len(siblings)),
level=Qgis.Success,
duration=5,
)
# ------------------------------------------------------------------
# Hilfsfunktionen
# ------------------------------------------------------------------
def _ensure_logged_in(self):
if self.client.is_authenticated:
return True
self.run_login()
return self.client.is_authenticated
def _show_error(self, title, message):
QMessageBox.warning(self.iface.mainWindow(), title, message)