API_Karte_QGISDemo/vln_karten/layer_manager.py

"""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