epyt_flow.visualization

epyt_flow.visualization.scenario_visualizer

Module provides a class for visualizing scenarios.

class epyt_flow.visualization.scenario_visualizer.Marker

Bases: object

The Marker class provides svg representations of hydraulic components (pump, reservoir, tank and valve), which are loaded from their respective svg paths and transformed into ~matplotlib.path.Path objects in order to be used with the matplotlib library.

pump

Marker for the pump, loaded from PUMP_PATH.

Type:

matplotlib.path.Path

reservoir

Marker for the reservoir, loaded from RESERVOIR_PATH.

Type:

matplotlib.path.Path

tank

Marker for the tank, loaded from TANK_PATH.

Type:

matplotlib.path.Path

valve

Marker for the valve, loaded from VALVE_PATH.

Type:

matplotlib.path.Path

__marker_from_path(path, scale_p=1)

Loads and applies transformations to the marker shape from the given path.

class epyt_flow.visualization.scenario_visualizer.ScenarioVisualizer(scenario: ~epyt_flow.simulation.scenario_simulator.ScenarioSimulator, color_scheme: ~epyt_flow.visualization.visualization_utils.ColorScheme = <epyt_flow.visualization.visualization_utils.ColorScheme object>)

Bases: object

This class provides the necessary function to generate visualizations in the form of plots or animations from water network data.

Given a ScenarioSimulator object, this class provides the necessary functions to plot the network topology and to color hydraulic elements according to simulation data. The resulting plot can then be displayed or saved.

__scenario

ScenarioSimulator object containing the network topology and configurations to obtain the simulation data which should be displayed.

Type:

ScenarioSimulator

fig

Figure object used for plotting, created and customized by calling the methods of this class, initialized as None.

Type:

matplotlib.pyplot.Figure or None

ax

The axes for plotting, initialized as None.

Type:

~matplotlib.axes.Axes or None

scada_data

SCADA data created by the ScenarioSimulator object, initialized as None.

Type:

ScadaData or None

topology

Topology object retrieved from the scenario, containing the structure of the water distribution network.

Type:

NetworkTopology

color_scheme

Contains the selected ColorScheme for visualization.

Type:

ColorScheme

pipe_parameters

Class contains parameters for visualizing pipes in the correct format for drawing.

Type:

EdgeObject

junction_parameters

Class contains parameters for visualizing junctions in the correct format for drawing.

Type:

JunctionObject

tank_parameters

Class contains parameters for visualizing tanks in the correct format for drawing.

Type:

JunctionObject

reservoir_parameters

Class contains parameters for visualizing reservoirs in the correct format for drawing.

Type:

JunctionObject

valve_parameters

Class contains parameters for visualizing valves in the correct format for drawing.

Type:

JunctionObject

pump_parameters

Class contains parameters for visualizing pumps in the correct format for drawing.

Type:

JunctionObject

colorbars

A dictionary containing the necessary data for drawing the required colorbars.

Type:

dict

labels

A dictionary containing components as keys and drawing information for the labels as values.

Type:

dict

add_labels(components: str = (), font_size: int = 8)

Adds labels to hydraulic components according to the specified components.

Parameters:

• components (str or list or tuple, default is ()) – Can either be ‘all’: all components, ‘sensor_config’: all nodes and pipes which have a sensor attached, or a list of the component names that are to be labeled: nodes, tanks, reservoirs, pipes, valves, pumps. If the list is empty, all nodes are labeled.

• font_size (int, default is 8) – Font size of the labels.

color_links(data: ScadaData | numpy.ndarray | None = None, parameter: str = 'flow_rate', statistic: str = 'mean', pit: int | Tuple[int, int] | None = None, species: str | None = None, colormap: str = 'coolwarm', intervals: int | List[int | float] | None = None, conversion: dict | None = None, show_colorbar: bool = False, use_sensor_data: bool = False) → None

Colors the links (pipes) in the water distribution network based on the SCADA data and the specified parameters.

This method either takes or generates SCADA data, applies a statistic to the chosen parameter, optionally groups the results and prepares the results to be either displayed statically ot animated.

Parameters:

• data (ScadaData or) – numpy.ndarray, optional The SCADA data object containing link data or a numpy array of the shape links*timesteps. If None, a simulation is run to generate SCADA data. Default is None.

• parameter (str, optional) – The link data to visualize. Options are ‘flow_rate’, ‘link_quality’, ‘custom_data’, ‘bulk_species_concentration’ or ‘diameter’. Default is ‘flow_rate’.

• statistic (str, optional) – The statistic to calculate for the data. Can be ‘mean’, ‘min’, ‘max’, or ‘time_step’. Default is ‘mean’.

• pit (int or tuple(int, int), optional) – The point in time or range of time steps for the ‘time_step’ statistic. If a tuple is provided, it should contain two integers representing the start and end time steps. A tuple is necessary to process the data for the show_animation() method. Default is None.

• species (str, optional) – Key of species. Only necessary for parameter ‘bulk_species_concentration’.

• colormap (str, optional) – The colormap to use for visualizing link values. Default is ‘coolwarm’.

• intervals (int, list[int], list[float], optional) – If provided, the data will be grouped into intervals. It can be an integer specifying the number of groups or a list of boundary points. Default is None.

• conversion (dict, optional) – A dictionary of conversion parameters to convert SCADA data units. Default is None.

• show_colorbar (bool, optional) – If True, a colorbar will be displayed on the plot to indicate the range of values. Default is False.

• use_sensor_data (bool, optional) – If True, instead of using raw simulation data, the data recorded by the corresponding sensors in the system is used for the visualization. Note: Not all components may have a sensor attached and sensors may be subject to sensor faults or noise.

Raises:

ValueError – If parameter is not a valid link data parameter or if pit is incorrectly provided for the ‘time_step’ statistic.

color_nodes(data: ScadaData | numpy.ndarray | None = None, parameter: str = 'pressure', statistic: str = 'mean', pit: int | Tuple[int, int] | None = None, species: str | None = None, colormap: str = 'viridis', intervals: int | List[int | float] | None = None, conversion: dict | None = None, show_colorbar: bool = False, use_sensor_data: bool = False) → None

Colors the nodes (junctions) in the water distribution network based on the SCADA data and the specified parameters.

This method either takes or generates SCADA data, applies a statistic to the chosen parameter, optionally groups the results and prepares the results to be either displayed statically ot animated.

Parameters:

• data (ScadaData or) –

  numpy.ndarray, optional The SCADA data object containing node data or a numpy array of the shape nodes*timesteps. If None, a simulation is run to generate SCADA data. Default is None.

• parameter (str, optional) – The node data to visualize. Must be ‘pressure’, ‘demand’, or ‘node_quality’. Default is ‘pressure’.

• statistic (str, optional) – The statistic to calculate for the data. Can be ‘mean’, ‘min’, ‘max’, or ‘time_step’. Default is ‘mean’.

• pit (int, tuple(int, int), optional) – The point in time or range of time steps for the ‘time_step’ statistic. If a tuple is provided, it should contain two integers representing the start and end time steps. A tuple is necessary to process the data for the show_animation() method. Default is None.

• species (str, optional) – Key of species. Only necessary for parameter ‘bulk_species_concentration’.

• colormap (str, optional) – The colormap to use for visualizing node values. Default is ‘viridis’.

• intervals (int, list[int] or list[float], optional) – If provided, the data will be grouped into intervals. It can be an integer specifying the number of groups or a list of boundary points. Default is None.

• conversion (dict, optional) – A dictionary of conversion parameters to convert SCADA data units. Default is None.

• show_colorbar (bool, optional) – If True, a colorbar will be displayed on the plot to indicate the range of node values. Default is False.

• use_sensor_data (bool, optional) – If True, instead of using raw simulation data, the data recorded by the corresponding sensors in the system is used for the visualization. Note: Not all components may have a sensor attached and sensors may be subject to sensor faults or noise.

Raises:

ValueError – If the parameter is not one of ‘pressure’, ‘demand’, or ‘node_quality’, or if pit is not correctly provided for the ‘time_step’ statistic.

color_pumps(data: ScadaData | numpy.ndarray | None = None, parameter: str = 'efficiency', statistic: str = 'mean', pit: int | Tuple[int] | None = None, intervals: int | List[int | float] | None = None, colormap: str = 'viridis', show_colorbar: bool = False, use_sensor_data: bool = False) → None

Colors the pumps in the water distribution network based on SCADA data and the specified parameters.

This method either takes or generates SCADA data, applies a statistic to the chosen parameter, optionally groups the results and prepares the results to be either displayed statically ot animated.

Parameters:

• data (ScadaData or) –

  numpy.ndarray, optional The SCADA data object containing pump data or a numpy array of the shape pumps*timesteps. If None, a simulation is run to generate SCADA data. Default is None.

• parameter (str, optional) – The pump data to visualize. Must be ‘efficiency’, ‘energy_consumption’, or ‘state’. Default is ‘efficiency’.

• statistic (str, optional) – The statistic to calculate for the data. Can be ‘mean’, ‘min’, ‘max’, or ‘time_step’. Default is ‘mean’.

• pit (int, tuple(int, int), optional) – The point in time or range of time steps for the ‘time_step’ statistic. If a tuple is provided, it should contain two integers representing the start and end time steps. A tuple is necessary to process the data for the show_animation() method. Default is None.

• intervals (int, list[int], list[float], optional) – If provided, the data will be grouped into intervals. It can be an integer specifying the number of groups or a list of boundary points. Default is None.

• colormap (str, optional) – The colormap to use for visualizing pump values. Default is ‘viridis’.

• show_colorbar (bool, optional) – If True, a colorbar will be displayed on the plot to indicate the range of pump values. Default is False.

• use_sensor_data (bool, optional) – If True, instead of using raw simulation data, the data recorded by the corresponding sensors in the system is used for the visualization. Note: Not all components may have a sensor attached and sensors may be subject to sensor faults or noise.

Raises:

ValueError – If the parameter is not one of ‘efficiency’, ‘energy_consumption’, or ‘state’, or if pit is not correctly provided for the ‘time_step’ statistic.

color_tanks(data: ScadaData | numpy.ndarray | None = None, statistic: str = 'mean', pit: int | Tuple[int, int] | None = None, intervals: int | List[int | float] | None = None, colormap: str = 'viridis', show_colorbar: bool = False, use_sensor_data: bool = False) → None

Colors the tanks in the water distribution network based on the SCADA tank volume data and the specified statistic.

This method either takes or generates SCADA data, applies a statistic to the tank volume data, optionally groups the results and prepares them to be either displayed statically ot animated.

Parameters:

• data (ScadaData or) –

  numpy.ndarray, optional The SCADA data object containing tank data or a numpy array of the shape tanks*timesteps. If None, a simulation is run to generate SCADA data. Default is None.

• statistic (str, optional) – The statistic to calculate for the data. Can be ‘mean’, ‘min’, ‘max’, or ‘time_step’. Default is ‘mean’.

• pit (int, tuple(int, int), optional) – The point in time or range of time steps for the ‘time_step’ statistic. If a tuple is provided, it should contain two integers representing the start and end time steps. A tuple is necessary to process the data for the show_animation() method. Default is None.

• intervals (int, list[int], list[float], optional) – If provided, the data will be grouped into intervals. It can be an integer specifying the number of groups or a list of boundary points. Default is None.

• colormap (str, optional) – The colormap to use for visualizing tank values. Default is ‘viridis’.

• show_colorbar (bool, optional) – If True, a colorbar will be displayed on the plot to indicate the range of tank volume values. Default is False.

• use_sensor_data (bool, optional) – If True, instead of using raw simulation data, the data recorded by the corresponding sensors in the system is used for the visualization. Note: Not all components may have a sensor attached and sensors may be subject to sensor faults or noise.

Raises:

ValueError – If pit is not correctly provided for the ‘time_step’ statistic.

color_valves(data: ScadaData | numpy.ndarray | None = None, statistic: str = 'mean', pit: int | Tuple[int, int] | None = None, intervals: int | List[int | float] | None = None, colormap: str = 'viridis', show_colorbar: bool = False, use_sensor_data: bool = False) → None

Colors the valves in the water distribution network based on SCADA valve state data and the specified statistic.

This method either takes or generates SCADA data, applies a statistic to the valve state data, optionally groups the results and prepares them to be either displayed statically ot animated.

Parameters:

• data (ScadaData or) –

  numpy.ndarray, optional The SCADA data object containing valve data or a numpy array of the shape valves*timesteps. If None, a simulation is run to generate SCADA data. Default is None.

• statistic (str, optional) – The statistic to calculate for the data. Can be ‘mean’, ‘min’, ‘max’, or ‘time_step’. Default is ‘mean’.

• pit (int, tuple(int), optional) – The point in time or range of time steps for the ‘time_step’ statistic. If a tuple is provided, it should contain two integers representing the start and end time steps. A tuple is necessary to process the data for the show_animation() method. Default is None.

• intervals (int, list[int], list[float], optional) – If provided, the data will be grouped into intervals. It can be an integer specifying the number of groups or a list of boundary points. Default is None.

• colormap (str, optional) – The colormap to use for visualizing valve state values. Default is ‘viridis’.

• show_colorbar (bool, optional) – If True, a colorbar will be displayed on the plot to indicate the range of valve state values. Default is False.

• use_sensor_data (bool, optional) – If True, instead of using raw simulation data, the data recorded by the corresponding sensors in the system is used for the visualization. Note: Not all components may have a sensor attached and sensors may be subject to sensor faults or noise.

Raises:

ValueError – If pit is not correctly provided for the ‘time_step’ statistic.

hide_nodes() → None

Hides all nodes (junctions) in the water distribution network visualization.

This method clears the node list from the junction_parameters class, effectively removing all nodes from view in the current visualization.

highlight_sensor_config() → None

Highlights nodes and links that have sensors in the sensor_config in the water distribution network visualization.

This method identifies nodes and links equipped with different types of sensors from the SensorConfig and updates their visual appearance. Nodes with sensors are highlighted with an orange border, while links with sensors are displayed with a dashed line style.

resize_links(data: ScadaData | numpy.ndarray | None = None, parameter: str = 'flow_rate', statistic: str = 'mean', line_widths: Tuple[int, int] = (1, 2), pit: int | Tuple[int, int] | None = None, species: str | None = None, intervals: int | List[int | float] | None = None, conversion: dict | None = None, use_sensor_data: bool = False) → None

Resizes the width of the links (pipes) in the water distribution network based on SCADA data and the specified parameters.

This method either takes or generates SCADA data, applies a statistic, optionally groups the results and prepares them to be either displayed statically ot animated as link width.

Parameters:

• data (ScadaData or) –

  numpy.ndarray, optional The SCADA data object containing link data or a numpy array of the shape links*timesteps. If None, a simulation is run to generate SCADA data. Default is None.

• parameter (str, optional) – The data used to resize to. Default is ‘flow_rate’.

• statistic (str, optional) – The statistic to calculate for the data. Can be ‘mean’, ‘min’, ‘max’, or ‘time_step’. Default is ‘mean’.

• line_widths (tuple(int, int), optional) – A tuple specifying the range of line widths to use when resizing links based on the data. Default is (1, 2).

• pit (int or tuple(int, int), optional) – The point in time or range of time steps for the ‘time_step’ statistic. If a tuple is provided, it should contain two integers representing the start and end time steps. A tuple is necessary to process the data for the show_animation() method. Default is None.

• species (str, optional) – Key of species. Only necessary for parameter ‘bulk_species_concentration’.

• intervals (int or list[int] or list[float], optional) – If provided, the data will be grouped into intervals. It can be an integer specifying the number of groups or a list of boundary points. Default is None.

• conversion (dict, optional) – A dictionary of conversion parameters to convert SCADA data units. Default is None.

• use_sensor_data (bool, optional) – If True, instead of using raw simulation data, the data recorded by the corresponding sensors in the system is used for the visualization. Note: Not all components may have a sensor attached and sensors may be subject to sensor faults or noise.

show_animation(export_to_file: str | None = None, return_animation: bool = False, duration: int = 5, fps: int = 15, interpolate: bool = True) → matplotlib.animation.FuncAnimation | None

Displays, exports, or returns an animation of a water distribution network over time.

This method generates an animation of a network and either shows it or returns the FuncAnimation object. Optionally, the animation is saved to a file.

Parameters:

• export_to_file (str, optional) – The file path where the animation should be saved, if provided. Default is None.

• return_animation (bool, optional) – If True, the animation object is returned. If False, the animation will be shown, but not returned. Default is False.

• duration (int, default is 5) – Duration of the animation in seconds.

• fps (int, default is 15) – Frames per seconds, is achieved through interpolation.

• interpolate (bool, default is True) – Whether to allow interpolating the sensor values or not. Necessary for fixed fps.

Returns:

anim – Returns the animation object if return_animation is True. Otherwise, returns None.

Return type:

FuncAnimation or None

show_plot(export_to_file: str | None = None, suppress_plot: bool = False, dpi: int | None = None) → None

Displays a static plot of the water distribution network.

This method generates a static plot of the water distribution network, visualizing pipes, junctions, tanks, reservoirs, valves, and pumps. The plot can be displayed and saved to a file.

Parameters:

• export_to_file (str, optional) – The file path where the plot should be saved, if provided. Default is None.

• suppress_plot (bool, default is False) – If true, no plot is displayed after running this method.

• dpi (int, optional) – Dpi of the generated plot. If None, standard values (200 when displaying and 900 when saving) are used.

epyt_flow.visualization.visualization_utils

Module provides helper functions and data management classes for visualizing scenarios.

class epyt_flow.visualization.visualization_utils.ColorScheme(pipe_color: str, node_color: str, pump_color: str, tank_color: str, reservoir_color: str, valve_color: str)

Bases: JsonSerializable

A class containing the color scheme for the ScenarioVisualizer.

static file_ext() → str

Returns the file extension of this class.

This function is automatically implemented by applying the serializable() decorator.

Returns:

File extension.

Return type:

str

get_attributes() → dict

Gets all attributes needed for serialization.

Returns:

attr

Return type:

A dictionary containing all attributes to be serialized.

class epyt_flow.visualization.visualization_utils.EdgeObject(edgelist: list, pos: dict, edge_color: str | list = 'k', interpolated: dict = {})

Bases: object

Represents an edge component (pipes) in a water distribution network and manages all relevant attributes for drawing.

edgelist

List of all edges in WDN pertaining to this component type.

Type:

list

pos

A dictionary mapping pipes to their coordinates in the correct format for drawing.

Type:

dict

edge_color

If string: the color for all edges, if list: a list of lists containing a numerical value for each edge per frame, which will be used for coloring.

Type:

str or list, default = ‘k’

interpolated

Filled with interpolated frames if interpolation method is called.

Type:

dict, default = {}

add_attributes(attributes: dict)

Adds the given attributes dict as class attributes.

Parameters:

attributes (dict) – Attributes dict, which is to be added as class attributes.

add_frame(topology, edge_param: str, scada_data: ScadaData | None, parameter: str = 'flow_rate', statistic: str = 'mean', pit: int | Tuple[int] | None = None, species: str | None = None, intervals: int | List[int | float] | None = None, use_sensor_data: bool | None = None)

Adds a new frame of edge_color or edge width based on the given data and statistic.

Parameters:

• topology (NetworkTopology) – Topology object retrieved from the scenario, containing the structure of the water distribution network.

• edge_param (str) – Method can be called with edge_width or edge_color to calculate either the width or color for the next frame.

• scada_data (ScadaData) – SCADA data created by the ScenarioSimulator instance, is used to retrieve data for the next frame.

• parameter (str, default = ‘flow_rate’) – The link data to visualize. Options are ‘flow_rate’, ‘link_quality’, ‘custom_data’, ‘bulk_species_concentration’ or ‘diameter’. Default is ‘flow_rate’.

• statistic (str, default = ‘mean’) –

  The statistic to calculate for the data. Can be ‘mean’, ‘min’,

  ’max’ or ‘time_step’.

• pit (int) – The point in time for the ‘time_step’ statistic.

• species (str, optional) – Key of the species. Necessary only for parameter ‘bulk_species_concentration’.

• intervals (int, list[int] or list[float]) – If provided, the data will be grouped into intervals. It can be an integer specifying the number of groups or a list of boundary points.

• use_sensor_data (bool, optional) – If True, instead of using raw simulation data, the data recorded by the corresponding sensors in the system is used for the visualization. Note: Not all components may have a sensor attached and sensors may be subject to sensor faults or noise.

Raises:

ValueError – If parameter, interval, pit or statistic is not set correctly.

get_frame(frame_number: int = 0)

Returns all attributes necessary for networkx to draw the specified frame.

Parameters:

frame_number (int, default = 0) – The frame whose parameters should be returned. Default is 0, this is also used if only 1 frame exists (e.g. for plots, not animations).

Returns:

valid_params – A dictionary containing all attributes that function as parameters for networkx.drawing.nx_pylab.draw_networkx_edges().

Return type:

dict

get_frame_mask(frame_number: int = 0, color='k')

Returns all attributes necessary for networkx to draw the specified frame mask.

Parameters:

• frame_number (int, default = 0) – The frame whose parameters should be returned. Default is 0, this is also used if only 1 frame exists (e.g. for plots, not animations).

• color – The default color of masked nodes.

Returns:

valid_params – A dictionary containing all attributes that function as parameters for networkx.drawing.nx_pylab.draw_networkx_edges().

Return type:

dict

interpolate(num_inter_frames: int)

Interpolates edge_color and width values for smoother animations.

Parameters:

num_inter_frames (int) – The number of total frames after interpolation.

rescale_widths(line_widths: Tuple[int, int] = (1, 2))

Rescales all edge widths to the given interval.

Parameters:

• line_widths (Tuple[int], default = (1, 2))

• value (Min and max)

• scaled. (to which the edge widths should be)

Raises:

AttributeError – If no edge width attribute exists yet.

class epyt_flow.visualization.visualization_utils.JunctionObject(nodelist: list, pos: dict, node_shape: matplotlib.path.Path | None = None, node_size: int = 10, node_color: str | list = 'k', interpolated: bool = False)

Bases: object

Represents a junction component (e.g. nodes, tanks, reservoirs, …) in a water distribution network and manages all relevant attributes for drawing.

nodelist

List of all nodes in WDN pertaining to this component type.

Type:

list

pos

A dictionary mapping nodes to their coordinates in the correct format for drawing.

Type:

dict

node_shape

A shape representing the object, if none, the networkx default circle is used.

Type:

matplotlib.path.Path or None

node_size

The size of each node.

Type:

int, default = 10

node_color

If string: the color for all nodes, if list: a list of lists containing a numerical value for each node per frame, which will be used for coloring.

Type:

str or list, default = ‘k’

interpolated

Set to True, if node_colors are interpolated for smoother animation.

Type:

bool, default = False

add_attributes(attributes: dict)

Adds the given attributes dict as class attributes.

Parameters:

attributes (dict) – Attributes dict, which is to be added as class attributes.

add_frame(statistic: str, values: numpy.ndarray, pit: int, intervals: int | List[int | float])

Adds a new frame of node_color based on a given statistic.

Parameters:

• statistic (str) –

  The statistic to calculate for the data. Can be ‘mean’, ‘min’,

  ’max’ or ‘time_step’.

• values – The node values over time as extracted from the scada data.

• pit (int) – The point in time for the ‘time_step’ statistic.

• intervals (int, list[int] or list[float]) – If provided, the data will be grouped into intervals. It can be an integer specifying the number of groups or a list of boundary points.

Raises:

ValueError – If interval, pit or statistic is not correctly provided.

get_frame(frame_number: int = 0)

Returns all attributes necessary for networkx to draw the specified frame.

Parameters:

frame_number (int, default = 0) – The frame whose parameters should be returned. Default is 0, this is also used if only 1 frame exists (e.g. for plots, not animations).

Returns:

valid_params – A dictionary containing all attributes that function as parameters for networkx.drawing.nx_pylab.draw_networkx_nodes().

Return type:

dict

get_frame_mask(mask, color)

Returns all attributes necessary for networkx to draw the specified frame mask. Meaning covering all masked junction objects with the default value.

Parameters:

• mask (np.ndarray) – An array consisting of 0 and 1, where 0 means no sensor. Nodes without sensor are to be masked.

• color – The default color of masked nodes.

Returns:

valid_params – A dictionary containing all attributes that function as parameters for networkx.drawing.nx_pylab.draw_networkx_nodes().

Return type:

dict

interpolate(num_inter_frames: int)

Interpolates node_color values for smoother animations.

Parameters:

num_inter_frames (int) – The number of total frames after interpolation.