#2640

Committed 19 Dec 2025 03:01PM UTC coverage: 36.526% (-0.02%) from 36.546%

Build # #2640

Build Type

push

coveralls-python

Committed by

web-flow

Commit Message

Rebranding the 3Di Results Analysis to Rana Results Analysis (#1151)

Run Details

13 of 77 new or added lines in 22 files covered. (16.88%)

3 existing lines in 3 files now uncovered.

5262 of 14406 relevant lines covered (36.53%)

0.37 hits per line

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

26.79

/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

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

nens / ThreeDiToolbox / #2640

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