Exporter

cli

allowed_extensions = ('qgz', 'qgs') module-attribute

allowed_output_formats = ('json', 'xml') module-attribute

qgs = QgsApplication([], False) module-attribute

cli()

Source code in src/qgis_server_light/exporter/cli.py

@click.group
def cli():
    pass

export(project, unify_layer_names_by_group=False, output_format=None)

Source code in src/qgis_server_light/exporter/cli.py

@click.option("--project")
@click.option("--unify_layer_names_by_group")
@click.option("--output_format")
@cli.command(
    "export",
    help=f"Export a QGIS project ({f'|'.join(allowed_extensions)}) (1st argument) file to json format",
)
def export(project: str, unify_layer_names_by_group: bool = False, output_format: str | None = None) -> None:
    serializer_config = SerializerConfig(indent="  ")
    if output_format is None:
        output_format = "json"
    if not project.lower().endswith(allowed_extensions):
        raise NotImplementedError(
            f'Allowed qgis project file extensions are: {"|".join(allowed_extensions)} not => {project}'
        )
    if not output_format.lower() in allowed_output_formats:
        raise NotImplementedError(
            f'Allowed output formats are: {"|".join(allowed_output_formats)} not => {output_format}'
        )
    if os.path.isfile(project):
        config = extract(path_to_project=project, unify_layer_names_by_group=bool(unify_layer_names_by_group))
        if output_format == "json":
            click.echo(JsonSerializer(config=serializer_config).render(config))
        elif output_format == "xml":
            click.echo(XmlSerializer(config=serializer_config).render(config))

    else:
        raise AttributeError

extract

create_unified_short_name(name, path)

Source code in src/qgis_server_light/exporter/extract.py

def create_unified_short_name(name: str, path: list[str]):
    short_name_part = path + [name]
    return ".".join(short_name_part)

extract(path_to_project, unify_layer_names_by_group=False)

Extract the styles and configuration of the given Qgis project to the given output directory.

Source code in src/qgis_server_light/exporter/extract.py

def extract(path_to_project: str, unify_layer_names_by_group=False) -> Config:
    """
    Extract the styles and configuration of the given Qgis project
    to the given output directory.
    """
    project, root = get_project_root(path_to_project)
    version, assembled_name = prepare_project_name(project)
    tree = Tree()
    datasets = Datasets()
    config = Config(
        project=Project(name=assembled_name, version=version),
        meta_data=extract_metadata(project),
        tree=tree,
        datasets=datasets,
    )
    extract_entities(project, root, tree, datasets, [], unify_layer_names_by_group)
    return config

extract_entities(project, entity, tree, datasets, path, unify_layer_names_by_group=False)

Source code in src/qgis_server_light/exporter/extract.py

def extract_entities(
    project: QgsProject,
    entity: Union[QgsLayerTree, QgsLayerTreeGroup, QgsLayerTreeLayer],
    tree: Tree,
    datasets: Datasets,
    path: list[str],
    unify_layer_names_by_group=False,
):
    if isinstance(entity, QgsLayerTreeLayer):
        extract_save_layer(
            project, entity, tree, datasets, path, unify_layer_names_by_group
        )

    # If the entity has an attribute `children`, assume it's a group
    elif isinstance(entity, QgsLayerTreeGroup) or isinstance(entity, QgsLayerTree):
        if entity.customProperty("wmsShortName") is not None:
            path = path + [entity.customProperty("wmsShortName")]
        extract_group(entity, tree, datasets, path, unify_layer_names_by_group)
        for child in entity.children():
            extract_entities(
                project, child, tree, datasets, path, unify_layer_names_by_group
            )

extract_fields(layer)

Source code in src/qgis_server_light/exporter/extract.py

def extract_fields(layer: QgsVectorLayer) -> List[Field]:
    fields = []
    for field in layer.fields():
        fields.append(Field(name=field.name(), type=field.typeName()))
    return fields

extract_group(group, tree, datasets, path, unify_layer_names_by_group=False)

Collects data pertaining to a QGIS layer tree group.

Source code in src/qgis_server_light/exporter/extract.py

def extract_group(
    group: QgsLayerTreeGroup,
    tree: Tree,
    datasets: Datasets,
    path: list[str],
    unify_layer_names_by_group=False,
):
    """Collects data pertaining to a QGIS layer tree group."""
    children = []
    for child in group.children():
        if isinstance(child, QgsLayerTreeGroup):
            children.append(child.customProperty("wmsShortName"))
        else:
            if unify_layer_names_by_group:
                children.append(
                    create_unified_short_name(
                        child.layer().shortName() or child.layer().id(), path
                    )
                )
            else:
                children.append(child.layer().shortName() or child.layer().id())
    tree.members.append(
        TreeGroup(
            name=group.customProperty("wmsShortName") or group.name(), children=children
        )
    )
    datasets.group.append(
        Group(
            name=group.customProperty("wmsShortName"),
            title=group.customProperty("wmsTitle"),
        )
    )

extract_metadata(project)

Construct a JSON object from the given layers and metadata

Source code in src/qgis_server_light/exporter/extract.py

def extract_metadata(project) -> MetaData:
    """Construct a JSON object from the given layers and metadata"""
    _meta = project.metadata()
    wms_entries = get_project_server_entries(project, "wms")
    service = Service(**dict(sorted({**wms_entries}.items())))
    return MetaData(
        service=service,
        author=_meta.author(),
        categories=_meta.categories(),
        creationDateTime=_meta.creationDateTime().toPyDateTime().isoformat(),
        language=_meta.language(),
        links=_meta.links(),
    )

extract_save_layer(project, child, tree, datasets, path, unify_layer_names_by_group=False)

Save the given layer to the output path.

Source code in src/qgis_server_light/exporter/extract.py

def extract_save_layer(
    project: QgsProject,
    child: QgsLayerTreeLayer,
    tree: Tree,
    datasets: Datasets,
    path: list[str],
    unify_layer_names_by_group=False,
):
    """Save the given layer to the output path."""
    if isinstance(child, QgsLayerTreeLayer):
        child = child.layer()
    os.path.join("/tmp", f"{str(uuid.uuid4())}.qml")
    style_doc = QDomDocument()
    child.exportNamedStyle(style_doc)
    layer_type = get_layer_type(child)
    decoded = QgsProviderRegistry.instance().decodeUri(
        child.providerType(), child.dataProvider().dataSourceUri()
    )
    for key in decoded:
        if str(decoded[key]) == "None":
            decoded[key] = None
        elif str(decoded[key]) == "NULL":
            decoded[key] = None
        else:
            decoded[key] = str(decoded[key])
        if key == "path":
            decoded[key] = decoded[key].replace(f'{project.readPath("./")}/', "")
    if child.shortName() == "":
        # if layer has no short name we fallback to the qgis_layer_id
        short_name = child.id()
    else:
        short_name = child.shortName()
    if unify_layer_names_by_group:
        short_name = create_unified_short_name(short_name, path)
    crs = Crs(
        postgis_srid=child.dataProvider().crs().postgisSrid(),
        auth_id=child.dataProvider().crs().authid(),
        ogc_uri=child.dataProvider().crs().toOgcUri(),
    )

    extent_wgs_84 = extent_in_wgs84(project, child)
    bbox_wgs84 = BBox(
        x_min=extent_wgs_84[0],
        x_max=extent_wgs_84[2],
        y_min=extent_wgs_84[1],
        y_max=extent_wgs_84[3],
    )
    extent = child.extent()
    bbox = BBox.from_list(
        [
            extent.xMinimum(),
            extent.yMinimum(),
            0.0,
            extent.xMaximum(),
            extent.yMaximum(),
            0.0,
        ]
    )
    if layer_type == "vector":
        source_path = child.source()
        if child.providerType().lower() == "ogr":
            source = DataSource(
                ogr=OgrSource(
                    path=decoded["path"],
                    layer_name=decoded["layerName"],
                    layer_id=decoded["layerId"],
                )
            )
        elif child.providerType().lower() == "postgres":
            config = decoded
            if decoded.get("service"):
                service_config = pgserviceparser.service_config(decoded["service"])
                # merging pg_service content with config of qgis project (qgis project config overwrites
                # pg_service configs
                config = service_config | decoded
            source = DataSource(
                postgres=PostgresSource(
                    dbname=config["dbname"],
                    geometry_column=config["geometrycolumn"],
                    host=config["host"],
                    key=config["key"],
                    password=config["password"],
                    port=config["port"],
                    schema=config["schema"],
                    srid=config["srid"],
                    table=config["table"],
                    type=config["type"],
                    username=config["username"],
                )
            )
            if decoded.get("service"):
                del config["service"]
            source_path = QgsProviderRegistry.instance().encodeUri(
                child.providerType(), config
            )

        elif child.providerType().lower() == "wfs":
            # TODO: Correctly implement source!
            source = WfsSource()
        else:
            raise NotImplementedError(
                f"Unknown provider type: {child.providerType().lower()}"
            )
        fields = extract_fields(child)

        datasets.vector.append(
            Vector(
                path=source_path.replace(f'{project.readPath("./")}/', ""),
                name=short_name,
                title=child.title() or child.name(),
                style=urlsafe_b64encode(style_doc.toByteArray()).decode(),
                driver=child.providerType(),
                bbox_wgs84=bbox_wgs84,
                fields=fields,
                source=source,
                id=child.id(),
                crs=crs,
                bbox=bbox,
                minimum_scale=child.minimumScale(),
                maximum_scale=child.maximumScale(),
            )
        )
    elif layer_type == "raster":
        if child.providerType() == "gdal":
            source = DataSource(
                gdal=GdalSource(path=decoded["path"], layer_name=decoded["layerName"])
            )
        elif child.providerType() == "wms":
            if "tileMatrixSet" in decoded:
                source = DataSource(
                    wmts=WmtsSource(
                        contextual_wms_legend=decoded.get("contextualWMSLegend"),
                        crs=decoded["crs"],
                        dpi_mode=decoded["dpiMode"],
                        feature_count=decoded.get("featureCount"),
                        format=decoded["format"],
                        layers=decoded["layers"],
                        styles=decoded["styles"],
                        tile_dimensions=decoded.get("tileDimensions"),
                        tile_matrix_set=decoded["tileMatrixSet"],
                        tile_pixel_ratio=decoded.get("tilePixelRatio"),
                        url=decoded["url"],
                    )
                )
            else:
                source = DataSource(
                    wms=WmsSource(
                        contextual_wms_legend=decoded["contextualWMSLegend"],
                        crs=decoded["crs"],
                        dpi_mode=decoded["dpiMode"],
                        feature_count=decoded["featureCount"],
                        format=decoded["format"],
                        layers=decoded["layers"],
                        url=decoded["url"],
                    )
                )
        else:
            raise NotImplementedError(
                f"Unknown provider type: {child.providerType().lower()}"
            )
        datasets.raster.append(
            Raster(
                path=child.source().replace(f'{project.readPath("./")}/', ""),
                name=short_name,
                title=child.title(),
                style=urlsafe_b64encode(style_doc.toByteArray()).decode(),
                driver=child.providerType(),
                bbox_wgs84=bbox_wgs84,
                source=source,
                id=child.id(),
                crs=crs,
                bbox=bbox,
                minimum_scale=child.minimumScale(),
                maximum_scale=child.maximumScale(),
            )
        )
    elif layer_type == "custom":
        if child.providerType().lower() == "xyzvectortiles":
            source = DataSource(
                vector_tile=VectorTileSource(
                    styleUrl=decoded["styleUrl"],
                    url=decoded["url"],
                    zmax=decoded["zmax"],
                    zmin=decoded["zmin"],
                    type=decoded["type"],
                )
            )
        else:
            raise NotImplementedError(
                f"Unknown provider type: {child.providerType().lower()}"
            )
            # TODO: make this more configurable
        datasets.custom.append(
            Custom(
                path=child.source().replace(f'{project.readPath("./")}/', ""),
                name=short_name,
                title=child.title(),
                style=urlsafe_b64encode(style_doc.toByteArray()).decode(),
                driver=child.providerType(),
                bbox_wgs84=bbox_wgs84,
                source=source,
                id=child.id(),
                crs=crs,
                bbox=bbox,
                minimum_scale=child.minimumScale(),
                maximum_scale=child.maximumScale(),
            )
        )
    else:
        raise NotImplementedError(f'Unknown layer_type "{layer_type}"')

get_project_root(path_to_project)

Returns a Tuple .

Source code in src/qgis_server_light/exporter/extract.py

def get_project_root(path_to_project) -> Tuple[QgsProject, QgsLayerTree]:
    """Returns a Tuple <Project, LayerTreeRoot>."""
    project = QgsProject.instance()
    project.read(path_to_project)
    return project, project.layerTreeRoot()

prepare_project_name(project)

Source code in src/qgis_server_light/exporter/extract.py

def prepare_project_name(project: QgsProject) -> tuple[str, str]:
    # TODO: Find a good approach to recognize different "versions" of a project.
    name = project.baseName()
    parts = name.split(".")
    version = parts.pop(0)
    assembled_name = ".".join(parts)
    if assembled_name == "":
        assembled_name = project.title()
    return version, assembled_name

save_to_disk(path_to_output_dir, built)

Source code in src/qgis_server_light/exporter/extract.py

def save_to_disk(path_to_output_dir, built):
    path_to_config_json = path.join(path_to_output_dir, "config.json")

    with open(path_to_config_json, "w") as fh:
        json.dump(built, fh, indent=2)

funcs

compose(*fs)

Returns the left-to-right composition of any Iterable of functions. Ex: compose(f, g, h)(args, kwargs) == h(g(f(args, **kwargs)))

Source code in src/qgis_server_light/exporter/funcs.py

def compose(*fs):
    """
    Returns the left-to-right composition of any Iterable of functions.
    Ex: compose(f, g, h)(*args, **kwargs) == h(g(f(*args, **kwargs)))
    """

    def reducer(out, f):
        if isinstance(out, tuple):
            args, kwargs = out
            if isinstance(args, tuple) and isinstance(kwargs, dict):
                return f(*args, **kwargs)

        return f(out)

    def capture_args(*args, **kwargs):
        return reduce(reducer, fs, (args, kwargs))

    return capture_args

extent_in_wgs84(project, layer)

Reprojects the layer's extent using a custom projection. Returns the coordinates as List of floats.

Source code in src/qgis_server_light/exporter/funcs.py

def extent_in_wgs84(project, layer) -> List:
    """
    Reprojects the layer's extent using a custom projection.
    Returns the coordinates as List of floats.
    """
    tr = make_wgs84_geom_transform(project, layer)
    rect = layer.extent()
    reprojected_rect = tr.transform(rect)
    return [
        reprojected_rect.xMinimum(),
        reprojected_rect.yMinimum(),
        reprojected_rect.xMaximum(),
        reprojected_rect.yMaximum(),
    ]

get_layer_type(layer)

Gets the type of the given Qgis layer as a string if the type is supported.

Source code in src/qgis_server_light/exporter/funcs.py

def get_layer_type(layer: QgsVectorLayer | QgsRasterLayer) -> str:
    """Gets the type of the given Qgis layer as a string if the type is supported."""
    if isinstance(layer, QgsRasterLayer):
        return "raster"
    elif isinstance(layer, QgsVectorLayer):
        return "vector"
    elif (
        isinstance(layer, QgsVectorTileLayer)
        or isinstance(layer, QgsTiledSceneLayer)
        or isinstance(layer, QgsPointCloudLayer)
        or isinstance(layer, QgsMeshLayer)
    ):
        return "custom"
    else:
        raise TypeError(f"Not implemented: {layer.type()}")

get_project_server_entries(project, scope_or_scopes)

Gets values from the fields displayed in QGIS under Project > Properties > Server. Returns a Dictionary holding all pairs of found at the corresponding scopes. Example: given scope_or_scope = "wms" (or: ["wms"]) returns { : , : ... } For now the implementation supports only WMS fields but can be easily expanded by adding to the Dictionary below.

Source code in src/qgis_server_light/exporter/funcs.py

def get_project_server_entries(project, scope_or_scopes: Union[str, List]) -> Dict:
    """
    Gets values from the fields displayed in QGIS under Project > Properties > Server.
    Returns a Dictionary holding all pairs of <key, value> found at the corresponding scopes.
    Example:
        given   scope_or_scope = "wms" (or: ["wms"])
        returns { <wms_key1>: <wms_key1_value>, <wms_key2>: <wms_key2_value> ... }
    For now the implementation supports only WMS fields but can be easily expanded by
    adding <key/values> to the Dictionary below.
    """
    supported_scopes = {
        "wms": {
            "scopes": [
                ("WMSContactOrganization", "contact_organization"),
                ("WMSContactMail", "contact_mail"),
                ("WMSContactPerson", "contact_person"),
                ("WMSContactPhone", "contact_phone"),
                ("WMSContactPosition", "contact_position"),
                ("WMSFees", "fees"),
                ("WMSKeywordList", "keyword_list"),
                ("WMSOnlineResource", "online_resource"),
                ("WMSServiceAbstract", "service_abstract"),
                ("WMSServiceTitle", "service_title"),
                ("WMSUrl", "resource_url"),
            ],
            "keys": ["/"],
        }
    }

    scopes = [scope_or_scopes] if isinstance(scope_or_scopes, str) else scope_or_scopes

    for scope in scopes:

        if not scope in supported_scopes:
            supported = ", ".join(supported_scopes.keys())
            error_detail = (
                f"This scope is not supported: {scope}. Supported scopes: {supported}"
            )
            raise ValueError(error_detail)

        scope_entries = supported_scopes[scope]["scopes"]
        key_entries = supported_scopes[scope]["keys"]
        to_collect = zip_longest(scope_entries, key_entries, fillvalue=key_entries[0])

        def collect(acc, pair):
            scope, key = pair
            qgis_scope_name, our_scope_name = scope

            if "list" in qgis_scope_name.lower():
                # PyQGIS sometimes violates Liskov's substitution principle so naming tricks needed
                list_as_text = ", ".join(project.readListEntry(qgis_scope_name, key)[0])
                acc.append((our_scope_name, list_as_text))
            else:
                acc.append((our_scope_name, project.readEntry(qgis_scope_name, key)[0]))

            return acc

        return dict(reduce(collect, to_collect, []))

make_wgs84_geom_transform(project, layer)

Makes a QgisCoordinateTransform to transform a layer to EPSG:4326).

Source code in src/qgis_server_light/exporter/funcs.py

def make_wgs84_geom_transform(project, layer) -> Any:
    """Makes a QgisCoordinateTransform to transform a layer to EPSG:4326)."""
    sourceCrs = layer.crs()
    EPSG_4326 = QgsCoordinateReferenceSystem("EPSG:4326")
    return QgsCoordinateTransform(sourceCrs, EPSG_4326, project)