#2589

Committed 19 Sep 2025 08:50AM UTC coverage: 35.01% (-0.1%) from 35.146%

Build # #2589

Build Type

push

coveralls-python

Committed by

web-flow

Commit Message

Merge 38792c162 into f6f4be1e7

Run Details

62 of 260 new or added lines in 40 files covered. (23.85%)

6 existing lines in 5 files now uncovered.

4859 of 13879 relevant lines covered (35.01%)

0.35 hits per line

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

42.86

/tool_graph/graph_model.py

from qgis.PyQt.QtCore import Qt
from qgis.PyQt.QtGui import QColor
from random import randint
from threedi_results_analysis.models.base import BaseModel
from threedi_results_analysis.models.base_fields import CHECKBOX_FIELD
from threedi_results_analysis.models.base_fields import CheckboxField
from threedi_results_analysis.models.base_fields import ValueField
from threedi_results_analysis.utils.color import COLOR_LIST
from threedigrid.admin.gridresultadmin import GridH5StructureControl

import logging
import numpy as np
import pyqtgraph as pg


logger = logging.getLogger(__name__)

EMPTY_TIMESERIES = np.array([], dtype=float)


class LocationTimeseriesModel(BaseModel):
    """Model implementation for (selected objects) for display in graph"""

    feature_color_map = {}
    colors = COLOR_LIST.copy()

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

    def get_color(self, idx: int, layer_id: str) -> QColor:
        key = (idx, layer_id)
        if not self.feature_color_map:
            self.feature_color_map[key] = 0
        elif key not in self.feature_color_map:
            current_color = max(self.feature_color_map.values())
            # if the list of colors is exhausted append a new random one
            if current_color + 1 == len(self.colors):
                new_random_color = (randint(0, 256), randint(0, 256), randint(0, 256))
                self.colors.append(new_random_color)
            # choose the next color in the list
            self.feature_color_map[key] = current_color + 1
            return self.colors[self.feature_color_map[key]]

        return self.colors[self.feature_color_map[key]]

    def flags(self, index):

        flags = Qt.ItemFlag.ItemIsEnabled | Qt.ItemFlag.ItemIsSelectable
        if self.columns[index.column()].field_type == CHECKBOX_FIELD:
            flags |= Qt.ItemFlag.ItemIsUserCheckable | Qt.ItemFlag.ItemIsEditable
        elif index.column() == 2:  # user-defined label
            flags |= Qt.ItemFlag.ItemIsEditable

        return flags

    def data(self, index, role=Qt.ItemDataRole.DisplayRole):
        """Qt function to get data from items for the visible columns"""

        if not index.isValid():
            return None

        if role == Qt.ItemDataRole.DisplayRole:
            if index.column() == 1:  # color
                return ""
            elif index.column() == 3:  # grid: take name from result parent
                return self.rows[index.row()][index.column()+1].value.parent().text()
            elif index.column() == 4:  # result
                return self.rows[index.row()][index.column()].value.text()

        return super().data(index, role)

    class Fields(object):
        """Fields and functions of ModelItem"""

        active = CheckboxField(
            show=True, default_value=True, column_width=20, column_name="active"
        )

        color = ValueField(
            show=True,
            column_width=70,
            column_name="pattern"
        )

        object_label = ValueField(show=True, column_width=100, column_name="label")  # user-defined label per feature

        grid_name = ValueField(show=True, column_width=100, column_name="grid", default_value="grid")
        result = ValueField(show=True, column_width=100, column_name="result")
        object_id = ValueField(show=True, column_width=50, column_name="id")
        object_name = ValueField(show=True, column_width=50, column_name="type")  # e.g. 2D-1D
        object_type = ValueField(show=False)  # e.g. flowline
        hover = ValueField(show=False, default_value=False)

        _plots = {}

        def plots(self, parameters, absolute, time_units):
            """
            Get pyqtgraph plot of selected object and timeseries.

            Performs some caching on key: "result_uuid, feature_id, layer_name (pump, flowlines), time-unit, absolute"
            :param parameters: string, parameter identification
            :param result_ds_nr: nr of result ts_datasources in model
            :return: pyqtgraph PlotDataItem
            """
            # Key is result uuid, feature id, layer name (pump, flowlines), time-unit, absolute
            result_key = (self.result.value.id, str(self.object_id.value), self.object_type.value, time_units, absolute)
            if not str(parameters) in self._plots:
                self._plots[str(parameters)] = {}
            if result_key not in self._plots[str(parameters)]:
                ts_table = self.timeseries_table(
                    parameters=parameters, absolute=absolute, time_units=time_units,
                )

                pen = pg.mkPen(color=self.color.value, width=2, style=self.result.value._pattern)

                logger.info(f"Creating plot item for {result_key}: {parameters}")
                self._plots[str(parameters)][result_key] = pg.PlotDataItem(ts_table, pen=pen)

            # logger.info(f"Retrieving plot for {result_key}: {parameters}")
            return self._plots[str(parameters)][result_key]

        def timeseries_table(self, parameters, absolute, time_units):
            """
            get list of timestamp values for object and parameters
            from result ts_datasources
            :param parameters:
            :param result_ds_nr:
            :return: numpy array with timestamp, values
            """
            threedi_result = self.result.value.threedi_result

            if (parameters not in threedi_result.available_subgrid_map_vars and
                    parameters not in threedi_result.available_aggregation_vars and
                    parameters not in [v["parameters"] for v in threedi_result.available_water_quality_vars] and
                    parameters not in [v["parameters"] for v in threedi_result.available_structure_control_actions_vars]):
                logger.warning(f"Parameter {parameters} not available in result {self.result.value.text()}")
                return EMPTY_TIMESERIES

            ga = threedi_result.get_gridadmin(parameters)
            if ga.has_pumpstations:
                # In some gridadmin types pumps do not have a Meta attribute... In
                # such cases (e.g. water quality) the attribute does not have a meaning and
                # the timeserie should be empty.
                try:
                    pump_fields = set(list(ga.pumps.Meta.composite_fields.keys()))
                except AttributeError:
                    pump_fields = {}
            else:
                pump_fields = {}
            if self.object_type.value == "pump_linestring" and parameters not in pump_fields and not isinstance(ga, GridH5StructureControl):
                return EMPTY_TIMESERIES
            if self.object_type.value == "flowline" and parameters in pump_fields:
                return EMPTY_TIMESERIES

            timeseries = threedi_result.get_timeseries(
                parameters, node_id=self.object_id.value, fill_value=np.NaN, selected_object_type=self.object_type.value
            )
            if timeseries.shape[1] == 1:
                logger.info("1-element timeserie, plotting empty serie")
                return EMPTY_TIMESERIES
            if absolute:
                timeseries = np.abs(timeseries)
            if time_units == "hrs":
                vector = np.array([3600, 1])
            elif time_units == "mins":
                vector = np.array([60, 1])
            else:
                vector = np.array([1, 1])
            return timeseries / vector

1	from qgis.PyQt.QtCore import Qt	1✔
2	from qgis.PyQt.QtGui import QColor	1✔
3	from random import randint	1✔
4	from threedi_results_analysis.models.base import BaseModel	1✔
5	from threedi_results_analysis.models.base_fields import CHECKBOX_FIELD	1✔
6	from threedi_results_analysis.models.base_fields import CheckboxField	1✔
7	from threedi_results_analysis.models.base_fields import ValueField	1✔
8	from threedi_results_analysis.utils.color import COLOR_LIST	1✔
9	from threedigrid.admin.gridresultadmin import GridH5StructureControl	1✔
10
11	import logging	1✔
12	import numpy as np	1✔
13	import pyqtgraph as pg	1✔
14
15
16	logger = logging.getLogger(__name__)	1✔
17
18	EMPTY_TIMESERIES = np.array([], dtype=float)	1✔
19
20
21	class LocationTimeseriesModel(BaseModel):	1✔
22	"""Model implementation for (selected objects) for display in graph"""
23
24	feature_color_map = {}	1✔
25	colors = COLOR_LIST.copy()	1✔
26
27	def __init__(self, args, *kwargs):	1✔
28	super().__init__(args, *kwargs)	1✔
29
30	def get_color(self, idx: int, layer_id: str) -> QColor:	1✔
31	key = (idx, layer_id)	×
32	if not self.feature_color_map:	×
33	self.feature_color_map[key] = 0	×
34	elif key not in self.feature_color_map:	×
35	current_color = max(self.feature_color_map.values())	×
36	# if the list of colors is exhausted append a new random one
37	if current_color + 1 == len(self.colors):	×
38	new_random_color = (randint(0, 256), randint(0, 256), randint(0, 256))	×
39	self.colors.append(new_random_color)	×
40	# choose the next color in the list
41	self.feature_color_map[key] = current_color + 1	×
42	return self.colors[self.feature_color_map[key]]	×
43
44	return self.colors[self.feature_color_map[key]]	×
45
46	def flags(self, index):	1✔
47
NEW 48	flags = Qt.ItemFlag.ItemIsEnabled \| Qt.ItemFlag.ItemIsSelectable	×
49	if self.columns[index.column()].field_type == CHECKBOX_FIELD:	×
NEW 50	flags \|= Qt.ItemFlag.ItemIsUserCheckable \| Qt.ItemFlag.ItemIsEditable	×
51	elif index.column() == 2: # user-defined label	×
NEW 52	flags \|= Qt.ItemFlag.ItemIsEditable	×
53
54	return flags	×
55
56	def data(self, index, role=Qt.ItemDataRole.DisplayRole):	1✔
57	"""Qt function to get data from items for the visible columns"""
58
59	if not index.isValid():	1✔
60	return None	×
61
62	if role == Qt.ItemDataRole.DisplayRole:	1✔
63	if index.column() == 1: # color	1✔
64	return ""	1✔
65	elif index.column() == 3: # grid: take name from result parent	1✔
66	return self.rows[index.row()][index.column()+1].value.parent().text()	×
67	elif index.column() == 4: # result	1✔
68	return self.rows[index.row()][index.column()].value.text()	×
69
70	return super().data(index, role)	1✔
71
72	class Fields(object):	1✔
73	"""Fields and functions of ModelItem"""
74
75	active = CheckboxField(	1✔
76	show=True, default_value=True, column_width=20, column_name="active"
77	)
78
79	color = ValueField(	1✔
80	show=True,
81	column_width=70,
82	column_name="pattern"
83	)
84
85	object_label = ValueField(show=True, column_width=100, column_name="label") # user-defined label per feature	1✔
86
87	grid_name = ValueField(show=True, column_width=100, column_name="grid", default_value="grid")	1✔
88	result = ValueField(show=True, column_width=100, column_name="result")	1✔
89	object_id = ValueField(show=True, column_width=50, column_name="id")	1✔
90	object_name = ValueField(show=True, column_width=50, column_name="type") # e.g. 2D-1D	1✔
91	object_type = ValueField(show=False) # e.g. flowline	1✔
92	hover = ValueField(show=False, default_value=False)	1✔
93
94	_plots = {}	1✔
95
96	def plots(self, parameters, absolute, time_units):	1✔
97	"""
98	Get pyqtgraph plot of selected object and timeseries.
99
100	Performs some caching on key: "result_uuid, feature_id, layer_name (pump, flowlines), time-unit, absolute"
101	:param parameters: string, parameter identification
102	:param result_ds_nr: nr of result ts_datasources in model
103	:return: pyqtgraph PlotDataItem
104	"""
105	# Key is result uuid, feature id, layer name (pump, flowlines), time-unit, absolute
106	result_key = (self.result.value.id, str(self.object_id.value), self.object_type.value, time_units, absolute)	×
107	if not str(parameters) in self._plots:	×
108	self._plots[str(parameters)] = {}	×
109	if result_key not in self._plots[str(parameters)]:	×
110	ts_table = self.timeseries_table(	×
111	parameters=parameters, absolute=absolute, time_units=time_units,
112	)
113
114	pen = pg.mkPen(color=self.color.value, width=2, style=self.result.value._pattern)	×
115
116	logger.info(f"Creating plot item for {result_key}: {parameters}")	×
117	self._plots[str(parameters)][result_key] = pg.PlotDataItem(ts_table, pen=pen)	×
118
119	# logger.info(f"Retrieving plot for {result_key}: {parameters}")
120	return self._plots[str(parameters)][result_key]	×
121
122	def timeseries_table(self, parameters, absolute, time_units):	1✔
123	"""
124	get list of timestamp values for object and parameters
125	from result ts_datasources
126	:param parameters:
127	:param result_ds_nr:
128	:return: numpy array with timestamp, values
129	"""
130	threedi_result = self.result.value.threedi_result	×
131
132	if (parameters not in threedi_result.available_subgrid_map_vars and	×
133	parameters not in threedi_result.available_aggregation_vars and
134	parameters not in [v["parameters"] for v in threedi_result.available_water_quality_vars] and
135	parameters not in [v["parameters"] for v in threedi_result.available_structure_control_actions_vars]):
136	logger.warning(f"Parameter {parameters} not available in result {self.result.value.text()}")	×
137	return EMPTY_TIMESERIES	×
138
139	ga = threedi_result.get_gridadmin(parameters)	×
140	if ga.has_pumpstations:	×
141	# In some gridadmin types pumps do not have a Meta attribute... In
142	# such cases (e.g. water quality) the attribute does not have a meaning and
143	# the timeserie should be empty.
144	try:	×
145	pump_fields = set(list(ga.pumps.Meta.composite_fields.keys()))	×
146	except AttributeError:	×
147	pump_fields = {}	×
148	else:
149	pump_fields = {}	×
150	if self.object_type.value == "pump_linestring" and parameters not in pump_fields and not isinstance(ga, GridH5StructureControl):	×
151	return EMPTY_TIMESERIES	×
152	if self.object_type.value == "flowline" and parameters in pump_fields:	×
153	return EMPTY_TIMESERIES	×
154
155	timeseries = threedi_result.get_timeseries(	×
156	parameters, node_id=self.object_id.value, fill_value=np.NaN, selected_object_type=self.object_type.value
157	)
158	if timeseries.shape[1] == 1:	×
159	logger.info("1-element timeserie, plotting empty serie")	×
160	return EMPTY_TIMESERIES	×
161	if absolute:	×
162	timeseries = np.abs(timeseries)	×
163	if time_units == "hrs":	×
164	vector = np.array([3600, 1])	×
165	elif time_units == "mins":	×
166	vector = np.array([60, 1])	×
167	else:
168	vector = np.array([1, 1])	×
169	return timeseries / vector	×

nens / ThreeDiToolbox / #2589

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