#2522

Committed 26 May 2025 06:53PM UTC coverage: 34.893% (-0.1%) from 35.003%

Build # #2522

Build Type

push

coveralls-python

Committed by

web-flow

Commit Message

Merge acd188079 into 67c8900c8

Run Details

6 of 72 new or added lines in 1 file covered. (8.33%)

4 existing lines in 1 file now uncovered.

4743 of 13593 relevant lines covered (34.89%)

0.35 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

25.37

/processing/schematisation_algorithms.py

# -*- coding: utf-8 -*-

"""
***************************************************************************
*                                                                         *
*   This program is free software; you can redistribute it and/or modify  *
*   it under the terms of the GNU General Public License as published by  *
*   the Free Software Foundation; either version 2 of the License, or     *
*   (at your option) any later version.                                   *
*                                                                         *
***************************************************************************
"""

import os
import shutil
import warnings

from pathlib import Path

from hydxlib.scripts import run_import_export
from hydxlib.scripts import write_logging_to_file

from osgeo import ogr, osr
from osgeo import gdal
from qgis.core import QgsProcessingAlgorithm
from qgis.core import QgsProcessingException
from qgis.core import QgsProcessingParameterBoolean
from qgis.core import QgsProcessingParameterFile
from qgis.core import QgsProcessingParameterFileDestination
from qgis.core import QgsProcessingParameterFolderDestination
from qgis.core import QgsProcessingParameterString
from qgis.core import QgsProject
from qgis.core import QgsVectorLayer
from qgis.PyQt.QtCore import QCoreApplication
from shapely import wkb  # Import Shapely wkb for geometry handling
from sqlalchemy.exc import DatabaseError
from sqlalchemy.exc import OperationalError

from threedi_modelchecker import ThreediModelChecker
from threedi_modelchecker.exporters import export_with_geom
from threedi_results_analysis.processing.download_hydx import download_hydx
from threedi_results_analysis.utils.utils import backup_sqlite
from threedi_schema import errors
from threedi_schema import ThreediDatabase

STYLE_DIR = Path(__file__).parent / "styles"


def get_threedi_database(filename, feedback):
    try:
        threedi_db = ThreediDatabase(filename)
        threedi_db.check_connection()
        return threedi_db
    except (OperationalError, DatabaseError):
        feedback.pushWarning("Invalid schematisation file")
        return None


def feedback_callback_factory(feedback):
    """Callback function to track schematisation migration progress."""

    def feedback_callback(progres_value, message):
        feedback.setProgress(progres_value)
        feedback.setProgressText(message)

    return feedback_callback


class MigrateAlgorithm(QgsProcessingAlgorithm):
    """
    Migrate 3Di model schema to the current version
    """

    INPUT = "INPUT"
    OUTPUT = "OUTPUT"

    def initAlgorithm(self, config):
        self.addParameter(
            QgsProcessingParameterFile(
                self.INPUT,
                "3Di schematisation database",
                fileFilter="3Di schematisation database (*.gpkg *.sqlite)"
            )
        )

    def processAlgorithm(self, parameters, context, feedback):
        filename = self.parameterAsFile(parameters, self.INPUT, context)
        threedi_db = get_threedi_database(filename=filename, feedback=feedback)
        if not threedi_db:
            return {self.OUTPUT: None}
        schema = threedi_db.schema

        # Check whether is it not an intermediate legacy geopackage created by
        # the schematisation editor
        if filename.endswith(".gpkg"):
            if schema.get_version() < 300:
                warn_msg = "The selected file is not a valid 3Di schematisation database.\n\nYou may have selected a geopackage that was created by an older version of the 3Di Schematisation Editor (before version 2.0). In that case, there will probably be a Spatialite (*.sqlite) in the same folder. Please use that file instead."
                feedback.pushWarning(warn_msg)
                return {self.OUTPUT: None}

        try:
            schema.validate_schema()
            schema.set_spatial_indexes()
        except errors.MigrationMissingError:
            backup_filepath = backup_sqlite(filename)

            srid, _ = schema._get_epsg_data()
            if srid is None:
                try:
                    srid = schema._get_dem_epsg()
                except errors.InvalidSRIDException:
                    srid = None
            if srid is None:
                feedback.pushWarning(
                    "Could not fetch valid EPSG code from database or DEM; aborting database migration."
                )
                return {self.OUTPUT: None}

            try:
                feedback_callback = feedback_callback_factory(feedback)
                with warnings.catch_warnings(record=True) as w:
                    warnings.simplefilter("always", UserWarning)
                    schema.upgrade(backup=False, epsg_code_override=srid, progress_func=feedback_callback)
                if w:
                    for warning in w:
                        feedback.pushWarning(f'{warning._category_name}: {warning.message}')
                schema.set_spatial_indexes()
                shutil.rmtree(os.path.dirname(backup_filepath))
            except errors.UpgradeFailedError:
                feedback.pushWarning(
                    "The schematisation database cannot be migrated to the current version. Please contact the service desk for assistance."
                )
                return {self.OUTPUT: None}
        success = True
        return {self.OUTPUT: success}

    def name(self):
        """
        Returns the algorithm name, used for identifying the algorithm. This
        string should be fixed for the algorithm, and must not be localised.
        The name should be unique within each provider. Names should contain
        lowercase alphanumeric characters only and no spaces or other
        formatting characters.
        """
        return "migrate"

    def displayName(self):
        """
        Returns the translated algorithm name, which should be used for any
        user-visible display of the algorithm name.
        """
        return self.tr("Migrate schematisation database")

    def group(self):
        """
        Returns the name of the group this algorithm belongs to. This string
        should be localised.
        """
        return self.tr(self.groupId())

    def groupId(self):
        """
        Returns the unique ID of the group this algorithm belongs to. This
        string should be fixed for the algorithm, and must not be localised.
        The group id should be unique within each provider. Group id should
        contain lowercase alphanumeric characters only and no spaces or other
        formatting characters.
        """
        return "Schematisation"

    def tr(self, string):
        return QCoreApplication.translate("Processing", string)

    def createInstance(self):
        return MigrateAlgorithm()


class CheckSchematisationAlgorithm(QgsProcessingAlgorithm):
    """
    Run the schematisation checker
    """

    INPUT = "INPUT"
    OUTPUT = "OUTPUT"
    ADD_TO_PROJECT = "ADD_TO_PROJECT"

    def initAlgorithm(self, config):
        self.addParameter(
            QgsProcessingParameterFile(
                self.INPUT, self.tr("3Di Schematisation"), fileFilter="GeoPackage (*.gpkg)"
            )
        )

        self.addParameter(
            QgsProcessingParameterFileDestination(
                self.OUTPUT, self.tr("Output"), fileFilter="csv"
            )
        )

        self.addParameter(
            QgsProcessingParameterBoolean(
                self.ADD_TO_PROJECT, self.tr("Add result to project"), defaultValue=True
            )
        )

    def processAlgorithm(self, parameters, context, feedback):
        self.add_to_project = self.parameterAsBoolean(
            parameters, self.ADD_TO_PROJECT, context
        )
        self.output_file_path = None
        input_filename = self.parameterAsFile(parameters, self.INPUT, context)
        self.schema_name = Path(input_filename).stem
        threedi_db = get_threedi_database(filename=input_filename, feedback=feedback)
        if not threedi_db:
            return {self.OUTPUT: None}
        try:
            model_checker = ThreediModelChecker(threedi_db)
        except errors.MigrationMissingError:
            feedback.pushWarning(
                "The selected 3Di model does not have the latest migration. Please "
                "migrate your model to the latest version."
            )
            return {self.OUTPUT: None}
        schema = threedi_db.schema
        schema.set_spatial_indexes()
        srid, _ = schema._get_epsg_data()
        generated_output_file_path = self.parameterAsFileOutput(
            parameters, self.OUTPUT, context
        )
        self.output_file_path = f"{os.path.splitext(generated_output_file_path)[0]}.gpkg"
        session = model_checker.db.get_session()
        session.model_checker_context = model_checker.context
        total_checks = len(model_checker.config.checks)
        progress_per_check = 100.0 / total_checks
        checks_passed = 0
        error_list = []
        for i, check in enumerate(model_checker.checks(level="info")):
            model_errors = check.get_invalid(session)
            error_list += [[check, error_row] for error_row in model_errors]
            checks_passed += 1
            feedback.setProgress(int(checks_passed * progress_per_check))
        error_details = export_with_geom(error_list)

        # Create an output GeoPackage
        gdal.UseExceptions()
        driver = ogr.GetDriverByName("GPKG")
        data_source = driver.CreateDataSource(self.output_file_path)
        if data_source is None:
            feedback.pushWarning(
                f"Unable to create the GeoPackage '{self.output_file_path}', check the directory permissions."
            )
            return {self.OUTPUT: None}

        # Loop through the error_details and group by geometry type
        grouped_errors = {'Point': [], 'LineString': [], 'Polygon': [], 'Table': []}
        for error in error_details:
            geom = wkb.loads(error.geom.data) if error.geom is not None else None
            if geom is None or geom.geom_type not in grouped_errors.keys():
                feature_type = 'Table'
            else:
                feature_type = geom.geom_type
            grouped_errors[feature_type].append(error)

        group_name_map = {'LineString': 'Line'}
        for feature_type, errors_group in grouped_errors.items():
            group_name = f'{group_name_map.get(feature_type, feature_type)} features'
            if feature_type == 'Table':
                # Create a table for non-geometry errors
                layer = data_source.CreateLayer(
                    group_name, None, ogr.wkbNone, options=["OVERWRITE=YES"]
                )
            else:
                # Create a layer for each type of geometry
                spatial_ref = osr.SpatialReference()
                spatial_ref.ImportFromEPSG(srid)
                layer = data_source.CreateLayer(
                    group_name,
                    srs=spatial_ref,
                    geom_type=getattr(ogr, f"wkb{feature_type}"),
                    options=["OVERWRITE=YES"],
                    )

            # Add fields
            layer.CreateField(ogr.FieldDefn("level", ogr.OFTString))
            layer.CreateField(ogr.FieldDefn("error_code", ogr.OFTString))
            layer.CreateField(ogr.FieldDefn("id", ogr.OFTString))
            layer.CreateField(ogr.FieldDefn("table", ogr.OFTString))
            layer.CreateField(ogr.FieldDefn("column", ogr.OFTString))
            layer.CreateField(ogr.FieldDefn("value", ogr.OFTString))
            layer.CreateField(ogr.FieldDefn("description", ogr.OFTString))

            defn = layer.GetLayerDefn()
            for error in errors_group:
                feat = ogr.Feature(defn)
                feat.SetField("level", error.name)
                feat.SetField("error_code", error.code)
                feat.SetField("id", error.id)
                feat.SetField("table", error.table)
                feat.SetField("column", error.column)
                feat.SetField("value", error.value)
                feat.SetField("description", error.description)
                if feature_type != 'Table':
                    geom = wkb.loads(error.geom.data)  # Convert WKB to a Shapely geometry object
                    feat.SetGeometry(ogr.CreateGeometryFromWkb(geom.wkb))  # Convert back to OGR-compatible WKB
                layer.CreateFeature(feat)

        feedback.pushInfo(f"GeoPackage successfully written to {self.output_file_path}")
        return {self.OUTPUT: self.output_file_path}

    def postProcessAlgorithm(self, context, feedback):
        if self.add_to_project and self.output_file_path:
            # Create a group for the GeoPackage layers
            group = QgsProject.instance().layerTreeRoot().insertGroup(0, f'Check results: {self.schema_name}')
            # Add all layers in the geopackage to the group
            conn = ogr.Open(self.output_file_path)
            if conn:
                for i in range(conn.GetLayerCount()):
                    layer_name = conn.GetLayerByIndex(i).GetName()
                    layer_uri = f"{self.output_file_path}|layername={layer_name}"
                    layer = QgsVectorLayer(layer_uri, layer_name.replace('errors_', '', 1), "ogr")
                    if layer.isValid():
                        added_layer = QgsProject.instance().addMapLayer(layer, False)
                        added_layer.loadNamedStyle(
                            str(STYLE_DIR / f"checker_{added_layer.geometryType().name.lower()}.qml")
                        )
                        group.addLayer(added_layer)
                    else:
                        feedback.reportError(f"Layer {layer_name} is not valid")
                conn = None  # Close the connection
            else:
                feedback.reportError(f"Could not open GeoPackage file: {self.output_file_path}")
        return {self.OUTPUT: self.output_file_path}

    def name(self):
        """
        Returns the algorithm name, used for identifying the algorithm. This
        string should be fixed for the algorithm, and must not be localised.
        The name should be unique within each provider. Names should contain
        lowercase alphanumeric characters only and no spaces or other
        formatting characters.
        """
        return "check_schematisation"

    def displayName(self):
        """
        Returns the translated algorithm name, which should be used for any
        user-visible display of the algorithm name.
        """
        return self.tr("Check Schematisation")

    def group(self):
        """
        Returns the name of the group this algorithm belongs to. This string
        should be localised.
        """
        return self.tr(self.groupId())

    def groupId(self):
        """
        Returns the unique ID of the group this algorithm belongs to. This
        string should be fixed for the algorithm, and must not be localised.
        The group id should be unique within each provider. Group id should
        contain lowercase alphanumeric characters only and no spaces or other
        formatting characters.
        """
        return "Schematisation"

    def tr(self, string):
        return QCoreApplication.translate("Processing", string)

    def createInstance(self):
        return CheckSchematisationAlgorithm()


class ImportHydXAlgorithm(QgsProcessingAlgorithm):
    """
    Import data from GWSW HydX to a 3Di Schematisation
    """

    INPUT_DATASET_NAME = "INPUT_DATASET_NAME"
    HYDX_DOWNLOAD_DIRECTORY = "HYDX_DOWNLOAD_DIRECTORY"
    INPUT_HYDX_DIRECTORY = "INPUT_HYDX_DIRECTORY"
    TARGET_SCHEMATISATION = "TARGET_SCHEMATISATION"

    def initAlgorithm(self, config):
        self.addParameter(
            QgsProcessingParameterFile(
                self.TARGET_SCHEMATISATION, "Target 3Di Schematisation", fileFilter="GeoPackage (*.gpkg)"
            )
        )

        self.addParameter(
            QgsProcessingParameterFile(
                self.INPUT_HYDX_DIRECTORY,
                "GWSW HydX directory (local)",
                behavior=QgsProcessingParameterFile.Folder,
                optional=True,
            )
        )

        self.addParameter(
            QgsProcessingParameterString(
                self.INPUT_DATASET_NAME, "GWSW dataset name (online)", optional=True
            )
        )

        self.addParameter(
            QgsProcessingParameterFolderDestination(
                self.HYDX_DOWNLOAD_DIRECTORY,
                "Destination directory for GWSW HydX dataset download",
                optional=True,
            )
        )

    def processAlgorithm(self, parameters, context, feedback):
        hydx_dataset_name = self.parameterAsString(
            parameters, self.INPUT_DATASET_NAME, context
        )
        hydx_download_dir = self.parameterAsString(
            parameters, self.HYDX_DOWNLOAD_DIRECTORY, context
        )
        hydx_path = self.parameterAsString(
            parameters, self.INPUT_HYDX_DIRECTORY, context
        )
        out_path = self.parameterAsFile(parameters, self.TARGET_SCHEMATISATION, context)
        threedi_db = get_threedi_database(filename=out_path, feedback=feedback)
        if not threedi_db:
            raise QgsProcessingException(
                f"Unable to connect to 3Di schematisation '{out_path}'"
            )
        try:
            schema = threedi_db.schema
            schema.validate_schema()

        except errors.MigrationMissingError:
            raise QgsProcessingException(
                "The selected 3Di schematisation does not have the latest database schema version. Please migrate this "
                "schematisation and try again: Processing > Toolbox > 3Di > Schematisation > Migrate schematisation database"
            )
        if not (hydx_dataset_name or hydx_path):
            raise QgsProcessingException(
                "Either 'GWSW HydX directory (local)' or 'GWSW dataset name (online)' must be filled in!"
            )
        if hydx_dataset_name and hydx_path:
            feedback.pushWarning(
                "Both 'GWSW dataset name (online)' and 'GWSW HydX directory (local)' are filled in. "
                "'GWSW dataset name (online)' will be ignored. This dataset will not be downloaded."
            )
        elif hydx_dataset_name:
            try:
                hydx_download_path = Path(hydx_download_dir)
                hydx_download_dir_is_valid = hydx_download_path.is_dir()
            except TypeError:
                hydx_download_dir_is_valid = False
            if parameters[self.HYDX_DOWNLOAD_DIRECTORY] == "TEMPORARY_OUTPUT":
                hydx_download_dir_is_valid = True
            if not hydx_download_dir_is_valid:
                raise QgsProcessingException(
                    f"'Destination directory for HydX dataset download' ({hydx_download_path}) is not a valid directory"
                )
            hydx_path = download_hydx(
                dataset_name=hydx_dataset_name,
                target_directory=hydx_download_path,
                wait_times=[0.1, 1, 2, 3, 4, 5, 10],
                feedback=feedback,
            )
            # hydx_path will be None if user has canceled the process during download
            if feedback.isCanceled():
                raise QgsProcessingException("Process canceled")
            if hydx_path is None:
                raise QgsProcessingException("Error in retrieving dataset (note case-sensitivity)")
        feedback.pushInfo(f"Starting import of {hydx_path} to {out_path}")
        log_path = Path(out_path).parent / "import_hydx.log"
        write_logging_to_file(log_path)
        feedback.pushInfo(f"Logging will be written to {log_path}")
        run_import_export(hydx_path=hydx_path, out_path=out_path)
        return {}

    def name(self):
        """
        Returns the algorithm name, used for identifying the algorithm. This
        string should be fixed for the algorithm, and must not be localised.
        The name should be unique within each provider. Names should contain
        lowercase alphanumeric characters only and no spaces or other
        formatting characters.
        """
        return "import_hydx"

    def displayName(self):
        """
        Returns the translated algorithm name, which should be used for any
        user-visible display of the algorithm name.
        """
        return self.tr("Import GWSW HydX")

    def shortHelpString(self):
        return """
        <h3>Introduction</h3>
        <p>Use this processing algorithm to import data in the format of the Dutch "Gegevenswoordenboek Stedelijk Water (GWSW)". Either select a previously downloaded local dataset, or download a dataset directly from the server.</p>
        <p>A log file will be created in the same directory as the Target 3Di schematisation. Please check this log file after the import has completed.&nbsp;&nbsp;</p>
        <h3>Parameters</h3>
        <h4>Target 3Di Schematisation</h4>
        <p>GeoPackage (.gpkg) file that contains the layers required by 3Di. Imported data will be added to any data already contained in the 3Di schematisation.</p>
        <h4>GWSW HydX directory (local)</h4>
        <p>Use this option if you have already downloaded a GWSW HydX dataset to a local directory.</p>
        <h4>GWSW dataset name (online)</h4>
        <p>Use this option if you want to download a GWSW HydX dataset.</p>
        <h4>Destination directory for GWSW HydX dataset download</h4>
        <p>If you have chosen to download a GWSW HydX dataset, this is the directory it will be downloaded to.</p>
        """

    def group(self):
        """
        Returns the name of the group this algorithm belongs to. This string
        should be localised.
        """
        return self.tr(self.groupId())

    def groupId(self):
        """
        Returns the unique ID of the group this algorithm belongs to. This
        string should be fixed for the algorithm, and must not be localised.
        The group id should be unique within each provider. Group id should
        contain lowercase alphanumeric characters only and no spaces or other
        formatting characters.
        """
        return "Schematisation"

    def tr(self, string):
        return QCoreApplication.translate("Processing", string)

    def createInstance(self):
        return ImportHydXAlgorithm()

1	# -- coding: utf-8 --
2
3	"""	1✔
4	***************************************************************************
5	* *
6	* This program is free software; you can redistribute it and/or modify *
7	* it under the terms of the GNU General Public License as published by *
8	* the Free Software Foundation; either version 2 of the License, or *
9	* (at your option) any later version. *
10	* *
11	***************************************************************************
12	"""
13
14	import os	1✔
15	import shutil	1✔
16	import warnings	1✔
17
18	from pathlib import Path	1✔
19
20	from hydxlib.scripts import run_import_export	1✔
21	from hydxlib.scripts import write_logging_to_file	1✔
22
23	from osgeo import ogr, osr	1✔
24	from osgeo import gdal	1✔
25	from qgis.core import QgsProcessingAlgorithm	1✔
26	from qgis.core import QgsProcessingException	1✔
27	from qgis.core import QgsProcessingParameterBoolean	1✔
28	from qgis.core import QgsProcessingParameterFile	1✔
29	from qgis.core import QgsProcessingParameterFileDestination	1✔
30	from qgis.core import QgsProcessingParameterFolderDestination	1✔
31	from qgis.core import QgsProcessingParameterString	1✔
32	from qgis.core import QgsProject	1✔
33	from qgis.core import QgsVectorLayer	1✔
34	from qgis.PyQt.QtCore import QCoreApplication	1✔
35	from shapely import wkb # Import Shapely wkb for geometry handling	1✔
36	from sqlalchemy.exc import DatabaseError	1✔
37	from sqlalchemy.exc import OperationalError	1✔
38
39	from threedi_modelchecker import ThreediModelChecker	1✔
40	from threedi_modelchecker.exporters import export_with_geom	1✔
41	from threedi_results_analysis.processing.download_hydx import download_hydx	1✔
42	from threedi_results_analysis.utils.utils import backup_sqlite	1✔
43	from threedi_schema import errors	1✔
44	from threedi_schema import ThreediDatabase	1✔
45
46	STYLE_DIR = Path(__file__).parent / "styles"	1✔
47
48
49	def get_threedi_database(filename, feedback):	1✔
50	try:	×
51	threedi_db = ThreediDatabase(filename)	×
52	threedi_db.check_connection()	×
53	return threedi_db	×
54	except (OperationalError, DatabaseError):	×
55	feedback.pushWarning("Invalid schematisation file")	×
56	return None	×
57
58
59	def feedback_callback_factory(feedback):	1✔
60	"""Callback function to track schematisation migration progress."""
61
62	def feedback_callback(progres_value, message):	×
63	feedback.setProgress(progres_value)	×
64	feedback.setProgressText(message)	×
65
66	return feedback_callback	×
67
68
69	class MigrateAlgorithm(QgsProcessingAlgorithm):	1✔
70	"""
71	Migrate 3Di model schema to the current version
72	"""
73
74	INPUT = "INPUT"	1✔
75	OUTPUT = "OUTPUT"	1✔
76
77	def initAlgorithm(self, config):	1✔
78	self.addParameter(	×
79	QgsProcessingParameterFile(
80	self.INPUT,
81	"3Di schematisation database",
82	fileFilter="3Di schematisation database (.gpkg .sqlite)"
83	)
84	)
85
86	def processAlgorithm(self, parameters, context, feedback):	1✔
87	filename = self.parameterAsFile(parameters, self.INPUT, context)	×
88	threedi_db = get_threedi_database(filename=filename, feedback=feedback)	×
89	if not threedi_db:	×
90	return {self.OUTPUT: None}	×
91	schema = threedi_db.schema	×
92
93	# Check whether is it not an intermediate legacy geopackage created by
94	# the schematisation editor
95	if filename.endswith(".gpkg"):	×
96	if schema.get_version() < 300:	×
97	warn_msg = "The selected file is not a valid 3Di schematisation database.\n\nYou may have selected a geopackage that was created by an older version of the 3Di Schematisation Editor (before version 2.0). In that case, there will probably be a Spatialite (*.sqlite) in the same folder. Please use that file instead."	×
98	feedback.pushWarning(warn_msg)	×
99	return {self.OUTPUT: None}	×
100
101	try:	×
102	schema.validate_schema()	×
103	schema.set_spatial_indexes()	×
104	except errors.MigrationMissingError:	×
105	backup_filepath = backup_sqlite(filename)	×
106
107	srid, _ = schema._get_epsg_data()	×
108	if srid is None:	×
109	try:	×
110	srid = schema._get_dem_epsg()	×
111	except errors.InvalidSRIDException:	×
112	srid = None	×
113	if srid is None:	×
114	feedback.pushWarning(	×
115	"Could not fetch valid EPSG code from database or DEM; aborting database migration."
116	)
117	return {self.OUTPUT: None}	×
118
119	try:	×
120	feedback_callback = feedback_callback_factory(feedback)	×
121	with warnings.catch_warnings(record=True) as w:	×
122	warnings.simplefilter("always", UserWarning)	×
123	schema.upgrade(backup=False, epsg_code_override=srid, progress_func=feedback_callback)	×
124	if w:	×
125	for warning in w:	×
126	feedback.pushWarning(f'{warning._category_name}: {warning.message}')	×
127	schema.set_spatial_indexes()	×
128	shutil.rmtree(os.path.dirname(backup_filepath))	×
129	except errors.UpgradeFailedError:	×
130	feedback.pushWarning(	×
131	"The schematisation database cannot be migrated to the current version. Please contact the service desk for assistance."
132	)
133	return {self.OUTPUT: None}	×
134	success = True	×
135	return {self.OUTPUT: success}	×
136
137	def name(self):	1✔
138	"""
139	Returns the algorithm name, used for identifying the algorithm. This
140	string should be fixed for the algorithm, and must not be localised.
141	The name should be unique within each provider. Names should contain
142	lowercase alphanumeric characters only and no spaces or other
143	formatting characters.
144	"""
145	return "migrate"	×
146
147	def displayName(self):	1✔
148	"""
149	Returns the translated algorithm name, which should be used for any
150	user-visible display of the algorithm name.
151	"""
152	return self.tr("Migrate schematisation database")	×
153
154	def group(self):	1✔
155	"""
156	Returns the name of the group this algorithm belongs to. This string
157	should be localised.
158	"""
159	return self.tr(self.groupId())	×
160
161	def groupId(self):	1✔
162	"""
163	Returns the unique ID of the group this algorithm belongs to. This
164	string should be fixed for the algorithm, and must not be localised.
165	The group id should be unique within each provider. Group id should
166	contain lowercase alphanumeric characters only and no spaces or other
167	formatting characters.
168	"""
169	return "Schematisation"	×
170
171	def tr(self, string):	1✔
172	return QCoreApplication.translate("Processing", string)	×
173
174	def createInstance(self):	1✔
175	return MigrateAlgorithm()	×
176
177
178	class CheckSchematisationAlgorithm(QgsProcessingAlgorithm):	1✔
179	"""
180	Run the schematisation checker
181	"""
182
183	INPUT = "INPUT"	1✔
184	OUTPUT = "OUTPUT"	1✔
185	ADD_TO_PROJECT = "ADD_TO_PROJECT"	1✔
186
187	def initAlgorithm(self, config):	1✔
188	self.addParameter(	×
189	QgsProcessingParameterFile(
190	self.INPUT, self.tr("3Di Schematisation"), fileFilter="GeoPackage (*.gpkg)"
191	)
192	)
193
194	self.addParameter(	×
195	QgsProcessingParameterFileDestination(
196	self.OUTPUT, self.tr("Output"), fileFilter="csv"
197	)
198	)
199
200	self.addParameter(	×
201	QgsProcessingParameterBoolean(
202	self.ADD_TO_PROJECT, self.tr("Add result to project"), defaultValue=True
203	)
204	)
205
206	def processAlgorithm(self, parameters, context, feedback):	1✔
207	self.add_to_project = self.parameterAsBoolean(	×
208	parameters, self.ADD_TO_PROJECT, context
209	)
210	self.output_file_path = None	×
211	input_filename = self.parameterAsFile(parameters, self.INPUT, context)	×
NEW 212	self.schema_name = Path(input_filename).stem	×
213	threedi_db = get_threedi_database(filename=input_filename, feedback=feedback)	×
214	if not threedi_db:	×
215	return {self.OUTPUT: None}	×
216	try:	×
217	model_checker = ThreediModelChecker(threedi_db)	×
218	except errors.MigrationMissingError:	×
219	feedback.pushWarning(	×
220	"The selected 3Di model does not have the latest migration. Please "
221	"migrate your model to the latest version."
222	)
223	return {self.OUTPUT: None}	×
224	schema = threedi_db.schema	×
225	schema.set_spatial_indexes()	×
NEW 226	srid, _ = schema._get_epsg_data()	×
UNCOV 227	generated_output_file_path = self.parameterAsFileOutput(	×
228	parameters, self.OUTPUT, context
229	)
NEW 230	self.output_file_path = f"{os.path.splitext(generated_output_file_path)[0]}.gpkg"	×
231	session = model_checker.db.get_session()	×
232	session.model_checker_context = model_checker.context	×
233	total_checks = len(model_checker.config.checks)	×
234	progress_per_check = 100.0 / total_checks	×
235	checks_passed = 0	×
NEW 236	error_list = []	×
NEW 237	for i, check in enumerate(model_checker.checks(level="info")):	×
NEW 238	model_errors = check.get_invalid(session)	×
NEW 239	error_list += [[check, error_row] for error_row in model_errors]	×
NEW 240	checks_passed += 1	×
NEW 241	feedback.setProgress(int(checks_passed * progress_per_check))	×
NEW 242	error_details = export_with_geom(error_list)	×
243
244	# Create an output GeoPackage
NEW 245	gdal.UseExceptions()	×
NEW 246	driver = ogr.GetDriverByName("GPKG")	×
NEW 247	data_source = driver.CreateDataSource(self.output_file_path)	×
NEW 248	if data_source is None:	×
UNCOV 249	feedback.pushWarning(	×
250	f"Unable to create the GeoPackage '{self.output_file_path}', check the directory permissions."
251	)
252	return {self.OUTPUT: None}	×
253
254	# Loop through the error_details and group by geometry type
NEW 255	grouped_errors = {'Point': [], 'LineString': [], 'Polygon': [], 'Table': []}	×
NEW 256	for error in error_details:	×
NEW 257	geom = wkb.loads(error.geom.data) if error.geom is not None else None	×
NEW 258	if geom is None or geom.geom_type not in grouped_errors.keys():	×
NEW 259	feature_type = 'Table'	×
260	else:
NEW 261	feature_type = geom.geom_type	×
NEW 262	grouped_errors[feature_type].append(error)	×
263
NEW 264	group_name_map = {'LineString': 'Line'}	×
NEW 265	for feature_type, errors_group in grouped_errors.items():	×
NEW 266	group_name = f'{group_name_map.get(feature_type, feature_type)} features'	×
NEW 267	if feature_type == 'Table':	×
268	# Create a table for non-geometry errors
NEW 269	layer = data_source.CreateLayer(	×
270	group_name, None, ogr.wkbNone, options=["OVERWRITE=YES"]
271	)
272	else:
273	# Create a layer for each type of geometry
NEW 274	spatial_ref = osr.SpatialReference()	×
NEW 275	spatial_ref.ImportFromEPSG(srid)	×
NEW 276	layer = data_source.CreateLayer(	×
277	group_name,
278	srs=spatial_ref,
279	geom_type=getattr(ogr, f"wkb{feature_type}"),
280	options=["OVERWRITE=YES"],
281	)
282
283	# Add fields
NEW 284	layer.CreateField(ogr.FieldDefn("level", ogr.OFTString))	×
NEW 285	layer.CreateField(ogr.FieldDefn("error_code", ogr.OFTString))	×
NEW 286	layer.CreateField(ogr.FieldDefn("id", ogr.OFTString))	×
NEW 287	layer.CreateField(ogr.FieldDefn("table", ogr.OFTString))	×
NEW 288	layer.CreateField(ogr.FieldDefn("column", ogr.OFTString))	×
NEW 289	layer.CreateField(ogr.FieldDefn("value", ogr.OFTString))	×
NEW 290	layer.CreateField(ogr.FieldDefn("description", ogr.OFTString))	×
291
NEW 292	defn = layer.GetLayerDefn()	×
NEW 293	for error in errors_group:	×
NEW 294	feat = ogr.Feature(defn)	×
NEW 295	feat.SetField("level", error.name)	×
NEW 296	feat.SetField("error_code", error.code)	×
NEW 297	feat.SetField("id", error.id)	×
NEW 298	feat.SetField("table", error.table)	×
NEW 299	feat.SetField("column", error.column)	×
NEW 300	feat.SetField("value", error.value)	×
NEW 301	feat.SetField("description", error.description)	×
NEW 302	if feature_type != 'Table':	×
NEW 303	geom = wkb.loads(error.geom.data) # Convert WKB to a Shapely geometry object	×
NEW 304	feat.SetGeometry(ogr.CreateGeometryFromWkb(geom.wkb)) # Convert back to OGR-compatible WKB	×
NEW 305	layer.CreateFeature(feat)	×
306
NEW 307	feedback.pushInfo(f"GeoPackage successfully written to {self.output_file_path}")	×
UNCOV 308	return {self.OUTPUT: self.output_file_path}	×
309
310	def postProcessAlgorithm(self, context, feedback):	1✔
NEW 311	if self.add_to_project and self.output_file_path:	×
312	# Create a group for the GeoPackage layers
NEW 313	group = QgsProject.instance().layerTreeRoot().insertGroup(0, f'Check results: {self.schema_name}')	×
314	# Add all layers in the geopackage to the group
NEW 315	conn = ogr.Open(self.output_file_path)	×
NEW 316	if conn:	×
NEW 317	for i in range(conn.GetLayerCount()):	×
NEW 318	layer_name = conn.GetLayerByIndex(i).GetName()	×
NEW 319	layer_uri = f"{self.output_file_path}\|layername={layer_name}"	×
NEW 320	layer = QgsVectorLayer(layer_uri, layer_name.replace('errors_', '', 1), "ogr")	×
NEW 321	if layer.isValid():	×
NEW 322	added_layer = QgsProject.instance().addMapLayer(layer, False)	×
NEW 323	added_layer.loadNamedStyle(	×
324	str(STYLE_DIR / f"checker_{added_layer.geometryType().name.lower()}.qml")
325	)
NEW 326	group.addLayer(added_layer)	×
327	else:
NEW 328	feedback.reportError(f"Layer {layer_name} is not valid")	×
NEW 329	conn = None # Close the connection	×
330	else:
NEW 331	feedback.reportError(f"Could not open GeoPackage file: {self.output_file_path}")	×
UNCOV 332	return {self.OUTPUT: self.output_file_path}	×
333
334	def name(self):	1✔
335	"""
336	Returns the algorithm name, used for identifying the algorithm. This
337	string should be fixed for the algorithm, and must not be localised.
338	The name should be unique within each provider. Names should contain
339	lowercase alphanumeric characters only and no spaces or other
340	formatting characters.
341	"""
342	return "check_schematisation"	×
343
344	def displayName(self):	1✔
345	"""
346	Returns the translated algorithm name, which should be used for any
347	user-visible display of the algorithm name.
348	"""
349	return self.tr("Check Schematisation")	×
350
351	def group(self):	1✔
352	"""
353	Returns the name of the group this algorithm belongs to. This string
354	should be localised.
355	"""
356	return self.tr(self.groupId())	×
357
358	def groupId(self):	1✔
359	"""
360	Returns the unique ID of the group this algorithm belongs to. This
361	string should be fixed for the algorithm, and must not be localised.
362	The group id should be unique within each provider. Group id should
363	contain lowercase alphanumeric characters only and no spaces or other
364	formatting characters.
365	"""
366	return "Schematisation"	×
367
368	def tr(self, string):	1✔
369	return QCoreApplication.translate("Processing", string)	×
370
371	def createInstance(self):	1✔
372	return CheckSchematisationAlgorithm()	×
373
374
375	class ImportHydXAlgorithm(QgsProcessingAlgorithm):	1✔
376	"""
377	Import data from GWSW HydX to a 3Di Schematisation
378	"""
379
380	INPUT_DATASET_NAME = "INPUT_DATASET_NAME"	1✔
381	HYDX_DOWNLOAD_DIRECTORY = "HYDX_DOWNLOAD_DIRECTORY"	1✔
382	INPUT_HYDX_DIRECTORY = "INPUT_HYDX_DIRECTORY"	1✔
383	TARGET_SCHEMATISATION = "TARGET_SCHEMATISATION"	1✔
384
385	def initAlgorithm(self, config):	1✔
386	self.addParameter(	×
387	QgsProcessingParameterFile(
388	self.TARGET_SCHEMATISATION, "Target 3Di Schematisation", fileFilter="GeoPackage (*.gpkg)"
389	)
390	)
391
392	self.addParameter(	×
393	QgsProcessingParameterFile(
394	self.INPUT_HYDX_DIRECTORY,
395	"GWSW HydX directory (local)",
396	behavior=QgsProcessingParameterFile.Folder,
397	optional=True,
398	)
399	)
400
401	self.addParameter(	×
402	QgsProcessingParameterString(
403	self.INPUT_DATASET_NAME, "GWSW dataset name (online)", optional=True
404	)
405	)
406
407	self.addParameter(	×
408	QgsProcessingParameterFolderDestination(
409	self.HYDX_DOWNLOAD_DIRECTORY,
410	"Destination directory for GWSW HydX dataset download",
411	optional=True,
412	)
413	)
414
415	def processAlgorithm(self, parameters, context, feedback):	1✔
416	hydx_dataset_name = self.parameterAsString(	×
417	parameters, self.INPUT_DATASET_NAME, context
418	)
419	hydx_download_dir = self.parameterAsString(	×
420	parameters, self.HYDX_DOWNLOAD_DIRECTORY, context
421	)
422	hydx_path = self.parameterAsString(	×
423	parameters, self.INPUT_HYDX_DIRECTORY, context
424	)
425	out_path = self.parameterAsFile(parameters, self.TARGET_SCHEMATISATION, context)	×
426	threedi_db = get_threedi_database(filename=out_path, feedback=feedback)	×
427	if not threedi_db:	×
428	raise QgsProcessingException(	×
429	f"Unable to connect to 3Di schematisation '{out_path}'"
430	)
431	try:	×
432	schema = threedi_db.schema	×
433	schema.validate_schema()	×
434
435	except errors.MigrationMissingError:	×
436	raise QgsProcessingException(	×
437	"The selected 3Di schematisation does not have the latest database schema version. Please migrate this "
438	"schematisation and try again: Processing > Toolbox > 3Di > Schematisation > Migrate schematisation database"
439	)
440	if not (hydx_dataset_name or hydx_path):	×
441	raise QgsProcessingException(	×
442	"Either 'GWSW HydX directory (local)' or 'GWSW dataset name (online)' must be filled in!"
443	)
444	if hydx_dataset_name and hydx_path:	×
445	feedback.pushWarning(	×
446	"Both 'GWSW dataset name (online)' and 'GWSW HydX directory (local)' are filled in. "
447	"'GWSW dataset name (online)' will be ignored. This dataset will not be downloaded."
448	)
449	elif hydx_dataset_name:	×
450	try:	×
451	hydx_download_path = Path(hydx_download_dir)	×
452	hydx_download_dir_is_valid = hydx_download_path.is_dir()	×
453	except TypeError:	×
454	hydx_download_dir_is_valid = False	×
455	if parameters[self.HYDX_DOWNLOAD_DIRECTORY] == "TEMPORARY_OUTPUT":	×
456	hydx_download_dir_is_valid = True	×
457	if not hydx_download_dir_is_valid:	×
458	raise QgsProcessingException(	×
459	f"'Destination directory for HydX dataset download' ({hydx_download_path}) is not a valid directory"
460	)
461	hydx_path = download_hydx(	×
462	dataset_name=hydx_dataset_name,
463	target_directory=hydx_download_path,
464	wait_times=[0.1, 1, 2, 3, 4, 5, 10],
465	feedback=feedback,
466	)
467	# hydx_path will be None if user has canceled the process during download
468	if feedback.isCanceled():	×
469	raise QgsProcessingException("Process canceled")	×
470	if hydx_path is None:	×
471	raise QgsProcessingException("Error in retrieving dataset (note case-sensitivity)")	×
472	feedback.pushInfo(f"Starting import of {hydx_path} to {out_path}")	×
473	log_path = Path(out_path).parent / "import_hydx.log"	×
474	write_logging_to_file(log_path)	×
475	feedback.pushInfo(f"Logging will be written to {log_path}")	×
476	run_import_export(hydx_path=hydx_path, out_path=out_path)	×
477	return {}	×
478
479	def name(self):	1✔
480	"""
481	Returns the algorithm name, used for identifying the algorithm. This
482	string should be fixed for the algorithm, and must not be localised.
483	The name should be unique within each provider. Names should contain
484	lowercase alphanumeric characters only and no spaces or other
485	formatting characters.
486	"""
487	return "import_hydx"	×
488
489	def displayName(self):	1✔
490	"""
491	Returns the translated algorithm name, which should be used for any
492	user-visible display of the algorithm name.
493	"""
494	return self.tr("Import GWSW HydX")	×
495
496	def shortHelpString(self):	1✔
497	return """	×
498	<h3>Introduction</h3>
499	<p>Use this processing algorithm to import data in the format of the Dutch "Gegevenswoordenboek Stedelijk Water (GWSW)". Either select a previously downloaded local dataset, or download a dataset directly from the server.</p>
500	<p>A log file will be created in the same directory as the Target 3Di schematisation. Please check this log file after the import has completed.  </p>
501	<h3>Parameters</h3>
502	<h4>Target 3Di Schematisation</h4>
503	<p>GeoPackage (.gpkg) file that contains the layers required by 3Di. Imported data will be added to any data already contained in the 3Di schematisation.</p>
504	<h4>GWSW HydX directory (local)</h4>
505	<p>Use this option if you have already downloaded a GWSW HydX dataset to a local directory.</p>
506	<h4>GWSW dataset name (online)</h4>
507	<p>Use this option if you want to download a GWSW HydX dataset.</p>
508	<h4>Destination directory for GWSW HydX dataset download</h4>
509	<p>If you have chosen to download a GWSW HydX dataset, this is the directory it will be downloaded to.</p>
510	"""
511
512	def group(self):	1✔
513	"""
514	Returns the name of the group this algorithm belongs to. This string
515	should be localised.
516	"""
517	return self.tr(self.groupId())	×
518
519	def groupId(self):	1✔
520	"""
521	Returns the unique ID of the group this algorithm belongs to. This
522	string should be fixed for the algorithm, and must not be localised.
523	The group id should be unique within each provider. Group id should
524	contain lowercase alphanumeric characters only and no spaces or other
525	formatting characters.
526	"""
527	return "Schematisation"	×
528
529	def tr(self, string):	1✔
530	return QCoreApplication.translate("Processing", string)	×
531
532	def createInstance(self):	1✔
533	return ImportHydXAlgorithm()	×

nens / ThreeDiToolbox / #2522

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous