#2616

Committed 22 Oct 2025 09:46AM UTC coverage: 36.554% (+1.8%) from 34.798%

Build # #2616

Build Type

push

coveralls-python

Committed by

web-flow

Commit Message

Concentration raster algorithms & general threedidepth algorithms refactor (#1147)

Functional improvements:
- Water depth/level raster algorithm has been split into two algorithms: single and multiple time steps (#958)
- Appropriately style and name water depth/level algorithm outputs.
- New processing algorithm: Concentration raster (single time step)
- New processing algorithm: Concentration raster (multiple time steps)
- New processing algorithm: Concentration raster (maximum)
- Bugfix: Processing algorithm "Maximum water depth/level" fails when writing to a temporary result (#945)
- Bugfix: Water depth tool raises KeyError: 's1_max' when using aggregate_results_3di.nc without the correct aggregation variables (#874)

Code changes:
- All `threedidepth`-based processing algorithm now derive from the same base class
- Utilility functions moved to utils files
- Moved util functions from water depth difference algo to utils files, because they are now also used by the concentration raster algo's. Also moved the tests for these util functions to the appropriate file
- Added smoke integration test for all six threedidepth-processing algo's
- Removed TimeSliderCheckbox widget (is no longer needed because single and multiple time steps algo's have been separated

Run Details

539 of 739 new or added lines in 7 files covered. (72.94%)

2 existing lines in 1 file now uncovered.

5282 of 14450 relevant lines covered (36.55%)

0.37 hits per line

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

26.83

/processing/widgets/widgets.py

import logging
from pathlib import Path
from typing import Dict

from processing.gui.wrappers import DIALOG_BATCH, DIALOG_STANDARD
from processing.gui.wrappers import WidgetWrapper
from qgis.gui import QgsGui
from qgis.gui import QgsProcessingGui
from qgis.PyQt import uic
from qgis.PyQt.QtWidgets import QComboBox
import h5py

from threedi_results_analysis.utils.user_messages import pop_up_info
from threedi_results_analysis.utils.netcdf import substances_from_netcdf

logger = logging.getLogger(__name__)
plugin_path = Path(__file__).resolve().parent.parent.parent


def format_timestep_value(value: float, drop_leading_zero: bool = False) -> str:
    days, seconds = divmod(int(value), 24 * 60 * 60)
    hours, seconds = divmod(seconds, 60 * 60)
    minutes, seconds = divmod(seconds, 60)

    if days == 0 and drop_leading_zero:
        formatted_display = "{:02d}:{:02d}".format(hours, minutes)
        return formatted_display

    formatted_display = "{:d} {:02d}:{:02d}".format(days, hours, minutes)
    return formatted_display


class ThreediResultTimeSliderWidgetWrapper(WidgetWrapper):
    def createWidget(self):
        if self.dialogType == DIALOG_STANDARD:
            self._widget = TimeSliderWidget()
        elif self.dialogType == DIALOG_BATCH:
            self._widget = TimeStepsCombobox()
        else:
            registry = QgsGui.instance().processingGuiRegistry()
            default_wrapper = registry.createParameterWidgetWrapper(
                self.parameterDefinition(),
                QgsProcessingGui.WidgetType.Modeler
            )
            self._widget = default_wrapper.createWidget()
        return self._widget

    def value(self):
        if self.dialogType in (DIALOG_STANDARD, DIALOG_BATCH):
            return self._widget.getValue()
        else:
            # widget is a QSpinBox
            return self._widget.value()

    def setValue(self, value):
        if value is not None:
            self._widget.setValue(int(value))

    def postInitialize(self, wrappers):
        # Connect the result-file parameter to the TimeSliderWidget/TimeStepsCombobox
        if self.dialogType in (DIALOG_STANDARD, DIALOG_BATCH):
            for wrapper in wrappers:
                if wrapper.parameterDefinition().name() == self.param.metadata().get("parentParameterName"):
                    wrapper.wrappedWidget().fileChanged.connect(self._widget.new_file_event)


WIDGET, BASE = uic.loadUiType(plugin_path / "processing" / "ui" / "widgetTimeSlider.ui")


class TimeSliderWidget(BASE, WIDGET):
    """
    Timeslider form widget. Provide a horizontal slider and an LCD connected to the slider.
    """
    def __init__(self):
        super(TimeSliderWidget, self).__init__(None)
        self.setupUi(self)
        self.horizontalSlider.valueChanged.connect(self.set_lcd_value)
        self.index = None
        self.timestamps = None
        self.reset()

    def getValue(self):
        return self.index

    def setValue(self, value):
        if value is not None:
            self.set_lcd_value(int(value))
            self.horizontalSlider.setValue(int(value))

    def set_timestamps(self, timestamps):
        self.setDisabled(False)
        self.horizontalSlider.setMinimum(0)
        self.horizontalSlider.setMaximum(len(timestamps) - 1)
        self.timestamps = timestamps
        self.set_lcd_value(0)  # also sets self.index

    def set_lcd_value(self, index: int):
        self.index = index
        if self.timestamps is not None:
            value = self.timestamps[index]
        else:
            value = 0
        lcd_value = format_timestep_value(value=value, drop_leading_zero=True)
        self.lcdNumber.display(lcd_value)

    def reset(self):
        self.setDisabled(True)
        self.index = None
        self.timestamps = None
        self.horizontalSlider.setMinimum(0)
        self.horizontalSlider.setMaximum(0)
        self.horizontalSlider.setValue(0)

    def new_file_event(self, file_path):
        """New file has been selected by the user. Try to read in the timestamps from the file."""
        if not file_path or not Path(file_path).is_file():
            self.reset()
            return

        try:
            with h5py.File(file_path, "r") as results:
                timestamps = results["time"][()]
                self.set_timestamps(timestamps)
        except Exception as e:
            logger.exception(e)
            pop_up_info(msg="Unable to read the file, see the logging for more information.")
            self.reset()


class TimeStepsCombobox(QComboBox):
    """Combobox with populated timestep data."""

    def getValue(self):
        return self.currentIndex()

    def setValue(self, value: int):
        self.setCurrentIndex(value)
        pass

    def populate_timestamps(self, timestamps):
        for i, value in enumerate(timestamps):
            human_readable_value = format_timestep_value(value)
            if human_readable_value.startswith("0"):
                human_readable_value = human_readable_value.split(" ", 1)[-1]
            self.addItem(human_readable_value)
        self.setCurrentIndex(0)

    def new_file_event(self, file_path):
        """New file has been selected by the user. Try to read in the timestamps from the file."""
        if not file_path or not Path(file_path).is_file():
            self.clear()
            return

        try:
            with h5py.File(file_path, "r") as results:
                timestamps = results["time"][()]
                self.populate_timestamps(timestamps)
        except Exception as e:
            logger.exception(e)
            pop_up_info(msg="Unable to read the file, see the logging for more information.")
            self.clear()


class SubstanceWidgetWrapper(WidgetWrapper):
    def createWidget(self):
        if self.dialogType in (DIALOG_STANDARD, DIALOG_BATCH):
            self._widget = SubstanceCombobox()
        else:
            registry = QgsGui.instance().processingGuiRegistry()
            default_wrapper = registry.createParameterWidgetWrapper(
                self.parameterDefinition(),
                QgsProcessingGui.WidgetType.Modeler
            )
            self._widget = default_wrapper.createWidget()
        return self._widget

    def value(self):
        if self.dialogType in (DIALOG_STANDARD, DIALOG_BATCH):
            return self._widget.getValue()
        else:
            # Widget is a QLineEdit
            return self._widget.text()

    def setValue(self, value):
        if value is not None:
            if self.dialogType in (DIALOG_STANDARD, DIALOG_BATCH):
                self._widget.setValue(str(value))
            else:
                self._widget.setText(str(value))

    def postInitialize(self, wrappers):
        # Connect the result-file parameter to the SubstanceCombobox
        if self.dialogType in (DIALOG_STANDARD, DIALOG_BATCH):
            for wrapper in wrappers:
                if wrapper.parameterDefinition().name() == self.param.metadata().get("parentParameterName"):
                    wrapper.wrappedWidget().fileChanged.connect(self._widget.new_file_event)


class SubstanceCombobox(QComboBox):
    """
    Combobox with populated substance data.
    Displayed texts are the substance names ("Chloride", "Phosphate", etc.).
    The user data behind it are the substance IDs ("substance1", "substance2", etc.)
    """
    def getValue(self):
        return self.currentData()

    def setValue(self, value: str):
        """Set combobox to the item whose substance ID (e.g. "substance1") matches the given value."""
        for i in range(self.count()):
            if self.itemData(i) == value:
                self.setCurrentIndex(i)
                return

    def populate(self, data: Dict[str, str]):
        """
        Populates the widget from a {substance id: substance name} Dict
        """
        self.clear()
        for substance_id, substance_name in data.items():
            self.addItem(substance_name, substance_id)
        if data:
            self.setCurrentIndex(0)

    def new_file_event(self, file_path):
        """New file has been selected by the user. Try to read in the substance data from the file."""
        if not file_path or not Path(file_path).is_file():
            self.clear()
            return

        try:
            substance_data = substances_from_netcdf(file_path)
            self.populate(substance_data)
        except Exception as e:
            logger.exception(e)
            pop_up_info(msg="Unable to read the file, see the logging for more information.")
            self.clear()

1	import logging	1✔
2	from pathlib import Path	1✔
3	from typing import Dict	1✔
4
5	from processing.gui.wrappers import DIALOG_BATCH, DIALOG_STANDARD	1✔
6	from processing.gui.wrappers import WidgetWrapper	1✔
7	from qgis.gui import QgsGui	1✔
8	from qgis.gui import QgsProcessingGui	1✔
9	from qgis.PyQt import uic	1✔
10	from qgis.PyQt.QtWidgets import QComboBox	1✔
11	import h5py	1✔
12
13	from threedi_results_analysis.utils.user_messages import pop_up_info	1✔
14	from threedi_results_analysis.utils.netcdf import substances_from_netcdf	1✔
15
16	logger = logging.getLogger(__name__)	1✔
17	plugin_path = Path(__file__).resolve().parent.parent.parent	1✔
18
19
20	def format_timestep_value(value: float, drop_leading_zero: bool = False) -> str:	1✔
NEW 21	days, seconds = divmod(int(value), 24 * 60 * 60)	×
NEW 22	hours, seconds = divmod(seconds, 60 * 60)	×
NEW 23	minutes, seconds = divmod(seconds, 60)	×
24
NEW 25	if days == 0 and drop_leading_zero:	×
NEW 26	formatted_display = "{:02d}:{:02d}".format(hours, minutes)	×
NEW 27	return formatted_display	×
28
NEW 29	formatted_display = "{:d} {:02d}:{:02d}".format(days, hours, minutes)	×
NEW 30	return formatted_display	×
31
32
33	class ThreediResultTimeSliderWidgetWrapper(WidgetWrapper):	1✔
34	def createWidget(self):	1✔
NEW 35	if self.dialogType == DIALOG_STANDARD:	×
NEW 36	self._widget = TimeSliderWidget()	×
NEW 37	elif self.dialogType == DIALOG_BATCH:	×
NEW 38	self._widget = TimeStepsCombobox()	×
39	else:
NEW 40	registry = QgsGui.instance().processingGuiRegistry()	×
NEW 41	default_wrapper = registry.createParameterWidgetWrapper(	×
42	self.parameterDefinition(),
43	QgsProcessingGui.WidgetType.Modeler
44	)
NEW 45	self._widget = default_wrapper.createWidget()	×
NEW 46	return self._widget	×
47
48	def value(self):	1✔
NEW 49	if self.dialogType in (DIALOG_STANDARD, DIALOG_BATCH):	×
NEW 50	return self._widget.getValue()	×
51	else:
52	# widget is a QSpinBox
NEW 53	return self._widget.value()	×
54
55	def setValue(self, value):	1✔
NEW 56	if value is not None:	×
NEW 57	self._widget.setValue(int(value))	×
58
59	def postInitialize(self, wrappers):	1✔
60	# Connect the result-file parameter to the TimeSliderWidget/TimeStepsCombobox
NEW 61	if self.dialogType in (DIALOG_STANDARD, DIALOG_BATCH):	×
NEW 62	for wrapper in wrappers:	×
NEW 63	if wrapper.parameterDefinition().name() == self.param.metadata().get("parentParameterName"):	×
NEW 64	wrapper.wrappedWidget().fileChanged.connect(self._widget.new_file_event)	×
65
66
67	WIDGET, BASE = uic.loadUiType(plugin_path / "processing" / "ui" / "widgetTimeSlider.ui")	1✔
68
69
70	class TimeSliderWidget(BASE, WIDGET):	1✔
71	"""
72	Timeslider form widget. Provide a horizontal slider and an LCD connected to the slider.
73	"""
74	def __init__(self):	1✔
NEW 75	super(TimeSliderWidget, self).__init__(None)	×
NEW 76	self.setupUi(self)	×
NEW 77	self.horizontalSlider.valueChanged.connect(self.set_lcd_value)	×
NEW 78	self.index = None	×
NEW 79	self.timestamps = None	×
NEW 80	self.reset()	×
81
82	def getValue(self):	1✔
NEW 83	return self.index	×
84
85	def setValue(self, value):	1✔
NEW 86	if value is not None:	×
NEW 87	self.set_lcd_value(int(value))	×
NEW 88	self.horizontalSlider.setValue(int(value))	×
89
90	def set_timestamps(self, timestamps):	1✔
NEW 91	self.setDisabled(False)	×
NEW 92	self.horizontalSlider.setMinimum(0)	×
NEW 93	self.horizontalSlider.setMaximum(len(timestamps) - 1)	×
NEW 94	self.timestamps = timestamps	×
NEW 95	self.set_lcd_value(0) # also sets self.index	×
96
97	def set_lcd_value(self, index: int):	1✔
NEW 98	self.index = index	×
NEW 99	if self.timestamps is not None:	×
NEW 100	value = self.timestamps[index]	×
101	else:
NEW 102	value = 0	×
NEW 103	lcd_value = format_timestep_value(value=value, drop_leading_zero=True)	×
NEW 104	self.lcdNumber.display(lcd_value)	×
105
106	def reset(self):	1✔
NEW 107	self.setDisabled(True)	×
NEW 108	self.index = None	×
NEW 109	self.timestamps = None	×
NEW 110	self.horizontalSlider.setMinimum(0)	×
NEW 111	self.horizontalSlider.setMaximum(0)	×
NEW 112	self.horizontalSlider.setValue(0)	×
113
114	def new_file_event(self, file_path):	1✔
115	"""New file has been selected by the user. Try to read in the timestamps from the file."""
NEW 116	if not file_path or not Path(file_path).is_file():	×
NEW 117	self.reset()	×
NEW 118	return	×
119
NEW 120	try:	×
NEW 121	with h5py.File(file_path, "r") as results:	×
NEW 122	timestamps = results["time"][()]	×
NEW 123	self.set_timestamps(timestamps)	×
NEW 124	except Exception as e:	×
NEW 125	logger.exception(e)	×
NEW 126	pop_up_info(msg="Unable to read the file, see the logging for more information.")	×
NEW 127	self.reset()	×
128
129
130	class TimeStepsCombobox(QComboBox):	1✔
131	"""Combobox with populated timestep data."""
132
133	def getValue(self):	1✔
NEW 134	return self.currentIndex()	×
135
136	def setValue(self, value: int):	1✔
NEW 137	self.setCurrentIndex(value)	×
NEW 138	pass	×
139
140	def populate_timestamps(self, timestamps):	1✔
NEW 141	for i, value in enumerate(timestamps):	×
NEW 142	human_readable_value = format_timestep_value(value)	×
NEW 143	if human_readable_value.startswith("0"):	×
NEW 144	human_readable_value = human_readable_value.split(" ", 1)[-1]	×
NEW 145	self.addItem(human_readable_value)	×
NEW 146	self.setCurrentIndex(0)	×
147
148	def new_file_event(self, file_path):	1✔
149	"""New file has been selected by the user. Try to read in the timestamps from the file."""
NEW 150	if not file_path or not Path(file_path).is_file():	×
NEW 151	self.clear()	×
NEW 152	return	×
153
NEW 154	try:	×
NEW 155	with h5py.File(file_path, "r") as results:	×
NEW 156	timestamps = results["time"][()]	×
NEW 157	self.populate_timestamps(timestamps)	×
NEW 158	except Exception as e:	×
NEW 159	logger.exception(e)	×
NEW 160	pop_up_info(msg="Unable to read the file, see the logging for more information.")	×
NEW 161	self.clear()	×
162
163
164	class SubstanceWidgetWrapper(WidgetWrapper):	1✔
165	def createWidget(self):	1✔
NEW 166	if self.dialogType in (DIALOG_STANDARD, DIALOG_BATCH):	×
NEW 167	self._widget = SubstanceCombobox()	×
168	else:
NEW 169	registry = QgsGui.instance().processingGuiRegistry()	×
NEW 170	default_wrapper = registry.createParameterWidgetWrapper(	×
171	self.parameterDefinition(),
172	QgsProcessingGui.WidgetType.Modeler
173	)
NEW 174	self._widget = default_wrapper.createWidget()	×
NEW 175	return self._widget	×
176
177	def value(self):	1✔
NEW 178	if self.dialogType in (DIALOG_STANDARD, DIALOG_BATCH):	×
NEW 179	return self._widget.getValue()	×
180	else:
181	# Widget is a QLineEdit
NEW 182	return self._widget.text()	×
183
184	def setValue(self, value):	1✔
NEW 185	if value is not None:	×
NEW 186	if self.dialogType in (DIALOG_STANDARD, DIALOG_BATCH):	×
NEW 187	self._widget.setValue(str(value))	×
188	else:
NEW 189	self._widget.setText(str(value))	×
190
191	def postInitialize(self, wrappers):	1✔
192	# Connect the result-file parameter to the SubstanceCombobox
NEW 193	if self.dialogType in (DIALOG_STANDARD, DIALOG_BATCH):	×
NEW 194	for wrapper in wrappers:	×
NEW 195	if wrapper.parameterDefinition().name() == self.param.metadata().get("parentParameterName"):	×
NEW 196	wrapper.wrappedWidget().fileChanged.connect(self._widget.new_file_event)	×
197
198
199	class SubstanceCombobox(QComboBox):	1✔
200	"""
201	Combobox with populated substance data.
202	Displayed texts are the substance names ("Chloride", "Phosphate", etc.).
203	The user data behind it are the substance IDs ("substance1", "substance2", etc.)
204	"""
205	def getValue(self):	1✔
NEW 206	return self.currentData()	×
207
208	def setValue(self, value: str):	1✔
209	"""Set combobox to the item whose substance ID (e.g. "substance1") matches the given value."""
NEW 210	for i in range(self.count()):	×
NEW 211	if self.itemData(i) == value:	×
NEW 212	self.setCurrentIndex(i)	×
NEW 213	return	×
214
215	def populate(self, data: Dict[str, str]):	1✔
216	"""
217	Populates the widget from a {substance id: substance name} Dict
218	"""
NEW 219	self.clear()	×
NEW 220	for substance_id, substance_name in data.items():	×
NEW 221	self.addItem(substance_name, substance_id)	×
NEW 222	if data:	×
NEW 223	self.setCurrentIndex(0)	×
224
225	def new_file_event(self, file_path):	1✔
226	"""New file has been selected by the user. Try to read in the substance data from the file."""
NEW 227	if not file_path or not Path(file_path).is_file():	×
NEW 228	self.clear()	×
NEW 229	return	×
230
NEW 231	try:	×
NEW 232	substance_data = substances_from_netcdf(file_path)	×
NEW 233	self.populate(substance_data)	×
NEW 234	except Exception as e:	×
NEW 235	logger.exception(e)	×
NEW 236	pop_up_info(msg="Unable to read the file, see the logging for more information.")	×
NEW 237	self.clear()	×

nens / ThreeDiToolbox / #2616

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