#2497

Committed 30 Apr 2025 03:24PM UTC coverage: 35.309%. Remained the same

Build # #2497

Build Type

push

coveralls-python

Committed by

web-flow

Commit Message

Fix path handling in computational grid processing algorithm (#1103)

Run Details

0 of 1 new or added line in 1 file covered. (0.0%)

4594 of 13011 relevant lines covered (35.31%)

0.35 hits per line

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

27.43

/processing/grid_creation_algorithm.py

from collections import OrderedDict
from qgis.core import QgsProcessingAlgorithm
from qgis.core import QgsProcessingException
from qgis.core import QgsProcessingParameterFile
from qgis.core import QgsProcessingParameterFileDestination
from qgis.core import QgsVectorLayer
from qgis.PyQt.QtCore import QCoreApplication
from threedi_results_analysis.processing.processing_utils import gridadmin2geopackage
from threedi_results_analysis.processing.processing_utils import (
    load_computational_layers,
)
from threedigrid_builder import make_gridadmin
from threedigrid_builder import SchematisationError

import io
import logging
import os


class ThreeDiGenerateCompGridAlgorithm(QgsProcessingAlgorithm):
    """
    Generate a gridadmin.h5 file out of schematisation database and convert it to GeoPackage.
    Created layers will be added to the map canvas after successful conversion.
    """

    INPUT_SCHEMATISATION = "INPUT_SCHEMATISATION"
    OUTPUT = "OUTPUT"
    LAYERS_TO_ADD = OrderedDict()

    def flags(self):
        return super().flags() | QgsProcessingAlgorithm.FlagNoThreading

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

    def createInstance(self):
        return ThreeDiGenerateCompGridAlgorithm()

    def name(self):
        return "threedi_generate_computational_grid"

    def displayName(self):
        return self.tr("Computational grid from schematisation")

    def group(self):
        return self.tr("Computational Grid")

    def groupId(self):
        return "computational_grid"

    def shortHelpString(self):
        return self.tr("Generate computational grid from schematization")

    def initAlgorithm(self, config=None):

        self.addParameter(
            QgsProcessingParameterFile(
                self.INPUT_SCHEMATISATION,
                self.tr("Input schematisation file"),
                behavior=QgsProcessingParameterFile.File,
                fileFilter="GeoPackage (*.gpkg);;Spatialite (*.sqlite)",
            )
        )

        self.addParameter(
            QgsProcessingParameterFileDestination(
                self.OUTPUT, self.tr("Output computational grid file"), fileFilter="*.gpkg",
            )
        )

    def processAlgorithm(self, parameters, context, feedback):
        input_schematisation = self.parameterAsString(parameters, self.INPUT_SCHEMATISATION, context)
        if not input_schematisation:
            raise QgsProcessingException(self.invalidSourceError(parameters, self.INPUT_SCHEMATISATION))

        set_dem_rel_path = self._get_rel_dem_path(input_schematisation, feedback)
        if set_dem_rel_path:
            input_schematisation_dir = os.path.dirname(input_schematisation)
            set_dem_path = os.path.join(input_schematisation_dir, set_dem_rel_path)
            feedback.pushInfo(f"DEM raster referenced in schematisation settings:\n{set_dem_path}")
            if not os.path.exists(set_dem_path):
                set_dem_path = None
                info = "The DEM referenced in the schematisation settings doesn't exist - skipping."
                feedback.pushInfo(info)
        else:
            set_dem_path = None
            info = "There is no DEM file referenced in the schematisation settings - skipping."
            feedback.pushInfo(info)
        output_gpkg_file = self.parameterAsFileOutput(parameters, self.OUTPUT, context)
        if output_gpkg_file is None:
            raise QgsProcessingException(self.invalidSourceError(parameters, self.OUTPUT))
        # If user is writing to the temporary file then QGIS adds '.file' extension, so we need to change it.
        output_file_without_extension = output_gpkg_file.rsplit(".", 1)[0]
        gridadmin_file = f"{output_file_without_extension}.h5"
        if output_gpkg_file.endswith(".file"):
            output_gpkg_file = f"{output_file_without_extension}.gpkg"

        def progress_rep(progress, info):
            feedback.setProgress(int(progress * 100))
            feedback.pushInfo(info)

        # Capture threedigridbuilder logging
        logger = logging.getLogger("threedigrid_builder.grid.connection_nodes")
        assert logger.hasHandlers()  # Check whether we have the right one
        log_capture_string = io.StringIO()
        ch = logging.StreamHandler(log_capture_string)
        ch.setFormatter(logging.Formatter(fmt='%(levelname)-8s :: %(message)s'))
        ch.setLevel(logging.DEBUG)
        logger.addHandler(ch)
        try:
            make_gridadmin(input_schematisation, set_dem_path, gridadmin_file, progress_callback=progress_rep)
        except SchematisationError as e:
            err = f"Creating grid file failed with the following error: {repr(e)}"
            raise QgsProcessingException(err)
        finally:
            # Pull the contents back into a string and close the stream
            log_contents = log_capture_string.getvalue()
            log_capture_string.close()
            logger.removeHandler(ch)
            if log_contents:
                feedback.pushWarning("3Di gridbuilder log:")
                feedback.pushWarning(log_contents)

        feedback.setProgress(0)
        gpkg_layers = gridadmin2geopackage(gridadmin_file, output_gpkg_file, context, feedback)
        self.LAYERS_TO_ADD.update(gpkg_layers)

        return {self.OUTPUT: output_gpkg_file}

    def postProcessAlgorithm(self, context, feedback):
        project = context.project()
        load_computational_layers(self.LAYERS_TO_ADD, project)
        self.LAYERS_TO_ADD.clear()
        return {}

    def _get_schematisation_version(self, input_schematisation: str) -> int:
        uri = input_schematisation + "|layername=schema_version"
        schema_lyr = QgsVectorLayer(uri, "schema_version", "ogr")
        if schema_lyr.isValid() and schema_lyr.featureCount() == 1:
            # Take first (and only)
            meta = next(schema_lyr.getFeatures())
            return int(meta["version_num"])
        else:
            return None

    def _get_rel_dem_path(self, input_schematisation: str, feedback) -> str:

        schematisation_version = self._get_schematisation_version(input_schematisation)
        setting_uri = None
        if schematisation_version < 222:
            setting_uri = input_schematisation + "|layername=v2_global_settings"
        else:
            setting_uri = input_schematisation + "|layername=model_settings"

        feedback.pushInfo(f"Reading DEM file from: {setting_uri}")
        settings_lyr = QgsVectorLayer(setting_uri, "settings", "ogr")
        if settings_lyr.isValid() and settings_lyr.featureCount() == 1:
            settings_feat = next(settings_lyr.getFeatures())
            set_dem_rel_path = settings_feat["dem_file"]
        else:
            err = f"No (or multiple) global settings entries in {setting_uri}" "Check your schematisation file."
            raise QgsProcessingException(f"Incorrect input schematisation file:\n{err}")

        if schematisation_version >= 222:
            set_dem_rel_path = os.path.join("rasters", str(set_dem_rel_path))

        return set_dem_rel_path

1	from collections import OrderedDict	1✔
2	from qgis.core import QgsProcessingAlgorithm	1✔
3	from qgis.core import QgsProcessingException	1✔
4	from qgis.core import QgsProcessingParameterFile	1✔
5	from qgis.core import QgsProcessingParameterFileDestination	1✔
6	from qgis.core import QgsVectorLayer	1✔
7	from qgis.PyQt.QtCore import QCoreApplication	1✔
8	from threedi_results_analysis.processing.processing_utils import gridadmin2geopackage	1✔
9	from threedi_results_analysis.processing.processing_utils import (	1✔
10	load_computational_layers,
11	)
12	from threedigrid_builder import make_gridadmin	1✔
13	from threedigrid_builder import SchematisationError	1✔
14
15	import io	1✔
16	import logging	1✔
17	import os	1✔
18
19
20	class ThreeDiGenerateCompGridAlgorithm(QgsProcessingAlgorithm):	1✔
21	"""
22	Generate a gridadmin.h5 file out of schematisation database and convert it to GeoPackage.
23	Created layers will be added to the map canvas after successful conversion.
24	"""
25
26	INPUT_SCHEMATISATION = "INPUT_SCHEMATISATION"	1✔
27	OUTPUT = "OUTPUT"	1✔
28	LAYERS_TO_ADD = OrderedDict()	1✔
29
30	def flags(self):	1✔
31	return super().flags() \| QgsProcessingAlgorithm.FlagNoThreading	×
32
33	def tr(self, string):	1✔
34	return QCoreApplication.translate("Processing", string)	×
35
36	def createInstance(self):	1✔
37	return ThreeDiGenerateCompGridAlgorithm()	×
38
39	def name(self):	1✔
40	return "threedi_generate_computational_grid"	×
41
42	def displayName(self):	1✔
43	return self.tr("Computational grid from schematisation")	×
44
45	def group(self):	1✔
46	return self.tr("Computational Grid")	×
47
48	def groupId(self):	1✔
49	return "computational_grid"	×
50
51	def shortHelpString(self):	1✔
52	return self.tr("Generate computational grid from schematization")	×
53
54	def initAlgorithm(self, config=None):	1✔
55
56	self.addParameter(	×
57	QgsProcessingParameterFile(
58	self.INPUT_SCHEMATISATION,
59	self.tr("Input schematisation file"),
60	behavior=QgsProcessingParameterFile.File,
61	fileFilter="GeoPackage (.gpkg);;Spatialite (.sqlite)",
62	)
63	)
64
65	self.addParameter(	×
66	QgsProcessingParameterFileDestination(
67	self.OUTPUT, self.tr("Output computational grid file"), fileFilter="*.gpkg",
68	)
69	)
70
71	def processAlgorithm(self, parameters, context, feedback):	1✔
72	input_schematisation = self.parameterAsString(parameters, self.INPUT_SCHEMATISATION, context)	×
73	if not input_schematisation:	×
74	raise QgsProcessingException(self.invalidSourceError(parameters, self.INPUT_SCHEMATISATION))	×
75
76	set_dem_rel_path = self._get_rel_dem_path(input_schematisation, feedback)	×
77	if set_dem_rel_path:	×
78	input_schematisation_dir = os.path.dirname(input_schematisation)	×
79	set_dem_path = os.path.join(input_schematisation_dir, set_dem_rel_path)	×
80	feedback.pushInfo(f"DEM raster referenced in schematisation settings:\n{set_dem_path}")	×
81	if not os.path.exists(set_dem_path):	×
82	set_dem_path = None	×
83	info = "The DEM referenced in the schematisation settings doesn't exist - skipping."	×
84	feedback.pushInfo(info)	×
85	else:
86	set_dem_path = None	×
87	info = "There is no DEM file referenced in the schematisation settings - skipping."	×
88	feedback.pushInfo(info)	×
89	output_gpkg_file = self.parameterAsFileOutput(parameters, self.OUTPUT, context)	×
90	if output_gpkg_file is None:	×
91	raise QgsProcessingException(self.invalidSourceError(parameters, self.OUTPUT))	×
92	# If user is writing to the temporary file then QGIS adds '.file' extension, so we need to change it.
93	output_file_without_extension = output_gpkg_file.rsplit(".", 1)[0]	×
94	gridadmin_file = f"{output_file_without_extension}.h5"	×
95	if output_gpkg_file.endswith(".file"):	×
96	output_gpkg_file = f"{output_file_without_extension}.gpkg"	×
97
98	def progress_rep(progress, info):	×
99	feedback.setProgress(int(progress * 100))	×
100	feedback.pushInfo(info)	×
101
102	# Capture threedigridbuilder logging
103	logger = logging.getLogger("threedigrid_builder.grid.connection_nodes")	×
104	assert logger.hasHandlers() # Check whether we have the right one	×
105	log_capture_string = io.StringIO()	×
106	ch = logging.StreamHandler(log_capture_string)	×
107	ch.setFormatter(logging.Formatter(fmt='%(levelname)-8s :: %(message)s'))	×
108	ch.setLevel(logging.DEBUG)	×
109	logger.addHandler(ch)	×
110	try:	×
111	make_gridadmin(input_schematisation, set_dem_path, gridadmin_file, progress_callback=progress_rep)	×
112	except SchematisationError as e:	×
113	err = f"Creating grid file failed with the following error: {repr(e)}"	×
114	raise QgsProcessingException(err)	×
115	finally:
116	# Pull the contents back into a string and close the stream
117	log_contents = log_capture_string.getvalue()	×
118	log_capture_string.close()	×
119	logger.removeHandler(ch)	×
120	if log_contents:	×
121	feedback.pushWarning("3Di gridbuilder log:")	×
122	feedback.pushWarning(log_contents)	×
123
124	feedback.setProgress(0)	×
125	gpkg_layers = gridadmin2geopackage(gridadmin_file, output_gpkg_file, context, feedback)	×
126	self.LAYERS_TO_ADD.update(gpkg_layers)	×
127
128	return {self.OUTPUT: output_gpkg_file}	×
129
130	def postProcessAlgorithm(self, context, feedback):	1✔
131	project = context.project()	×
132	load_computational_layers(self.LAYERS_TO_ADD, project)	×
133	self.LAYERS_TO_ADD.clear()	×
134	return {}	×
135
136	def _get_schematisation_version(self, input_schematisation: str) -> int:	1✔
137	uri = input_schematisation + "\|layername=schema_version"	×
138	schema_lyr = QgsVectorLayer(uri, "schema_version", "ogr")	×
139	if schema_lyr.isValid() and schema_lyr.featureCount() == 1:	×
140	# Take first (and only)
141	meta = next(schema_lyr.getFeatures())	×
142	return int(meta["version_num"])	×
143	else:
144	return None	×
145
146	def _get_rel_dem_path(self, input_schematisation: str, feedback) -> str:	1✔
147
148	schematisation_version = self._get_schematisation_version(input_schematisation)	×
149	setting_uri = None	×
150	if schematisation_version < 222:	×
151	setting_uri = input_schematisation + "\|layername=v2_global_settings"	×
152	else:
153	setting_uri = input_schematisation + "\|layername=model_settings"	×
154
155	feedback.pushInfo(f"Reading DEM file from: {setting_uri}")	×
156	settings_lyr = QgsVectorLayer(setting_uri, "settings", "ogr")	×
157	if settings_lyr.isValid() and settings_lyr.featureCount() == 1:	×
158	settings_feat = next(settings_lyr.getFeatures())	×
159	set_dem_rel_path = settings_feat["dem_file"]	×
160	else:
161	err = f"No (or multiple) global settings entries in {setting_uri}" "Check your schematisation file."	×
162	raise QgsProcessingException(f"Incorrect input schematisation file:\n{err}")	×
163
164	if schematisation_version >= 222:	×
NEW 165	set_dem_rel_path = os.path.join("rasters", str(set_dem_rel_path))	×
166
167	return set_dem_rel_path	×

nens / ThreeDiToolbox / #2497

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