#2610

Committed 21 Oct 2025 08:45AM UTC coverage: 34.548% (-0.3%) from 34.798%

Build # #2610

Build Type

push

coveralls-python

Committed by

web-flow

Commit Message

Merge 55e1ecf69 into 7ff7d80d5

Run Details

255 of 753 new or added lines in 7 files covered. (33.86%)

2 existing lines in 1 file now uncovered.

4997 of 14464 relevant lines covered (34.55%)

0.35 hits per line

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

27.53

/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:
            if not self.param.metadata().get("optional"):
                self._widget = TimeSliderWidget()
            else:
                self._widget = CheckboxTimeSliderWidget()
        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()


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


class CheckboxTimeSliderWidget(TimeSliderWidget, WIDGET, BASE):
    """Time slider widget with a checkbox to enable/disable the time slider"""

    def __init__(self):
        super().__init__()
        self.horizontalSlider.setDisabled(not self.checkBox.isChecked())
        self.checkBox.stateChanged.connect(self.new_check_box_event)

    def getValue(self):
        if self.checkBox.isChecked():
            return self.index
        else:
            return None

    def new_check_box_event(self, state):
        self.horizontalSlider.setDisabled(not self.checkBox.isChecked())


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	if not self.param.metadata().get("optional"):	×
NEW 37	self._widget = TimeSliderWidget()	×
38	else:
NEW 39	self._widget = CheckboxTimeSliderWidget()	×
NEW 40	elif self.dialogType == DIALOG_BATCH:	×
NEW 41	self._widget = TimeStepsCombobox()	×
42	else:
NEW 43	registry = QgsGui.instance().processingGuiRegistry()	×
NEW 44	default_wrapper = registry.createParameterWidgetWrapper(	×
45	self.parameterDefinition(),
46	QgsProcessingGui.WidgetType.Modeler
47	)
NEW 48	self._widget = default_wrapper.createWidget()	×
NEW 49	return self._widget	×
50
51	def value(self):	1✔
NEW 52	if self.dialogType in (DIALOG_STANDARD, DIALOG_BATCH):	×
NEW 53	return self._widget.getValue()	×
54	else:
55	# widget is a QSpinBox
NEW 56	return self._widget.value()	×
57
58	def setValue(self, value):	1✔
NEW 59	if value is not None:	×
NEW 60	self._widget.setValue(int(value))	×
61
62	def postInitialize(self, wrappers):	1✔
63	# Connect the result-file parameter to the TimeSliderWidget/TimeStepsCombobox
NEW 64	if self.dialogType in (DIALOG_STANDARD, DIALOG_BATCH):	×
NEW 65	for wrapper in wrappers:	×
NEW 66	if wrapper.parameterDefinition().name() == self.param.metadata().get("parentParameterName"):	×
NEW 67	wrapper.wrappedWidget().fileChanged.connect(self._widget.new_file_event)	×
68
69
70	WIDGET, BASE = uic.loadUiType(plugin_path / "processing" / "ui" / "widgetTimeSlider.ui")	1✔
71
72
73	class TimeSliderWidget(BASE, WIDGET):	1✔
74	"""
75	Timeslider form widget. Provide a horizontal slider and an LCD connected to the slider.
76	"""
77	def __init__(self):	1✔
NEW 78	super(TimeSliderWidget, self).__init__(None)	×
NEW 79	self.setupUi(self)	×
NEW 80	self.horizontalSlider.valueChanged.connect(self.set_lcd_value)	×
NEW 81	self.index = None	×
NEW 82	self.timestamps = None	×
NEW 83	self.reset()	×
84
85	def getValue(self):	1✔
NEW 86	return self.index	×
87
88	def setValue(self, value):	1✔
NEW 89	if value is not None:	×
NEW 90	self.set_lcd_value(int(value))	×
NEW 91	self.horizontalSlider.setValue(int(value))	×
92
93	def set_timestamps(self, timestamps):	1✔
NEW 94	self.setDisabled(False)	×
NEW 95	self.horizontalSlider.setMinimum(0)	×
NEW 96	self.horizontalSlider.setMaximum(len(timestamps) - 1)	×
NEW 97	self.timestamps = timestamps	×
NEW 98	self.set_lcd_value(0) # also sets self.index	×
99
100	def set_lcd_value(self, index: int):	1✔
NEW 101	self.index = index	×
NEW 102	if self.timestamps is not None:	×
NEW 103	value = self.timestamps[index]	×
104	else:
NEW 105	value = 0	×
NEW 106	lcd_value = format_timestep_value(value=value, drop_leading_zero=True)	×
NEW 107	self.lcdNumber.display(lcd_value)	×
108
109	def reset(self):	1✔
NEW 110	self.setDisabled(True)	×
NEW 111	self.index = None	×
NEW 112	self.timestamps = None	×
NEW 113	self.horizontalSlider.setMinimum(0)	×
NEW 114	self.horizontalSlider.setMaximum(0)	×
NEW 115	self.horizontalSlider.setValue(0)	×
116
117	def new_file_event(self, file_path):	1✔
118	"""New file has been selected by the user. Try to read in the timestamps from the file."""
NEW 119	if not file_path or not Path(file_path).is_file():	×
NEW 120	self.reset()	×
NEW 121	return	×
122
NEW 123	try:	×
NEW 124	with h5py.File(file_path, "r") as results:	×
NEW 125	timestamps = results["time"][()]	×
NEW 126	self.set_timestamps(timestamps)	×
NEW 127	except Exception as e:	×
NEW 128	logger.exception(e)	×
NEW 129	pop_up_info(msg="Unable to read the file, see the logging for more information.")	×
NEW 130	self.reset()	×
131
132
133	WIDGET, BASE = uic.loadUiType(plugin_path / "processing" / "ui" / "widgetTimeSliderCheckbox.ui")	1✔
134
135
136	class CheckboxTimeSliderWidget(TimeSliderWidget, WIDGET, BASE):	1✔
137	"""Time slider widget with a checkbox to enable/disable the time slider"""
138
139	def __init__(self):	1✔
NEW 140	super().__init__()	×
NEW 141	self.horizontalSlider.setDisabled(not self.checkBox.isChecked())	×
NEW 142	self.checkBox.stateChanged.connect(self.new_check_box_event)	×
143
144	def getValue(self):	1✔
NEW 145	if self.checkBox.isChecked():	×
NEW 146	return self.index	×
147	else:
NEW 148	return None	×
149
150	def new_check_box_event(self, state):	1✔
NEW 151	self.horizontalSlider.setDisabled(not self.checkBox.isChecked())	×
152
153
154	class TimeStepsCombobox(QComboBox):	1✔
155	"""Combobox with populated timestep data."""
156
157	def getValue(self):	1✔
NEW 158	return self.currentIndex()	×
159
160	def setValue(self, value: int):	1✔
NEW 161	self.setCurrentIndex(value)	×
NEW 162	pass	×
163
164	def populate_timestamps(self, timestamps):	1✔
NEW 165	for i, value in enumerate(timestamps):	×
NEW 166	human_readable_value = format_timestep_value(value)	×
NEW 167	if human_readable_value.startswith("0"):	×
NEW 168	human_readable_value = human_readable_value.split(" ", 1)[-1]	×
NEW 169	self.addItem(human_readable_value)	×
NEW 170	self.setCurrentIndex(0)	×
171
172	def new_file_event(self, file_path):	1✔
173	"""New file has been selected by the user. Try to read in the timestamps from the file."""
NEW 174	if not file_path or not Path(file_path).is_file():	×
NEW 175	self.clear()	×
NEW 176	return	×
177
NEW 178	try:	×
NEW 179	with h5py.File(file_path, "r") as results:	×
NEW 180	timestamps = results["time"][()]	×
NEW 181	self.populate_timestamps(timestamps)	×
NEW 182	except Exception as e:	×
NEW 183	logger.exception(e)	×
NEW 184	pop_up_info(msg="Unable to read the file, see the logging for more information.")	×
NEW 185	self.clear()	×
186
187
188	class SubstanceWidgetWrapper(WidgetWrapper):	1✔
189	def createWidget(self):	1✔
NEW 190	if self.dialogType in (DIALOG_STANDARD, DIALOG_BATCH):	×
NEW 191	self._widget = SubstanceCombobox()	×
192	else:
NEW 193	registry = QgsGui.instance().processingGuiRegistry()	×
NEW 194	default_wrapper = registry.createParameterWidgetWrapper(	×
195	self.parameterDefinition(),
196	QgsProcessingGui.WidgetType.Modeler
197	)
NEW 198	self._widget = default_wrapper.createWidget()	×
NEW 199	return self._widget	×
200
201	def value(self):	1✔
NEW 202	if self.dialogType in (DIALOG_STANDARD, DIALOG_BATCH):	×
NEW 203	return self._widget.getValue()	×
204	else:
205	# Widget is a QLineEdit
NEW 206	return self._widget.text()	×
207
208	def setValue(self, value):	1✔
NEW 209	if value is not None:	×
NEW 210	if self.dialogType in (DIALOG_STANDARD, DIALOG_BATCH):	×
NEW 211	self._widget.setValue(str(value))	×
212	else:
NEW 213	self._widget.setText(str(value))	×
214
215	def postInitialize(self, wrappers):	1✔
216	# Connect the result-file parameter to the SubstanceCombobox
NEW 217	if self.dialogType in (DIALOG_STANDARD, DIALOG_BATCH):	×
NEW 218	for wrapper in wrappers:	×
NEW 219	if wrapper.parameterDefinition().name() == self.param.metadata().get("parentParameterName"):	×
NEW 220	wrapper.wrappedWidget().fileChanged.connect(self._widget.new_file_event)	×
221
222
223	class SubstanceCombobox(QComboBox):	1✔
224	"""
225	Combobox with populated substance data.
226	Displayed texts are the substance names ("Chloride", "Phosphate", etc.).
227	The user data behind it are the substance IDs ("substance1", "substance2", etc.)
228	"""
229	def getValue(self):	1✔
NEW 230	return self.currentData()	×
231
232	def setValue(self, value: str):	1✔
233	"""Set combobox to the item whose substance ID (e.g. "substance1") matches the given value."""
NEW 234	for i in range(self.count()):	×
NEW 235	if self.itemData(i) == value:	×
NEW 236	self.setCurrentIndex(i)	×
NEW 237	return	×
238
239	def populate(self, data: Dict[str, str]):	1✔
240	"""
241	Populates the widget from a {substance id: substance name} Dict
242	"""
NEW 243	self.clear()	×
NEW 244	for substance_id, substance_name in data.items():	×
NEW 245	self.addItem(substance_name, substance_id)	×
NEW 246	if data:	×
NEW 247	self.setCurrentIndex(0)	×
248
249	def new_file_event(self, file_path):	1✔
250	"""New file has been selected by the user. Try to read in the substance data from the file."""
NEW 251	if not file_path or not Path(file_path).is_file():	×
NEW 252	self.clear()	×
NEW 253	return	×
254
NEW 255	try:	×
NEW 256	substance_data = substances_from_netcdf(file_path)	×
NEW 257	self.populate(substance_data)	×
NEW 258	except Exception as e:	×
NEW 259	logger.exception(e)	×
NEW 260	pop_up_info(msg="Unable to read the file, see the logging for more information.")	×
NEW 261	self.clear()	×

nens / ThreeDiToolbox / #2610

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