#2523

Committed 26 May 2025 08:43PM UTC coverage: 34.895% (-0.1%) from 35.003%

Build # #2523

Build Type

push

coveralls-python

Committed by

web-flow

Commit Message

Merge f0e2a318c into 67c8900c8

Run Details

7 of 74 new or added lines in 1 file covered. (9.46%)

4 existing lines in 1 file now uncovered.

4744 of 13595 relevant lines covered (34.9%)

0.35 hits per line

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

25.55

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

nens / ThreeDiToolbox / #2523

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