1"""
2Module provides a class for visualizing scenarios.
3"""
4from typing import Optional, Union, List, Tuple
5
6import numpy as np
7
8import matplotlib.pyplot as plt
9from matplotlib.animation import FuncAnimation
10import matplotlib as mpl
11import networkx.drawing.nx_pylab as nxp
12from svgpath2mpl import parse_path
13
14from ..simulation.scenario_simulator import ScenarioSimulator
15from ..simulation.scada.scada_data import ScadaData
16from ..visualization import JunctionObject, EdgeObject, ColorScheme, \
17 epyt_flow_colors, my_draw_networkx_nodes
18
19PUMP_PATH = ('M 202.5 93 A 41.5 42 0 0 0 161 135 A 41.5 42 0 0 0 202.5 177 A '
20 '41.5 42 0 0 0 244 135 A 41.5 42 0 0 0 241.94922 122 L 278 122 '
21 'L 278 93 L 203 93 L 203 93.011719 A 41.5 42 0 0 0 202.5 93 z')
22RESERVOIR_PATH = ('M 325 41 A 43 24.5 0 0 0 282.05664 65 L 282 65 L 282 65.5 '
23 'L 282 163 L 282 168 L 282 216 L 305 216 L 305 168 L 345 '
24 '168 L 345 216 L 368 216 L 368 168 L 368 163 L 368 65.5 L '
25 '368 65 L 367.98047 65 A 43 24.5 0 0 0 325 41 z')
26TANK_PATH = ('M 325 41 A 43 24.5 0 0 0 282.05664 65 L 282 65 L 282 65.5 L 282 '
27 '185 L 368 185 L 368 65.5 L 368 65 L 367.98047 65 A 43 24.5 0 0'
28 ' 0 325 41 z')
29VALVE_PATH = ('M 9.9999064 9.9999064 L 9.9999064 110 L 69.999862 59.999955 L '
30 '9.9999064 9.9999064 z M 69.999862 59.999955 L 129.99982 110 L '
31 '129.99982 9.9999064 L 69.999862 59.999955 z')
32
33
[docs]
34class Marker:
35 """
36 The Marker class provides svg representations of hydraulic components
37 (pump, reservoir, tank and valve), which are loaded from their respective
38 svg paths and transformed into
39 `~matplotlib.path.Path <https://matplotlib.org/stable/api/path_api.html#matplotlib.path.Path>`_
40 objects in order to be used with the matplotlib library.
41
42 Attributes
43 ----------
44 pump : `matplotlib.path.Path <https://matplotlib.org/stable/api/path_api.html#matplotlib.path.Path>`_
45 Marker for the pump, loaded from PUMP_PATH.
46 reservoir : `matplotlib.path.Path <https://matplotlib.org/stable/api/path_api.html#matplotlib.path.Path>`_
47 Marker for the reservoir, loaded from RESERVOIR_PATH.
48 tank : `matplotlib.path.Path <https://matplotlib.org/stable/api/path_api.html#matplotlib.path.Path>`_
49 Marker for the tank, loaded from TANK_PATH.
50 valve : `matplotlib.path.Path <https://matplotlib.org/stable/api/path_api.html#matplotlib.path.Path>`_
51 Marker for the valve, loaded from VALVE_PATH.
52
53 Methods
54 -------
55 __marker_from_path(path, scale_p=1)
56 Loads and applies transformations to the marker shape from the given
57 path.
58 """
59
60 def __init__(self):
61 """
62 Initializes the Marker class and assigns
63 `matplotlib.path.Path <https://matplotlib.org/stable/api/path_api.html#matplotlib.path.Path>`_
64 markers for pump, reservoir, tank, and valve components.
65 """
66 self.pump = self.__marker_from_path(PUMP_PATH, 2)
67 self.reservoir = self.__marker_from_path(RESERVOIR_PATH)
68 self.tank = self.__marker_from_path(TANK_PATH)
69 self.valve = self.__marker_from_path(VALVE_PATH)
70
71 @staticmethod
72 def __marker_from_path(path: str, scale_p: int = 1) -> mpl.path.Path:
73 """
74 Loads the marker from the specified path and adjusts it representation
75 by aligning, rotating and scaling it.
76
77 Parameters
78 ----------
79 path : `str`
80 The svg path describing the marker shape.
81 scale_p : `float`, optional
82 Scaling factor for the marker (default is 1).
83
84 Returns
85 -------
86 `matplotlib.path.Path <https://matplotlib.org/stable/api/path_api.html#matplotlib.path.Path>`_
87 The transformed marker object after loading and adjusting it.
88 """
89 marker_tmp = parse_path(path)
90 marker_tmp.vertices -= marker_tmp.vertices.mean(axis=0)
91 marker_tmp = marker_tmp.transformed(
92 mpl.transforms.Affine2D().rotate_deg(180))
93 marker_tmp = marker_tmp.transformed(
94 mpl.transforms.Affine2D().scale(-scale_p, scale_p))
95 return marker_tmp
96
97
[docs]
98class ScenarioVisualizer:
99 """
100 This class provides the necessary function to generate visualizations in
101 the form of plots or animations from water network data.
102
103 Given a :class:`~epyt_flow.simulation.scenario_simulator.ScenarioSimulator`
104 object, this class provides the necessary functions to plot the network
105 topology and to color hydraulic elements according to simulation data. The
106 resulting plot can then be displayed or saved.
107
108 Attributes
109 ----------
110 __scenario : :class:`~epyt_flow.simulation.scenario_simulator.ScenarioSimulator`
111 ScenarioSimulator object containing the network topology and
112 configurations to obtain the simulation data which should be displayed.
113 fig : `matplotlib.pyplot.Figure <https://matplotlib.org/stable/api/_as_gen/matplotlib.figure.Figure.html#matplotlib.figure.Figure>`_ or None
114 Figure object used for plotting, created and customized by calling the
115 methods of this class, initialized as None.
116 ax : `~matplotlib.axes.Axes <https://matplotlib.org/stable/api/_as_gen/matplotlib.axes.Axes.html#matplotlib-axes-axes>`_ or None
117 The axes for plotting, initialized as None.
118 scada_data : :class:`~epyt_flow.simulation.scada.scada_data.ScadaData` or None
119 SCADA data created by the ScenarioSimulator object, initialized as
120 None.
121 topology : :class:`~epyt_flow.topology.NetworkTopology`
122 Topology object retrieved from the scenario, containing the structure
123 of the water distribution network.
124 color_scheme : :class:`~epyt_flow.visualization.visualization_utils.ColorScheme`
125 Contains the selected ColorScheme for visualization.
126 pipe_parameters : :class:`~epyt_flow.visualization.visualization_utils.EdgeObject`
127 Class contains parameters for visualizing pipes in the correct format
128 for drawing.
129 junction_parameters : :class:`~epyt_flow.visualization.visualization_utils.JunctionObject`
130 Class contains parameters for visualizing junctions in the correct
131 format for drawing.
132 tank_parameters : :class:`~epyt_flow.visualization.visualization_utils.JunctionObject`
133 Class contains parameters for visualizing tanks in the correct format
134 for drawing.
135 reservoir_parameters : :class:`~epyt_flow.visualization.visualization_utils.JunctionObject`
136 Class contains parameters for visualizing reservoirs in the correct
137 format for
138 drawing.
139 valve_parameters : :class:`~epyt_flow.visualization.visualization_utils.JunctionObject`
140 Class contains parameters for visualizing valves in the correct format
141 for drawing.
142 pump_parameters : :class:`~epyt_flow.visualization.visualization_utils.JunctionObject`
143 Class contains parameters for visualizing pumps in the correct format
144 for drawing.
145 colorbars : `dict`
146 A dictionary containing the necessary data for drawing the required
147 colorbars.
148 labels : `dict`
149 A dictionary containing components as keys and drawing information for
150 the labels as values.
151
152 """
153
154 def __init__(self, scenario: ScenarioSimulator,
155 color_scheme: ColorScheme = epyt_flow_colors) -> None:
156 """
157 Initializes the class with a given scenario, sets up the topology,
158 SCADA data, and the classes containing parameters for visualizing
159 various hydraulic components (pipes, junctions, tanks, reservoirs,
160 valves, and pumps).
161
162 Parameters
163 ----------
164 scenario : :class:`~epyt_flow.simulation.scenario_simulator.ScenarioSimulator`
165 An instance of the `ScenarioSimulator` class, used to simulate and
166 retrieve the system topology.
167 color_scheme : :class:`~epyt_flow.visualization.visualization_utils.ColorScheme`
168 Contains the selected ColorScheme for visualization. Default is
169 EPYT_FLOW.
170
171 Raises
172 ------
173 TypeError
174 If `scenario` is not an instance of
175 :class:`~epyt_flow.simulation.scenario_simulator.ScenarioSimulator`.
176 """
177 if not isinstance(scenario, ScenarioSimulator):
178 raise TypeError("'scenario' must be an instance of " +
179 "'epyt_flow.simulation.ScenarioSimulator' " +
180 f"but not of '{type(scenario)}'")
181
182 self.__scenario = scenario
183 self.fig = None
184 self.ax = None
185 self.scada_data = None
186 markers = Marker()
187 self.topology = self.__scenario.get_topology()
188
189 pos_dict = {x: self.topology.get_node_info(x)['coord'] for x in
190 self.topology.get_all_nodes()}
191
192 self.color_scheme = color_scheme
193
194 self.pipe_parameters = EdgeObject(
195 [x[1] for x in self.topology.get_all_links()], pos_dict,
196 self.color_scheme.pipe_color)
197 self.junction_parameters = JunctionObject(
198 self.topology.get_all_junctions(), pos_dict,
199 node_color=self.color_scheme.node_color)
200 self.tank_parameters = JunctionObject(self.topology.get_all_tanks(),
201 pos_dict, node_size=100,
202 node_shape=markers.tank,
203 node_color=self.color_scheme.tank_color)
204 self.reservoir_parameters = JunctionObject(
205 self.topology.get_all_reservoirs(), pos_dict, node_size=100,
206 node_shape=markers.reservoir,
207 node_color=self.color_scheme.reservoir_color)
208 self.valve_parameters = JunctionObject(self.topology.get_all_valves(),
209 self._get_midpoints(
210 self.topology.get_all_valves()),
211 node_size=50,
212 node_shape=markers.valve,
213 node_color=self.color_scheme.valve_color)
214 self.pump_parameters = JunctionObject(self.topology.get_all_pumps(),
215 self._get_midpoints(
216 self.topology.get_all_pumps()),
217 node_size=50,
218 node_shape=markers.pump,
219 node_color=self.color_scheme.pump_color)
220
221 self.colorbars = {}
222 self.labels = {}
223 self.masks = {}
224
225 def _get_midpoints(self, elements: List[str]) -> dict[
226 str, tuple[float, float]]:
227 """
228 Computes and returns the midpoints for drawing either valves or pumps
229 in a water distribution network.
230
231 For each element ID in the provided list, the method calculates the
232 midpoint between its start and end nodes' coordinates.
233
234 Parameters
235 ----------
236 elements : `list[str]`
237 A list of element IDs (e.g., pump IDs, valve IDs) for which to
238 compute the midpoints.
239
240 Returns
241 -------
242 elements_dict : `dict`
243 A dictionary where the keys are element IDs and the values are the
244 corresponding midpoints, represented as 2D coordinates [x, y].
245 """
246 elements_pos_dict = {}
247 for element in elements:
248 if element in self.topology.pumps:
249 start_node, end_node = self.topology.get_pump_info(element)[
250 'end_points']
251 elif element in self.topology.valves:
252 start_node, end_node = self.topology.get_valve_info(element)[
253 'end_points']
254 else:
255 raise ValueError(f"Unknown element '{element}'")
256 start_pos = self.topology.get_node_info(start_node)['coord']
257 end_pos = self.topology.get_node_info(end_node)['coord']
258 pos = [(start_pos[0] + end_pos[0]) / 2,
259 (start_pos[1] + end_pos[1]) / 2]
260 elements_pos_dict[element] = pos
261 return elements_pos_dict
262
263 def _get_next_frame(self, frame_number: int) -> None:
264 """
265 Draws the next frame of a water distribution network animation.
266
267 This method updates a visualization animation with the hydraulic
268 components colored according to the scada data corresponding to the
269 current frame.
270
271 Parameters
272 ----------
273 frame_number : `int`
274 The current frame number used to retrieve the data corresponding to
275 that frame
276 """
277 plt.clf()
278 self.ax = self.fig.add_subplot(111)
279 self.ax.axis('off')
280
281 nxp.draw_networkx_edges(self.topology, ax=self.ax,
282 label='Pipes',
283 **self.pipe_parameters.get_frame(frame_number))
284 my_draw_networkx_nodes(self.topology, ax=self.ax, label='Junctions',
285 **self.junction_parameters.get_frame(frame_number))
286 my_draw_networkx_nodes(self.topology, ax=self.ax, label='Tanks',
287 **self.tank_parameters.get_frame(frame_number))
288 my_draw_networkx_nodes(self.topology, ax=self.ax, label='Reservoirs',
289 **self.reservoir_parameters.get_frame(frame_number))
290 my_draw_networkx_nodes(self.topology, ax=self.ax, label='Valves',
291 **self.valve_parameters.get_frame(frame_number))
292 my_draw_networkx_nodes(self.topology, ax=self.ax, label='Pumps',
293 **self.pump_parameters.get_frame(frame_number))
294
295 for key, mask in self.masks.items():
296 if key == 'nodes':
297 my_draw_networkx_nodes(self.topology, ax=self.ax,
298 **self.junction_parameters.
299 get_frame_mask(mask, self.color_scheme.node_color))
300 if key == 'pumps':
301 my_draw_networkx_nodes(
302 self.topology,
303 ax=self.ax,
304 **self.pump_parameters.get_frame_mask(mask,
305 self.color_scheme.pump_color))
306 if key == 'links':
307 my_draw_networkx_nodes(self.topology, ax=self.ax,
308 **self.pipe_parameters.
309 get_frame_mask(frame_number, self.color_scheme.pipe_color))
310 if key == 'tanks':
311 my_draw_networkx_nodes(self.topology, ax=self.ax,
312 **self.tank_parameters.
313 get_frame_mask(mask, self.color_scheme.tank_color))
314 if key == 'valves':
315 my_draw_networkx_nodes(self.topology, ax=self.ax,
316 **self.valve_parameters.
317 get_frame_mask(mask, self.color_scheme.valve_color))
318
319 self._draw_labels()
320 self.ax.legend(fontsize=6)
321
322 for colorbar_stats in self.colorbars.values():
323 self.fig.colorbar(ax=self.ax, **colorbar_stats)
324
325 def _interpolate_frames(self, num_inter_frames: int):
326 """
327 Interpolates intermediate values between frames using cubic spline
328 interpolation for smoother animation.
329
330 Parameters
331 ----------
332 num_inter_frames : `int`
333 Number of total frames after interpolation.
334
335 Returns
336 -------
337 num_inter_frames : `int`
338 Number of total frames after interpolation.
339 """
340 for node_source in [self.junction_parameters, self.tank_parameters,
341 self.reservoir_parameters, self.valve_parameters,
342 self.pump_parameters]:
343 node_source.interpolate(num_inter_frames)
344 self.pipe_parameters.interpolate(num_inter_frames)
345
346 return num_inter_frames
347
348 def _draw_labels(self):
349 """
350 Method accesses the dict `self.labels` and draws all generated labels
351 within.
352 """
353 for k, v in self.labels.items():
354 if k in ['pipes']:
355 nxp.draw_networkx_edge_labels(self.topology, ax=self.ax, **v)
356 continue
357 nxp.draw_networkx_labels(self.topology, ax=self.ax, **v)
358
359 def _get_sensor_config_nodes_and_links(self):
360 """
361 Iterates through the sensor config and collects all nodes and links
362 within, that have a sensor attached.
363
364 Returns
365 -------
366 highlighted_links : `list`
367 List of all links with sensors.
368 highlighted_nodes : `list`
369 List of all nodes with sensors.
370 """
371 highlighted_nodes = []
372 highlighted_links = []
373
374 sensor_config = self.__scenario.sensor_config
375 highlighted_nodes += (sensor_config.pressure_sensors
376 + sensor_config.demand_sensors
377 + sensor_config.quality_node_sensors)
378 highlighted_links += (sensor_config.flow_sensors
379 + sensor_config.quality_link_sensors)
380 return highlighted_nodes, highlighted_links
381
[docs]
382 def add_labels(self, components: str or list or tuple = (),
383 font_size: int = 8):
384 """
385 Adds labels to hydraulic components according to the specified
386 components.
387
388 Parameters
389 ----------
390 components : `str` or `list` or `tuple`, default is ()
391 Can either be 'all': all components, 'sensor_config': all nodes and
392 pipes which have a sensor attached, or a list of the component
393 names that are to be labeled: nodes, tanks, reservoirs, pipes,
394 valves, pumps. If the list is empty, all nodes are labeled.
395 font_size : `int`, default is 8
396 Font size of the labels.
397 """
398 sc_nodes, sc_links = None, None
399 if components == 'all':
400 components = ['nodes', 'tanks', 'reservoirs', 'pipes', 'valves',
401 'pumps']
402 elif components == 'sensor_config':
403 components = ['nodes', 'tanks', 'reservoirs', 'pipes', 'valves',
404 'pumps']
405 sc_nodes, sc_links = self._get_sensor_config_nodes_and_links()
406
407 elif len(components) == 0:
408 components = ['nodes']
409
410 component_mapping = {
411 'nodes': self.junction_parameters,
412 'tanks': self.tank_parameters,
413 'reservoirs': self.reservoir_parameters,
414 'valves': self.valve_parameters,
415 'pumps': self.pump_parameters
416 }
417
418 for component, parameters in component_mapping.items():
419 if component in components:
420 labels = {n: str(n) for n in parameters.nodelist if
421 sc_nodes is None or n in sc_nodes}
422 self.labels[component] = {'pos': parameters.pos,
423 'labels': labels,
424 'font_size': font_size}
425 if component in ['pumps', 'valves']:
426 self.labels[component]['verticalalignment'] = 'bottom'
427 if 'pipes' in components:
428 labels = {tuple(n[1]): n[0] for n in self.topology.get_all_links()
429 if sc_links is None or n[0] in sc_links}
430 self.labels['pipes'] = {'pos': self.pipe_parameters.pos,
431 'edge_labels': labels,
432 'font_size': font_size}
433
[docs]
434 def show_animation(self, export_to_file: str = None,
435 return_animation: bool = False, duration: int = 5,
436 fps: int = 15, interpolate: bool = True) \
437 -> Optional[FuncAnimation]:
438 """
439 Displays, exports, or returns an animation of a water distribution
440 network over time.
441
442 This method generates an animation of a network and either shows it or
443 returns the :class:`~FuncAnimation` object. Optionally, the animation
444 is saved to a file.
445
446 Parameters
447 ----------
448 export_to_file : `str`, optional
449 The file path where the animation should be saved, if provided.
450 Default is `None`.
451 return_animation : `bool`, optional
452 If `True`, the animation object is returned. If `False`, the
453 animation will be shown, but not returned. Default is `False`.
454 duration : `int`, default is 5
455 Duration of the animation in seconds.
456 fps : `int`, default is 15
457 Frames per seconds, is achieved through interpolation.
458 interpolate : `bool`, default is True
459 Whether to allow interpolating the sensor values or not. Necessary
460 for fixed fps.
461
462 Returns
463 -------
464 anim : :class:`~FuncAnimation` or None
465 Returns the animation object if `return_animation` is `True`.
466 Otherwise, returns `None`.
467 """
468 self.fig = plt.figure(figsize=(6.4, 4.8), dpi=200)
469
470 total_frames = float('inf')
471 for node_source in [self.junction_parameters, self.tank_parameters,
472 self.reservoir_parameters, self.valve_parameters,
473 self.pump_parameters]:
474 if not isinstance(node_source.node_color, str) and len(
475 node_source.node_color) > 1:
476 total_frames = min(total_frames, len(node_source.node_color))
477 if hasattr(self.pipe_parameters, 'edge_color'):
478 if not isinstance(self.pipe_parameters.edge_color, str) and len(
479 self.pipe_parameters.edge_color) > 1:
480 total_frames = min(total_frames,
481 len(self.pipe_parameters.edge_color))
482 if hasattr(self.pipe_parameters, 'width'):
483 if not isinstance(self.pipe_parameters.width, str) and len(
484 self.pipe_parameters.width) > 1:
485 total_frames = min(total_frames,
486 len(self.pipe_parameters.width))
487
488 if total_frames == 0 or total_frames == float('inf'):
489 raise RuntimeError("The color or resize functions must be called "
490 "with a time_step range (pit) > 1 to enable "
491 "animations")
492
493 if interpolate:
494 total_frames = self._interpolate_frames(fps * duration)
495
496 anim = FuncAnimation(self.fig, self._get_next_frame,
497 frames=total_frames,
498 interval=round(duration * 100 / total_frames))
499
500 if export_to_file is not None:
501 anim.save(export_to_file, writer='ffmpeg', fps=fps)
502 if return_animation:
503 plt.close(self.fig)
504 return anim
505 plt.show()
506 return None
507
[docs]
508 def show_plot(self, export_to_file: str = None,
509 suppress_plot: bool = False, dpi: int = None) -> None:
510 """
511 Displays a static plot of the water distribution network.
512
513 This method generates a static plot of the water distribution network,
514 visualizing pipes, junctions, tanks, reservoirs, valves, and pumps.
515 The plot can be displayed and saved to a file.
516
517 Parameters
518 ----------
519 export_to_file : `str`, optional
520 The file path where the plot should be saved, if provided.
521 Default is `None`.
522 suppress_plot : `bool`, default is False
523 If true, no plot is displayed after running this method.
524 dpi : `int`, optional
525 Dpi of the generated plot. If None, standard values (200 when
526 displaying and 900 when saving) are used.
527 """
528 if dpi is None:
529 plot_dpi = 200
530 save_dpi = 900
531 else:
532 plot_dpi = save_dpi = dpi
533 self.fig = plt.figure(figsize=(6.4, 4.8), dpi=plot_dpi)
534 self._get_next_frame(0)
535
536 if export_to_file is not None:
537 plt.savefig(export_to_file, transparent=True, bbox_inches='tight',
538 dpi=save_dpi)
539 if not suppress_plot:
540 plt.show()
541 else:
542 plt.close(self.fig)
543 plt.clf()
544
[docs]
545 def color_nodes(
546 self, data: Optional[Union[ScadaData, np.ndarray]] = None,
547 parameter: str = 'pressure', statistic: str = 'mean',
548 pit: Optional[Union[int, Tuple[int, int]]] = None,
549 species: str = None,
550 colormap: str = 'viridis',
551 intervals: Optional[Union[int, List[Union[int, float]]]] = None,
552 conversion: Optional[dict] = None,
553 show_colorbar: bool = False,
554 use_sensor_data: bool = False) -> None:
555 """
556 Colors the nodes (junctions) in the water distribution network based on
557 the SCADA data and the specified parameters.
558
559 This method either takes or generates SCADA data, applies a statistic
560 to the chosen parameter, optionally groups the results and prepares the
561 results to be either displayed statically ot animated.
562
563 Parameters
564 ----------
565 data : :class:`~epyt_flow.simulation.scada.scada_data.ScadaData` or
566 `numpy.ndarray <https://numpy.org/doc/stable/reference/generated/numpy.ndarray.html>`_, optional
567 The SCADA data object containing node data or a numpy array of the
568 shape nodes*timesteps. If `None`, a simulation is run to generate
569 SCADA data. Default is `None`.
570 parameter : `str`, optional
571 The node data to visualize. Must be 'pressure', 'demand', or
572 'node_quality'. Default is 'pressure'.
573 statistic : `str`, optional
574 The statistic to calculate for the data. Can be 'mean', 'min',
575 'max', or 'time_step'. Default is 'mean'.
576 pit : `int`, `tuple(int, int)`, optional
577 The point in time or range of time steps for the 'time_step'
578 statistic. If a tuple is provided, it should contain two integers
579 representing the start and end time steps. A tuple is necessary to
580 process the data for the :meth:`~ScenarioVisualizer.show_animation`
581 method. Default is `None`.
582 species: `str`, optional
583 Key of species. Only necessary for parameter
584 'bulk_species_concentration'.
585 colormap : `str`, optional
586 The colormap to use for visualizing node values. Default is
587 'viridis'.
588 intervals : `int`, `list[int]` or `list[float]`, optional
589 If provided, the data will be grouped into intervals. It can be an
590 integer specifying the number of groups or a list of boundary
591 points. Default is `None`.
592 conversion : `dict`, optional
593 A dictionary of conversion parameters to convert SCADA data units.
594 Default is `None`.
595 show_colorbar : `bool`, optional
596 If `True`, a colorbar will be displayed on the plot to indicate the
597 range of node values. Default is `False`.
598 use_sensor_data : `bool`, optional
599 If `True`, instead of using raw simulation data, the data recorded
600 by the corresponding sensors in the system is used for the
601 visualization. Note: Not all components may have a sensor attached
602 and sensors may be subject to sensor faults or noise.
603
604 Raises
605 ------
606 ValueError
607 If the `parameter` is not one of 'pressure', 'demand', or
608 'node_quality', or if `pit` is not correctly provided for the
609 'time_step' statistic.
610
611 """
612 self.junction_parameters.cmap = colormap
613
614 if data is not None:
615 self.scada_data = data
616 elif not self.scada_data:
617 self.scada_data = self.__scenario.run_simulation()
618
619 if conversion:
620 self.scada_data = self.scada_data.convert_units(**conversion)
621
622 # TODO: is there any way to make this look better (e.g. do a mapping somewhere??)
623 if parameter == 'pressure':
624 if use_sensor_data:
625 values, self.masks[
626 'nodes'] = self.scada_data.get_data_pressures_as_node_features()
627 else:
628 values = self.scada_data.pressure_data_raw
629 elif parameter == 'demand':
630 if use_sensor_data:
631 values, self.masks[
632 'nodes'] = self.scada_data.get_data_demands_as_node_features()
633 else:
634 values = self.scada_data.demand_data_raw
635 elif parameter == 'node_quality':
636 if use_sensor_data:
637 values, self.masks[
638 'nodes'] = self.scada_data.get_data_nodes_quality_as_node_features()
639 else:
640 values = self.scada_data.node_quality_data_raw
641 elif parameter == 'custom_data':
642 # Custom should have the dimensions (timesteps, nodes)
643 values = self.scada_data
644 elif parameter == 'bulk_species_concentration':
645 if not species:
646 raise ValueError('Species must be set when using bulk_species_'
647 'concentration.')
648 if use_sensor_data:
649 values, self.masks[
650 'nodes'] = self.scada_data.get_data_bulk_species_concentrations_as_node_features()
651 self.masks['nodes'] = self.masks['nodes'][:,
652 self.scada_data.sensor_config.bulk_species.index(
653 species)]
654 values = values[:, :,
655 self.scada_data.sensor_config.bulk_species.index(
656 species)]
657 else:
658 values = self.scada_data.bulk_species_node_concentration_raw[:,
659 self.scada_data.sensor_config.bulk_species.index(
660 species), :]
661 else:
662 raise ValueError(
663 'Parameter must be pressure, demand, node_quality or custom_'
664 'data.')
665
666 if statistic == 'time_step' and isinstance(pit, tuple) and len(
667 pit) == 2 and all(isinstance(i, int) for i in pit):
668 rng = pit
669 if pit[1] == -1:
670 rng = (pit[0], values.shape[0])
671 for frame in range(*rng):
672 if frame > values.shape[0] - 1:
673 break
674 self.junction_parameters.add_frame(statistic, values, frame,
675 intervals)
676 else:
677 self.junction_parameters.add_frame(statistic, values, pit,
678 intervals)
679
680 if show_colorbar:
681 if statistic == 'time_step':
682 label = str(parameter).capitalize() + ' at timestep ' + str(
683 pit)
684 else:
685 label = str(statistic).capitalize() + ' ' + str(
686 parameter).replace('_', ' ')
687 self.colorbars['junctions'] = {'mappable': plt.cm.ScalarMappable(
688 norm=mpl.colors.Normalize(
689 vmin=self.junction_parameters.vmin,
690 vmax=self.junction_parameters.vmax), cmap=colormap),
691 'label': label}
692
[docs]
693 def color_links(
694 self, data: Optional[Union[ScadaData, np.ndarray]] = None,
695 parameter: str = 'flow_rate', statistic: str = 'mean',
696 pit: Optional[Union[int, Tuple[int, int]]] = None,
697 species: str = None,
698 colormap: str = 'coolwarm',
699 intervals: Optional[Union[int, List[Union[int, float]]]] = None,
700 conversion: Optional[dict] = None,
701 show_colorbar: bool = False,
702 use_sensor_data: bool = False) -> None:
703 """
704 Colors the links (pipes) in the water distribution network based on the
705 SCADA data and the specified parameters.
706
707 This method either takes or generates SCADA data, applies a statistic
708 to the chosen parameter, optionally groups the results and prepares the
709 results to be either displayed statically ot animated.
710
711 Parameters
712 ----------
713 data : :class:`~epyt_flow.simulation.scada.scada_data.ScadaData` or
714 `numpy.ndarray <https://numpy.org/doc/stable/reference/generated/numpy.ndarray.html>`_, optional
715 The SCADA data object containing link data or a numpy array of the
716 shape links*timesteps. If `None`, a simulation is run to generate
717 SCADA data. Default is `None`.
718 parameter : `str`, optional
719 The link data to visualize. Options are 'flow_rate', 'link_quality',
720 'custom_data', 'bulk_species_concentration' or 'diameter'.
721 Default is 'flow_rate'.
722 statistic : `str`, optional
723 The statistic to calculate for the data. Can be 'mean', 'min',
724 'max', or 'time_step'. Default is 'mean'.
725 pit : `int` or `tuple(int, int)`, optional
726 The point in time or range of time steps for the 'time_step'
727 statistic. If a tuple is provided, it should contain two integers
728 representing the start and end time steps. A tuple is necessary to
729 process the data for the :func:`~ScenarioVisualizer.show_animation`
730 method. Default is `None`.
731 species: `str`, optional
732 Key of species. Only necessary for parameter
733 'bulk_species_concentration'.
734 colormap : `str`, optional
735 The colormap to use for visualizing link values. Default is
736 'coolwarm'.
737 intervals : `int`, `list[int]`, `list[float]`, optional
738 If provided, the data will be grouped into intervals. It can be an
739 integer specifying the number of groups or a list of boundary
740 points. Default is `None`.
741 conversion : `dict`, optional
742 A dictionary of conversion parameters to convert SCADA data units.
743 Default is `None`.
744 show_colorbar : `bool`, optional
745 If `True`, a colorbar will be displayed on the plot to indicate the
746 range of values. Default is `False`.
747 use_sensor_data : `bool`, optional
748 If `True`, instead of using raw simulation data, the data recorded
749 by the corresponding sensors in the system is used for the
750 visualization. Note: Not all components may have a sensor attached
751 and sensors may be subject to sensor faults or noise.
752
753 Raises
754 ------
755 ValueError
756 If `parameter` is not a valid link data parameter or if `pit` is
757 incorrectly provided for the 'time_step' statistic.
758
759 """
760 sim_length = None
761
762 if data is not None:
763 self.scada_data = data
764 if not isinstance(self.scada_data, ScadaData):
765 sim_length = self.scada_data.shape[0]
766 elif not self.scada_data:
767 self.scada_data = self.__scenario.run_simulation()
768
769 if conversion:
770 self.scada_data = self.scada_data.convert_units(**conversion)
771
772 self.pipe_parameters.edge_cmap = mpl.colormaps[colormap]
773
774 if sim_length is None:
775 sim_length = self.scada_data.sensor_readings_time.shape[0]
776
777 if statistic == 'time_step' and isinstance(pit, tuple) and len(
778 pit) == 2 and all(isinstance(i, int) for i in pit):
779 rng = pit
780 if pit[1] == -1:
781 rng = (pit[0], sim_length)
782 for frame in range(*rng):
783 if frame >= sim_length:
784 break
785 self.pipe_parameters.add_frame(self.topology, 'edge_color',
786 self.scada_data, parameter,
787 statistic, frame, species,
788 intervals, use_sensor_data)
789 else:
790 self.pipe_parameters.add_frame(self.topology, 'edge_color',
791 self.scada_data, parameter,
792 statistic, pit, species, intervals,
793 use_sensor_data)
794
795 if hasattr(self.pipe_parameters, 'mask'):
796 self.masks['links'] = self.pipe_parameters.mask
797
798 if show_colorbar:
799 if statistic == 'time_step':
800 label = (str(parameter).capitalize().replace('_', ' ')
801 + ' at timestep ' + str(pit))
802 else:
803 label = str(statistic).capitalize() + ' ' + str(
804 parameter).replace('_', ' ')
805 self.colorbars['pipes'] = {'mappable': plt.cm.ScalarMappable(
806 norm=mpl.colors.Normalize(
807 vmin=self.pipe_parameters.edge_vmin,
808 vmax=self.pipe_parameters.edge_vmax), cmap=colormap),
809 'label': label}
810
[docs]
811 def color_pumps(
812 self, data: Optional[Union[ScadaData, np.ndarray]] = None,
813 parameter: str = 'efficiency', statistic: str = 'mean',
814 pit: Optional[Union[int, Tuple[int]]] = None,
815 intervals: Optional[Union[int, List[Union[int, float]]]] = None,
816 colormap: str = 'viridis', show_colorbar: bool = False,
817 use_sensor_data: bool = False) -> None:
818 """
819 Colors the pumps in the water distribution network based on SCADA data
820 and the specified parameters.
821
822 This method either takes or generates SCADA data, applies a statistic
823 to the chosen parameter, optionally groups the results and prepares the
824 results to be either displayed statically ot animated.
825
826 Parameters
827 ----------
828 data : :class:`~epyt_flow.simulation.scada.scada_data.ScadaData` or
829 `numpy.ndarray <https://numpy.org/doc/stable/reference/generated/numpy.ndarray.html>`_, optional
830 The SCADA data object containing pump data or a numpy array of the
831 shape pumps*timesteps. If `None`, a simulation is run to generate
832 SCADA data. Default is `None`.
833 parameter : `str`, optional
834 The pump data to visualize. Must be 'efficiency',
835 'energy_consumption', or 'state'. Default is 'efficiency'.
836 statistic : `str`, optional
837 The statistic to calculate for the data. Can be 'mean', 'min',
838 'max', or 'time_step'. Default is 'mean'.
839 pit : `int`, `tuple(int, int)`, optional
840 The point in time or range of time steps for the 'time_step'
841 statistic. If a tuple is provided, it should contain two integers
842 representing the start and end time steps. A tuple is necessary to
843 process the data for the :meth:`~ScenarioVisualizer.show_animation`
844 method. Default is `None`.
845 intervals : `int`, `list[int]`, `list[float]`, optional
846 If provided, the data will be grouped into intervals. It can be an
847 integer specifying the number of groups or a list of boundary
848 points. Default is `None`.
849 colormap : `str`, optional
850 The colormap to use for visualizing pump values. Default is
851 'viridis'.
852 show_colorbar : `bool`, optional
853 If `True`, a colorbar will be displayed on the plot to indicate the
854 range of pump values. Default is `False`.
855 use_sensor_data : `bool`, optional
856 If `True`, instead of using raw simulation data, the data recorded
857 by the corresponding sensors in the system is used for the
858 visualization. Note: Not all components may have a sensor attached
859 and sensors may be subject to sensor faults or noise.
860
861 Raises
862 ------
863 ValueError
864 If the `parameter` is not one of 'efficiency',
865 'energy_consumption', or 'state', or if `pit` is not correctly
866 provided for the 'time_step' statistic.
867
868 """
869
870 self.pump_parameters.cmap = colormap
871
872 if data is not None:
873 self.scada_data = data
874 elif not self.scada_data:
875 self.scada_data = self.__scenario.run_simulation()
876
877 if parameter == 'efficiency':
878 if use_sensor_data:
879 values, self.masks[
880 'pumps'] = self.scada_data.get_data_pumps_efficiency_as_node_features()
881 else:
882 values = self.scada_data.pumps_efficiency_data_raw
883 elif parameter == 'energy_consumption':
884 if use_sensor_data:
885 values, self.masks[
886 'pumps'] = self.scada_data.get_data_pumps_energyconsumption_as_node_features()
887 else:
888 values = self.scada_data.pumps_energyconsumption_data_raw
889 elif parameter == 'state':
890 if use_sensor_data:
891 values, self.masks[
892 'pumps'] = self.scada_data.get_data_pumps_state_as_node_features()
893 else:
894 values = self.scada_data.pumps_state_data_raw
895 elif parameter == 'custom_data':
896 values = self.scada_data
897 else:
898 raise ValueError(
899 'Parameter must be efficiency, energy_consumption, state or custom_data')
900
901 if statistic == 'time_step' and isinstance(pit, tuple) and len(
902 pit) == 2 and all(isinstance(i, int) for i in pit):
903 rng = pit
904 if pit[1] == -1:
905 rng = (pit[0], values.shape[0])
906 for frame in range(*rng):
907 if frame > values.shape[0] - 1:
908 break
909 self.pump_parameters.add_frame(statistic, values, frame,
910 intervals)
911 else:
912 self.pump_parameters.add_frame(statistic, values, pit, intervals)
913
914 if show_colorbar:
915 if statistic == 'time_step':
916 label = str(parameter).capitalize().replace(
917 '_', ' ') + ' at timestep ' + str(pit)
918 else:
919 label = str(statistic).capitalize() + ' ' + str(
920 parameter).replace('_', ' ')
921 self.colorbars['pumps'] = {'mappable': plt.cm.ScalarMappable(
922 norm=mpl.colors.Normalize(vmin=self.pump_parameters.vmin,
923 vmax=self.pump_parameters.vmax),
924 cmap=colormap), 'label': label}
925
[docs]
926 def color_tanks(
927 self, data: Optional[Union[ScadaData, np.ndarray]] = None,
928 statistic: str = 'mean',
929 pit: Optional[Union[int, Tuple[int, int]]] = None,
930 intervals: Optional[Union[int, List[Union[int, float]]]] = None,
931 colormap: str = 'viridis', show_colorbar: bool = False,
932 use_sensor_data: bool = False) -> None:
933 """
934 Colors the tanks in the water distribution network based on the SCADA
935 tank volume data and the specified statistic.
936
937 This method either takes or generates SCADA data, applies a statistic
938 to the tank volume data, optionally groups the results and prepares
939 them to be either displayed statically ot animated.
940
941 Parameters
942 ----------
943 data : :class:`~epyt_flow.simulation.scada.scada_data.ScadaData` or
944 `numpy.ndarray <https://numpy.org/doc/stable/reference/generated/numpy.ndarray.html>`_, optional
945 The SCADA data object containing tank data or a numpy array of the
946 shape tanks*timesteps. If `None`, a simulation is run to generate
947 SCADA data. Default is `None`.
948 statistic : `str`, optional
949 The statistic to calculate for the data. Can be 'mean', 'min',
950 'max', or 'time_step'. Default is 'mean'.
951 pit : `int`, `tuple(int, int)`, optional
952 The point in time or range of time steps for the 'time_step'
953 statistic. If a tuple is provided, it should contain two integers
954 representing the start and end time steps. A tuple is necessary to
955 process the data for the :meth:`~ScenarioVisualizer.show_animation`
956 method. Default is `None`.
957 intervals : `int`, `list[int]`, `list[float]`, optional
958 If provided, the data will be grouped into intervals. It can be an
959 integer specifying the number of groups or a list of boundary
960 points. Default is `None`.
961 colormap : `str`, optional
962 The colormap to use for visualizing tank values. Default is
963 'viridis'.
964 show_colorbar : `bool`, optional
965 If `True`, a colorbar will be displayed on the plot to indicate the
966 range of tank volume values. Default is `False`.
967 use_sensor_data : `bool`, optional
968 If `True`, instead of using raw simulation data, the data recorded
969 by the corresponding sensors in the system is used for the
970 visualization. Note: Not all components may have a sensor attached
971 and sensors may be subject to sensor faults or noise.
972
973 Raises
974 ------
975 ValueError
976 If `pit` is not correctly provided for the 'time_step' statistic.
977
978 """
979 self.tank_parameters.cmap = colormap
980
981 if data is not None:
982 self.scada_data = data
983 elif not self.scada_data:
984 self.scada_data = self.__scenario.run_simulation()
985
986 if isinstance(self.scada_data, ScadaData):
987 if use_sensor_data:
988 values, self.masks[
989 'tanks'] = self.scada_data.get_data_tanks_water_volume_as_node_features()
990 else:
991 values = self.scada_data.tanks_volume_data_raw
992 parameter = 'tank volume'
993 else:
994 values = self.scada_data
995 parameter = 'custom data'
996
997 if statistic == 'time_step' and isinstance(pit, tuple) and len(
998 pit) == 2 and all(isinstance(i, int) for i in pit):
999 rng = pit
1000 if pit[1] == -1:
1001 rng = (pit[0], values.shape[0])
1002 for frame in range(*rng):
1003 if frame > values.shape[0] - 1:
1004 break
1005 self.tank_parameters.add_frame(statistic, values, frame,
1006 intervals)
1007 else:
1008 self.tank_parameters.add_frame(statistic, values, pit, intervals)
1009
1010 if show_colorbar:
1011 if statistic == 'time_step':
1012 label = parameter.capitalize() + ' at timestep ' + str(pit)
1013 else:
1014 label = str(statistic).capitalize() + ' ' + parameter
1015 self.colorbars['tanks'] = {'mappable': plt.cm.ScalarMappable(
1016 norm=mpl.colors.Normalize(vmin=self.tank_parameters.vmin,
1017 vmax=self.tank_parameters.vmin),
1018 cmap=colormap), 'label': label}
1019
[docs]
1020 def color_valves(
1021 self, data: Optional[Union[ScadaData, np.ndarray]] = None,
1022 statistic: str = 'mean',
1023 pit: Optional[Union[int, Tuple[int, int]]] = None,
1024 intervals: Optional[Union[int, List[Union[int, float]]]] = None,
1025 colormap: str = 'viridis', show_colorbar: bool = False,
1026 use_sensor_data: bool = False) -> None:
1027 """
1028 Colors the valves in the water distribution network based on SCADA
1029 valve state data and the specified statistic.
1030
1031 This method either takes or generates SCADA data, applies a statistic
1032 to the valve state data, optionally groups the results and prepares
1033 them to be either displayed statically ot animated.
1034
1035 Parameters
1036 ----------
1037 data : :class:`~epyt_flow.simulation.scada.scada_data.ScadaData` or
1038 `numpy.ndarray <https://numpy.org/doc/stable/reference/generated/numpy.ndarray.html>`_, optional
1039 The SCADA data object containing valve data or a numpy array of the
1040 shape valves*timesteps. If `None`, a simulation is run to generate
1041 SCADA data. Default is `None`.
1042 statistic : `str`, optional
1043 The statistic to calculate for the data. Can be 'mean', 'min',
1044 'max', or 'time_step'. Default is 'mean'.
1045 pit : `int`, `tuple(int)`, optional
1046 The point in time or range of time steps for the 'time_step'
1047 statistic. If a tuple is provided, it should contain two integers
1048 representing the start and end time steps. A tuple is necessary to
1049 process the data for the :meth:`~ScenarioVisualizer.show_animation`
1050 method. Default is `None`.
1051 intervals : `int`, `list[int]`, `list[float]`, optional
1052 If provided, the data will be grouped into intervals. It can be an
1053 integer specifying the number of groups or a list of
1054 boundary points. Default is `None`.
1055 colormap : `str`, optional
1056 The colormap to use for visualizing valve state values. Default is
1057 'viridis'.
1058 show_colorbar : `bool`, optional
1059 If `True`, a colorbar will be displayed on the plot to indicate the
1060 range of valve state values. Default is `False`.
1061 use_sensor_data : `bool`, optional
1062 If `True`, instead of using raw simulation data, the data recorded
1063 by the corresponding sensors in the system is used for the
1064 visualization. Note: Not all components may have a sensor attached
1065 and sensors may be subject to sensor faults or noise.
1066
1067 Raises
1068 ------
1069 ValueError
1070 If `pit` is not correctly provided for the 'time_step' statistic.
1071
1072 """
1073
1074 self.valve_parameters.cmap = colormap
1075
1076 if data is not None:
1077 self.scada_data = data
1078 elif not self.scada_data:
1079 self.scada_data = self.__scenario.run_simulation()
1080
1081 if isinstance(self.scada_data, ScadaData):
1082 if use_sensor_data:
1083 values, self.masks[
1084 'valves'] = self.scada_data.get_data_valves_state_as_node_features()
1085 else:
1086 values = self.scada_data.valves_state_data_raw
1087 parameter = 'valve state'
1088 else:
1089 values = self.scada_data
1090 parameter = 'custom data'
1091
1092 if statistic == 'time_step' and isinstance(pit, tuple) and len(
1093 pit) == 2 and all(isinstance(i, int) for i in pit):
1094 rng = pit
1095 if pit[1] == -1:
1096 rng = (pit[0], values.shape[0])
1097 for frame in range(*rng):
1098 if frame > values.shape[0] - 1:
1099 break
1100 self.valve_parameters.add_frame(statistic, values, frame,
1101 intervals)
1102 else:
1103 self.valve_parameters.add_frame(statistic, values, pit,
1104 intervals)
1105
1106 if show_colorbar:
1107 if statistic == 'time_step':
1108 label = parameter.capitalize() + ' at timestep ' + str(pit)
1109 else:
1110 label = str(statistic).capitalize() + ' ' + 'valve state'
1111 self.colorbars['valves'] = {'mappable': plt.cm.ScalarMappable(
1112 norm=mpl.colors.Normalize(vmin=self.valve_parameters.vmin,
1113 vmax=self.valve_parameters.vmax),
1114 cmap=colormap), 'label': label}
1115
[docs]
1116 def resize_links(
1117 self, data: Optional[Union[ScadaData, np.ndarray]] = None,
1118 parameter: str = 'flow_rate', statistic: str = 'mean',
1119 line_widths: Tuple[int, int] = (1, 2),
1120 pit: Optional[Union[int, Tuple[int, int]]] = None,
1121 species: str = None,
1122 intervals: Optional[Union[int, List[Union[int, float]]]] = None,
1123 conversion: Optional[dict] = None,
1124 use_sensor_data: bool = False) -> None:
1125 """
1126 Resizes the width of the links (pipes) in the water distribution
1127 network based on SCADA data and the specified parameters.
1128
1129 This method either takes or generates SCADA data, applies a statistic,
1130 optionally groups the results and prepares them to be either displayed
1131 statically ot animated as link width.
1132
1133 Parameters
1134 ----------
1135 data : :class:`~epyt_flow.simulation.scada.scada_data.ScadaData` or
1136 `numpy.ndarray <https://numpy.org/doc/stable/reference/generated/numpy.ndarray.html>`_, optional
1137 The SCADA data object containing link data or a numpy array of the
1138 shape links*timesteps. If `None`, a simulation is run to generate
1139 SCADA data. Default is `None`.
1140 parameter : `str`, optional
1141 The data used to resize to. Default is 'flow_rate'.
1142 statistic : `str`, optional
1143 The statistic to calculate for the data. Can be 'mean', 'min',
1144 'max', or 'time_step'. Default is 'mean'.
1145 line_widths : `tuple(int, int)`, optional
1146 A tuple specifying the range of line widths to use when resizing
1147 links based on the data. Default is (1, 2).
1148 pit : `int` or `tuple(int, int)`, optional
1149 The point in time or range of time steps for the 'time_step'
1150 statistic. If a tuple is provided, it should contain two integers
1151 representing the start and end time steps. A tuple is necessary to
1152 process the data for the :meth:`~ScenarioVisualizer.show_animation`
1153 method. Default is `None`.
1154 species: `str`, optional
1155 Key of species. Only necessary for parameter
1156 'bulk_species_concentration'.
1157 intervals : `int` or `list[int]` or `list[float]`, optional
1158 If provided, the data will be grouped into intervals. It can be an
1159 integer specifying the number of groups or a list of boundary
1160 points. Default is `None`.
1161 conversion : `dict`, optional
1162 A dictionary of conversion parameters to convert SCADA data units.
1163 Default is `None`.
1164 use_sensor_data : `bool`, optional
1165 If `True`, instead of using raw simulation data, the data recorded
1166 by the corresponding sensors in the system is used for the
1167 visualization. Note: Not all components may have a sensor attached
1168 and sensors may be subject to sensor faults or noise.
1169 """
1170 sim_length = None
1171
1172 if data is not None:
1173 self.scada_data = data
1174 if not isinstance(self.scada_data, ScadaData):
1175 sim_length = self.scada_data.shape[0]
1176 elif not self.scada_data:
1177 self.scada_data = self.__scenario.run_simulation()
1178
1179 if conversion:
1180 self.scada_data = self.scada_data.convert_units(**conversion)
1181
1182 if sim_length is None:
1183 sim_length = self.scada_data.sensor_readings_time.shape[0]
1184
1185 if statistic == 'time_step' and isinstance(pit, tuple) and len(
1186 pit) == 2 and all(isinstance(i, int) for i in pit):
1187 rng = pit
1188 if pit[1] == -1:
1189 rng = (pit[0], sim_length)
1190 for frame in range(*rng):
1191 if frame >= sim_length:
1192 break
1193 self.pipe_parameters.add_frame(self.topology, 'edge_width',
1194 self.scada_data, parameter,
1195 statistic, frame, species,
1196 intervals, use_sensor_data)
1197 else:
1198 self.pipe_parameters.add_frame(self.topology, 'edge_width',
1199 self.scada_data, parameter,
1200 statistic, pit, species, intervals,
1201 use_sensor_data)
1202 self.pipe_parameters.rescale_widths(line_widths)
1203
[docs]
1204 def hide_nodes(self) -> None:
1205 """
1206 Hides all nodes (junctions) in the water distribution network
1207 visualization.
1208
1209 This method clears the node list from the `junction_parameters`
1210 class, effectively removing all nodes from view in the current
1211 visualization.
1212 """
1213 self.junction_parameters.nodelist = []
1214
[docs]
1215 def highlight_sensor_config(self) -> None:
1216 """
1217 Highlights nodes and links that have sensors in the sensor_config in
1218 the water distribution network visualization.
1219
1220 This method identifies nodes and links equipped with different types of
1221 sensors from the
1222 :class:`~epyt_flow.simulation.sensor_config.SensorConfig` and
1223 updates their visual appearance. Nodes with sensors are highlighted
1224 with an orange border, while links with sensors are displayed with a
1225 dashed line style.
1226 """
1227 highlighted_nodes, highlighted_links = self._get_sensor_config_nodes_and_links()
1228
1229 node_edges = [
1230 (17, 163, 252) if node in highlighted_nodes else (0, 0, 0) for node
1231 in self.topology]
1232 pipe_style = ['dashed' if link[0] in highlighted_links else 'solid' for
1233 link in self.topology.get_all_links()]
1234
1235 self.junction_parameters.add_attributes(
1236 {'linewidths': 1, 'edgecolors': node_edges})
1237 self.pipe_parameters.add_attributes({'style': pipe_style})