#2482

Committed 07 Apr 2025 08:38AM UTC coverage: 35.956% (+0.04%) from 35.917%

Build # #2482

Build Type

push

coveralls-python

Committed by

benvanbasten-ns

Commit Message

CHANGES

Run Details

4784 of 13305 relevant lines covered (35.96%)

0.36 hits per line

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

30.19

/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.                                   *
*                                                                         *
***************************************************************************
"""

from hydxlib.scripts import run_import_export
from hydxlib.scripts import write_logging_to_file
from pathlib import Path
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 sqlalchemy.exc import DatabaseError
from sqlalchemy.exc import OperationalError
from threedi_modelchecker import ThreediModelChecker
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

import csv
import os
import shutil


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 = "Perhaps you have selected a geopackage that was created by an older version (< 2.0) of the 3Di Schematisation Editor. In that case, please use the processing algorithm Migrate schematisation database on the Spatialite in the same folder to solve this problem."
                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)
                schema.upgrade(backup=False, upgrade_spatialite_version=True, epsg_code_override=srid,
                               progress_func=feedback_callback)
                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)
        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()
        generated_output_file_path = self.parameterAsFileOutput(
            parameters, self.OUTPUT, context
        )
        self.output_file_path = f"{os.path.splitext(generated_output_file_path)[0]}.csv"
        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
        try:
            with open(self.output_file_path, "w", newline="") as output_file:
                writer = csv.writer(output_file)
                writer.writerow(
                    [
                        "level",
                        "error_code",
                        "id",
                        "table",
                        "column",
                        "value",
                        "description",
                    ]
                )
                for i, check in enumerate(model_checker.checks(level="info")):
                    model_errors = check.get_invalid(session)
                    for error_row in model_errors:
                        writer.writerow(
                            [
                                check.level.name,
                                check.error_code,
                                error_row.id,
                                check.table.name,
                                check.column.name,
                                getattr(error_row, check.column.name),
                                check.description(),
                            ]
                        )
                    checks_passed += 1
                    feedback.setProgress(int(checks_passed * progress_per_check))
        except PermissionError:
            # PermissionError happens for example when a user has the file already open
            # with Excel on Windows, which locks the file.
            feedback.pushWarning(
                f"Not enough permissions to write the file '{self.output_file_path}'.\n\n"
                "The file may be used by another program. Please close all "
                "other programs using the file or select another output "
                "file."
            )
            return {self.OUTPUT: None}

        return {self.OUTPUT: self.output_file_path}

    def postProcessAlgorithm(self, context, feedback):
        if self.add_to_project:
            if self.output_file_path:
                result_layer = QgsVectorLayer(
                    self.output_file_path, "3Di schematisation errors"
                )
                QgsProject.instance().addMapLayer(result_layer)
        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	from hydxlib.scripts import run_import_export	1✔
15	from hydxlib.scripts import write_logging_to_file	1✔
16	from pathlib import Path	1✔
17	from qgis.core import QgsProcessingAlgorithm	1✔
18	from qgis.core import QgsProcessingException	1✔
19	from qgis.core import QgsProcessingParameterBoolean	1✔
20	from qgis.core import QgsProcessingParameterFile	1✔
21	from qgis.core import QgsProcessingParameterFileDestination	1✔
22	from qgis.core import QgsProcessingParameterFolderDestination	1✔
23	from qgis.core import QgsProcessingParameterString	1✔
24	from qgis.core import QgsProject	1✔
25	from qgis.core import QgsVectorLayer	1✔
26	from qgis.PyQt.QtCore import QCoreApplication	1✔
27	from sqlalchemy.exc import DatabaseError	1✔
28	from sqlalchemy.exc import OperationalError	1✔
29	from threedi_modelchecker import ThreediModelChecker	1✔
30	from threedi_results_analysis.processing.download_hydx import download_hydx	1✔
31	from threedi_results_analysis.utils.utils import backup_sqlite	1✔
32	from threedi_schema import errors	1✔
33	from threedi_schema import ThreediDatabase	1✔
34
35	import csv	1✔
36	import os	1✔
37	import shutil	1✔
38
39
40	def get_threedi_database(filename, feedback):	1✔
41	try:	×
42	threedi_db = ThreediDatabase(filename)	×
43	threedi_db.check_connection()	×
44	return threedi_db	×
45	except (OperationalError, DatabaseError):	×
46	feedback.pushWarning("Invalid schematisation file")	×
47	return None	×
48
49
50	def feedback_callback_factory(feedback):	1✔
51	"""Callback function to track schematisation migration progress."""
52
53	def feedback_callback(progres_value, message):	×
54	feedback.setProgress(progres_value)	×
55	feedback.setProgressText(message)	×
56
57	return feedback_callback	×
58
59
60	class MigrateAlgorithm(QgsProcessingAlgorithm):	1✔
61	"""
62	Migrate 3Di model schema to the current version
63	"""
64
65	INPUT = "INPUT"	1✔
66	OUTPUT = "OUTPUT"	1✔
67
68	def initAlgorithm(self, config):	1✔
69	self.addParameter(	×
70	QgsProcessingParameterFile(
71	self.INPUT,
72	"3Di schematisation database",
73	fileFilter="3Di schematisation database (.gpkg .sqlite)"
74	)
75	)
76
77	def processAlgorithm(self, parameters, context, feedback):	1✔
78	filename = self.parameterAsFile(parameters, self.INPUT, context)	×
79	threedi_db = get_threedi_database(filename=filename, feedback=feedback)	×
80	if not threedi_db:	×
81	return {self.OUTPUT: None}	×
82	schema = threedi_db.schema	×
83
84	# Check whether is it not an intermediate legacy geopackage created by
85	# the schematisation editor
86	if filename.endswith(".gpkg"):	×
87	if schema.get_version() < 300:	×
88	warn_msg = "Perhaps you have selected a geopackage that was created by an older version (< 2.0) of the 3Di Schematisation Editor. In that case, please use the processing algorithm Migrate schematisation database on the Spatialite in the same folder to solve this problem."	×
89	feedback.pushWarning(warn_msg)	×
90	return {self.OUTPUT: None}	×
91
92	try:	×
93	schema.validate_schema()	×
94	schema.set_spatial_indexes()	×
95	except errors.MigrationMissingError:	×
96	backup_filepath = backup_sqlite(filename)	×
97
98	srid, _ = schema._get_epsg_data()	×
99	if srid is None:	×
100	try:	×
101	srid = schema._get_dem_epsg()	×
102	except errors.InvalidSRIDException:	×
103	srid = None	×
104	if srid is None:	×
105	feedback.pushWarning(	×
106	"Could not fetch valid EPSG code from database or DEM; aborting database migration."
107	)
108	return {self.OUTPUT: None}	×
109
110	try:	×
111	feedback_callback = feedback_callback_factory(feedback)	×
112	schema.upgrade(backup=False, upgrade_spatialite_version=True, epsg_code_override=srid,	×
113	progress_func=feedback_callback)
114	schema.set_spatial_indexes()	×
115	shutil.rmtree(os.path.dirname(backup_filepath))	×
116	except errors.UpgradeFailedError:	×
117	feedback.pushWarning(	×
118	"The schematisation database cannot be migrated to the current version. Please contact the service desk for assistance."
119	)
120	return {self.OUTPUT: None}	×
121	success = True	×
122	return {self.OUTPUT: success}	×
123
124	def name(self):	1✔
125	"""
126	Returns the algorithm name, used for identifying the algorithm. This
127	string should be fixed for the algorithm, and must not be localised.
128	The name should be unique within each provider. Names should contain
129	lowercase alphanumeric characters only and no spaces or other
130	formatting characters.
131	"""
132	return "migrate"	×
133
134	def displayName(self):	1✔
135	"""
136	Returns the translated algorithm name, which should be used for any
137	user-visible display of the algorithm name.
138	"""
139	return self.tr("Migrate schematisation database")	×
140
141	def group(self):	1✔
142	"""
143	Returns the name of the group this algorithm belongs to. This string
144	should be localised.
145	"""
146	return self.tr(self.groupId())	×
147
148	def groupId(self):	1✔
149	"""
150	Returns the unique ID of the group this algorithm belongs to. This
151	string should be fixed for the algorithm, and must not be localised.
152	The group id should be unique within each provider. Group id should
153	contain lowercase alphanumeric characters only and no spaces or other
154	formatting characters.
155	"""
156	return "Schematisation"	×
157
158	def tr(self, string):	1✔
159	return QCoreApplication.translate("Processing", string)	×
160
161	def createInstance(self):	1✔
162	return MigrateAlgorithm()	×
163
164
165	class CheckSchematisationAlgorithm(QgsProcessingAlgorithm):	1✔
166	"""
167	Run the schematisation checker
168	"""
169
170	INPUT = "INPUT"	1✔
171	OUTPUT = "OUTPUT"	1✔
172	ADD_TO_PROJECT = "ADD_TO_PROJECT"	1✔
173
174	def initAlgorithm(self, config):	1✔
175	self.addParameter(	×
176	QgsProcessingParameterFile(
177	self.INPUT, self.tr("3Di Schematisation"), fileFilter="GeoPackage (*.gpkg)"
178	)
179	)
180
181	self.addParameter(	×
182	QgsProcessingParameterFileDestination(
183	self.OUTPUT, self.tr("Output"), fileFilter="csv"
184	)
185	)
186
187	self.addParameter(	×
188	QgsProcessingParameterBoolean(
189	self.ADD_TO_PROJECT, self.tr("Add result to project"), defaultValue=True
190	)
191	)
192
193	def processAlgorithm(self, parameters, context, feedback):	1✔
194	self.add_to_project = self.parameterAsBoolean(	×
195	parameters, self.ADD_TO_PROJECT, context
196	)
197	self.output_file_path = None	×
198	input_filename = self.parameterAsFile(parameters, self.INPUT, context)	×
199	threedi_db = get_threedi_database(filename=input_filename, feedback=feedback)	×
200	if not threedi_db:	×
201	return {self.OUTPUT: None}	×
202	try:	×
203	model_checker = ThreediModelChecker(threedi_db)	×
204	except errors.MigrationMissingError:	×
205	feedback.pushWarning(	×
206	"The selected 3Di model does not have the latest migration. Please "
207	"migrate your model to the latest version."
208	)
209	return {self.OUTPUT: None}	×
210	schema = threedi_db.schema	×
211	schema.set_spatial_indexes()	×
212	generated_output_file_path = self.parameterAsFileOutput(	×
213	parameters, self.OUTPUT, context
214	)
215	self.output_file_path = f"{os.path.splitext(generated_output_file_path)[0]}.csv"	×
216	session = model_checker.db.get_session()	×
217	session.model_checker_context = model_checker.context	×
218	total_checks = len(model_checker.config.checks)	×
219	progress_per_check = 100.0 / total_checks	×
220	checks_passed = 0	×
221	try:	×
222	with open(self.output_file_path, "w", newline="") as output_file:	×
223	writer = csv.writer(output_file)	×
224	writer.writerow(	×
225	[
226	"level",
227	"error_code",
228	"id",
229	"table",
230	"column",
231	"value",
232	"description",
233	]
234	)
235	for i, check in enumerate(model_checker.checks(level="info")):	×
236	model_errors = check.get_invalid(session)	×
237	for error_row in model_errors:	×
238	writer.writerow(	×
239	[
240	check.level.name,
241	check.error_code,
242	error_row.id,
243	check.table.name,
244	check.column.name,
245	getattr(error_row, check.column.name),
246	check.description(),
247	]
248	)
249	checks_passed += 1	×
250	feedback.setProgress(int(checks_passed * progress_per_check))	×
251	except PermissionError:	×
252	# PermissionError happens for example when a user has the file already open
253	# with Excel on Windows, which locks the file.
254	feedback.pushWarning(	×
255	f"Not enough permissions to write the file '{self.output_file_path}'.\n\n"
256	"The file may be used by another program. Please close all "
257	"other programs using the file or select another output "
258	"file."
259	)
260	return {self.OUTPUT: None}	×
261
262	return {self.OUTPUT: self.output_file_path}	×
263
264	def postProcessAlgorithm(self, context, feedback):	1✔
265	if self.add_to_project:	×
266	if self.output_file_path:	×
267	result_layer = QgsVectorLayer(	×
268	self.output_file_path, "3Di schematisation errors"
269	)
270	QgsProject.instance().addMapLayer(result_layer)	×
271	return {self.OUTPUT: self.output_file_path}	×
272
273	def name(self):	1✔
274	"""
275	Returns the algorithm name, used for identifying the algorithm. This
276	string should be fixed for the algorithm, and must not be localised.
277	The name should be unique within each provider. Names should contain
278	lowercase alphanumeric characters only and no spaces or other
279	formatting characters.
280	"""
281	return "check_schematisation"	×
282
283	def displayName(self):	1✔
284	"""
285	Returns the translated algorithm name, which should be used for any
286	user-visible display of the algorithm name.
287	"""
288	return self.tr("Check Schematisation")	×
289
290	def group(self):	1✔
291	"""
292	Returns the name of the group this algorithm belongs to. This string
293	should be localised.
294	"""
295	return self.tr(self.groupId())	×
296
297	def groupId(self):	1✔
298	"""
299	Returns the unique ID of the group this algorithm belongs to. This
300	string should be fixed for the algorithm, and must not be localised.
301	The group id should be unique within each provider. Group id should
302	contain lowercase alphanumeric characters only and no spaces or other
303	formatting characters.
304	"""
305	return "Schematisation"	×
306
307	def tr(self, string):	1✔
308	return QCoreApplication.translate("Processing", string)	×
309
310	def createInstance(self):	1✔
311	return CheckSchematisationAlgorithm()	×
312
313
314	class ImportHydXAlgorithm(QgsProcessingAlgorithm):	1✔
315	"""
316	Import data from GWSW HydX to a 3Di Schematisation
317	"""
318
319	INPUT_DATASET_NAME = "INPUT_DATASET_NAME"	1✔
320	HYDX_DOWNLOAD_DIRECTORY = "HYDX_DOWNLOAD_DIRECTORY"	1✔
321	INPUT_HYDX_DIRECTORY = "INPUT_HYDX_DIRECTORY"	1✔
322	TARGET_SCHEMATISATION = "TARGET_SCHEMATISATION"	1✔
323
324	def initAlgorithm(self, config):	1✔
325	self.addParameter(	×
326	QgsProcessingParameterFile(
327	self.TARGET_SCHEMATISATION, "Target 3Di Schematisation", fileFilter="GeoPackage (*.gpkg)"
328	)
329	)
330
331	self.addParameter(	×
332	QgsProcessingParameterFile(
333	self.INPUT_HYDX_DIRECTORY,
334	"GWSW HydX directory (local)",
335	behavior=QgsProcessingParameterFile.Folder,
336	optional=True,
337	)
338	)
339
340	self.addParameter(	×
341	QgsProcessingParameterString(
342	self.INPUT_DATASET_NAME, "GWSW dataset name (online)", optional=True
343	)
344	)
345
346	self.addParameter(	×
347	QgsProcessingParameterFolderDestination(
348	self.HYDX_DOWNLOAD_DIRECTORY,
349	"Destination directory for GWSW HydX dataset download",
350	optional=True,
351	)
352	)
353
354	def processAlgorithm(self, parameters, context, feedback):	1✔
355	hydx_dataset_name = self.parameterAsString(	×
356	parameters, self.INPUT_DATASET_NAME, context
357	)
358	hydx_download_dir = self.parameterAsString(	×
359	parameters, self.HYDX_DOWNLOAD_DIRECTORY, context
360	)
361	hydx_path = self.parameterAsString(	×
362	parameters, self.INPUT_HYDX_DIRECTORY, context
363	)
364	out_path = self.parameterAsFile(parameters, self.TARGET_SCHEMATISATION, context)	×
365	threedi_db = get_threedi_database(filename=out_path, feedback=feedback)	×
366	if not threedi_db:	×
367	raise QgsProcessingException(	×
368	f"Unable to connect to 3Di schematisation '{out_path}'"
369	)
370	try:	×
371	schema = threedi_db.schema	×
372	schema.validate_schema()	×
373
374	except errors.MigrationMissingError:	×
375	raise QgsProcessingException(	×
376	"The selected 3Di schematisation does not have the latest database schema version. Please migrate this "
377	"schematisation and try again: Processing > Toolbox > 3Di > Schematisation > Migrate schematisation database"
378	)
379	if not (hydx_dataset_name or hydx_path):	×
380	raise QgsProcessingException(	×
381	"Either 'GWSW HydX directory (local)' or 'GWSW dataset name (online)' must be filled in!"
382	)
383	if hydx_dataset_name and hydx_path:	×
384	feedback.pushWarning(	×
385	"Both 'GWSW dataset name (online)' and 'GWSW HydX directory (local)' are filled in. "
386	"'GWSW dataset name (online)' will be ignored. This dataset will not be downloaded."
387	)
388	elif hydx_dataset_name:	×
389	try:	×
390	hydx_download_path = Path(hydx_download_dir)	×
391	hydx_download_dir_is_valid = hydx_download_path.is_dir()	×
392	except TypeError:	×
393	hydx_download_dir_is_valid = False	×
394	if parameters[self.HYDX_DOWNLOAD_DIRECTORY] == "TEMPORARY_OUTPUT":	×
395	hydx_download_dir_is_valid = True	×
396	if not hydx_download_dir_is_valid:	×
397	raise QgsProcessingException(	×
398	f"'Destination directory for HydX dataset download' ({hydx_download_path}) is not a valid directory"
399	)
400	hydx_path = download_hydx(	×
401	dataset_name=hydx_dataset_name,
402	target_directory=hydx_download_path,
403	wait_times=[0.1, 1, 2, 3, 4, 5, 10],
404	feedback=feedback,
405	)
406	# hydx_path will be None if user has canceled the process during download
407	if feedback.isCanceled():	×
408	raise QgsProcessingException("Process canceled")	×
409	if hydx_path is None:	×
410	raise QgsProcessingException("Error in retrieving dataset (note case-sensitivity)")	×
411	feedback.pushInfo(f"Starting import of {hydx_path} to {out_path}")	×
412	log_path = Path(out_path).parent / "import_hydx.log"	×
413	write_logging_to_file(log_path)	×
414	feedback.pushInfo(f"Logging will be written to {log_path}")	×
415	run_import_export(hydx_path=hydx_path, out_path=out_path)	×
416	return {}	×
417
418	def name(self):	1✔
419	"""
420	Returns the algorithm name, used for identifying the algorithm. This
421	string should be fixed for the algorithm, and must not be localised.
422	The name should be unique within each provider. Names should contain
423	lowercase alphanumeric characters only and no spaces or other
424	formatting characters.
425	"""
426	return "import_hydx"	×
427
428	def displayName(self):	1✔
429	"""
430	Returns the translated algorithm name, which should be used for any
431	user-visible display of the algorithm name.
432	"""
433	return self.tr("Import GWSW HydX")	×
434
435	def shortHelpString(self):	1✔
436	return """	×
437	<h3>Introduction</h3>
438	<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>
439	<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>
440	<h3>Parameters</h3>
441	<h4>Target 3Di Schematisation</h4>
442	<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>
443	<h4>GWSW HydX directory (local)</h4>
444	<p>Use this option if you have already downloaded a GWSW HydX dataset to a local directory.</p>
445	<h4>GWSW dataset name (online)</h4>
446	<p>Use this option if you want to download a GWSW HydX dataset.</p>
447	<h4>Destination directory for GWSW HydX dataset download</h4>
448	<p>If you have chosen to download a GWSW HydX dataset, this is the directory it will be downloaded to.</p>
449	"""
450
451	def group(self):	1✔
452	"""
453	Returns the name of the group this algorithm belongs to. This string
454	should be localised.
455	"""
456	return self.tr(self.groupId())	×
457
458	def groupId(self):	1✔
459	"""
460	Returns the unique ID of the group this algorithm belongs to. This
461	string should be fixed for the algorithm, and must not be localised.
462	The group id should be unique within each provider. Group id should
463	contain lowercase alphanumeric characters only and no spaces or other
464	formatting characters.
465	"""
466	return "Schematisation"	×
467
468	def tr(self, string):	1✔
469	return QCoreApplication.translate("Processing", string)	×
470
471	def createInstance(self):	1✔
472	return ImportHydXAlgorithm()	×

nens / ThreeDiToolbox / #2482

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