epyt_flow

epyt_flow.serialization

Module provides functions and classes for serialization.

class epyt_flow.serialization.JsonSerializable(_parent_path: str = '', **kwds)[source]

Bases: Serializable

Base class for JSON serializable classes. Inherits from Serializable.

static load_from_json(data: str) → Any[source]

Loads (i.e. deserializes) an instance of this class from given JSON data.

Parameters:

data (str) – JSON data.

Returns:

Deserialized instance of this class.

Return type:

Any

static load_from_json_file(f_in: str) → Any[source]

Deserializes an instance of this class from a JSON file.

Parameters:

f_in (str) – Path to the JSON file from which to deserialize the object.

Returns:

Deserialized object.

Return type:

Any

save_to_json_file(f_out: str) → None[source]

Serializes this instance and stores it in a JSON file.

Parameters:

f_in (str) – Path to the JSON file where this serialized object will be stored.

to_json() → str[source]

Serializes this instance to JSON.

Returns:

JSON data.

Return type:

str

class epyt_flow.serialization.Serializable(_parent_path: str = '', **kwds)[source]

Bases: ABC

Base class for a serializable class – must be used in conjunction with the serializable() decorator.

dump(stream_out: BufferedIOBase | None = None) → Any[source]

Serializes this object to a byte array.

Parameters:

stream_out (io.BufferedIOBase, optional) –

Stream to which the serialized object is written. If None, the serialized object is returned as a bytes array.

The default is None.

Returns:

If stream_out is None, the serialized object is returned as a bytes array.

Return type:

bytes

file_ext() → str[source]

Returns the file extension of this class.

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

Returns:

File extension.

Return type:

str

abstract get_attributes() → dict[source]

Gets all attributes to be serialized – these attributes are passed to the constructor when the object is deserialized.

Returns:

Dictionary of attributes – i.e. pairs of attribute name + value.

Return type:

dict

static load(data: bytes | BufferedIOBase) → Any[source]

Deserializes an instance of this class.

Parameters:

data (bytes or io.BufferedIOBase) – Serialized data or stream from which serialized data can be read.

Returns:

Deserialized object.

Return type:

Any

static load_from_file(f_in: str, use_zip: bool = True) → Any[source]

Deserializes an instance of this class from a (compressed) file.

Parameters:

  • f_in (str) – Path to the file from which to deserialize the object.

  • use_zip (bool, optional) –

    If True, the file f_in is supposed to be zip compressed – False, if no compression was used when serializing the object.

    The default is True.

Returns:

Deserialized object.

Return type:

Any

packb() → bytes[source]

Serializes the attributes of this object.

Returns:

Serialized object.

Return type:

bytes

save_to_file(f_out: str, use_zip: bool = True) → None[source]

Serializes this instance and stores it in a (compressed) file.

Parameters:

  • f_in (str) – Path to the file where this serialized object will be stored.

  • use_zip (bool, optional) –

    If True, the file f_in will be zip compressed – False, if no compression is wanted.

    The default is True.

epyt_flow.serialization.dump(data: Any, stream_out: BufferedIOBase | None = None) → bytes | None[source]

Serializes some given data to a byte array.

Parameters:

stream_out (io.BufferedIOBase, optional) –

Stream to which the serialized object is written. If None, the serialized object is returned as a bytes array.

The default is None.

Returns:

Serialized data if stream_out is None – otherwise, nothing is returned.

Return type:

bytes

epyt_flow.serialization.load(data: bytes | BufferedIOBase) → Any[source]

Deserializes data.

Parameters:

data (bytes or io.BufferedIOBase) – Serialized data or stream from which serialized data can be read.

Returns:

Deserialized data.

Return type:

Any

epyt_flow.serialization.load_from_file(f_in: str, use_compression: bool = True) → Any[source]

Deserializes data from a (compressed) file.

Parameters:

  • f_in (str) – Path to the file from which to deserialize the data.

  • use_compression (bool, optional) –

    If True, the file f_in is supposed to be gzip compressed – False, if no compression was used when serializing the data.

    The default is True.

Returns:

Deserialized data.

Return type:

Any

epyt_flow.serialization.my_load_from_json(data: str) → Any[source]

Loads (i.e. deserializes) an object from given JSON data.

Parameters:

data (str) – JSON data.

Returns:

Deserialized object.

Return type:

Any

epyt_flow.serialization.my_packb(data: Any) → bytes[source]

Overriden umsgpack.packb method to support custom serialization handlers.

epyt_flow.serialization.my_to_json(obj: Any) → str[source]

Serializes a given object to JSON.

Parameters:

obj (Any) – Object to be serialized.

Returns:

JSON data.

Return type:

str

epyt_flow.serialization.my_unpackb(data: Any) → Any[source]

Overriden umsgpack.unpackb method to support custom serialization handlers.

epyt_flow.serialization.save_to_file(f_out: str, data: Any, use_compression: bool = True) → None[source]

Serializes data and stores it in a (compressed) file.

Parameters:

  • f_in (str) – Path to the file where the serialized data will be stored.

  • use_compression (bool, optional) –

    If True, the file f_in will be gzip compressed – False, if no compression is wanted.

    The default is True.

epyt_flow.serialization.serializable(my_id: int, my_file_ext: str) → Any[source]

Decorator for a serializable class – i.e. subclass of Serializable.

This decorator registers a new class as a serializable class.

Parameters:

  • my_id (int) – ID of the class.

  • my_file_ext (str) – File extension.

epyt_flow.topology

Module provides a class for representing the topology of WDN.

class epyt_flow.topology.NetworkTopology(f_inp: str, nodes: list[tuple[str, dict]], links: list[tuple[str, tuple[str, str], dict]], pumps: dict, valves: dict, curves: dict[str, tuple[int, list[tuple[float, float]]]], patterns: dict[str, list[float]], flow_units: int, pressure_units: int, **kwds)[source]

Bases: Graph, JsonSerializable

Class representing the topology of a WDN.

Parameters:

  • f_inp (str) – Path to .inp file to which this topology belongs.

  • nodes (list[tuple[str, dict]]) – List of all nodes – i.e. node ID and node information such as type and elevation.

  • links (list[tuple[str, tuple[str, str], dict]]) – List of all links/pipes – i.e. link ID, ID of connecting nodes, and link information such as pipe diameter, length, etc.

  • pumps (dict) – List of all pumps – i.e. valve ID, and information such as pump type and connecting nodes.

  • valves (dict) – List of all valves – i.e. valve ID, and information such as valve type and connecting nodes.

  • curves (dict[str, tuple[int, list[tuple[float, float]]]]) – All curves – i.e. curve ID, and list of points.

  • patterns (dict[str, list[float]]) – All time patterns – i.e., pattern ID and list of multipliers.

  • flow_units (int) –

    Flow units ID.

    Must be one of the following EPANET constants:

    • EN_CFS = 0 (cubic foot/sec)

    • EN_GPM = 1 (gal/min)

    • EN_MGD = 2 (Million gal/day)

    • EN_IMGD = 3 (Imperial MGD)

    • EN_AFD = 4 (ac-foot/day)

    • EN_LPS = 5 (liter/sec)

    • EN_LPM = 6 (liter/min)

    • EN_MLD = 7 (Megaliter/day)

    • EN_CMH = 8 (cubic meter/hr)

    • EN_CMD = 9 (cubic meter/day)

    • EN_CMS = 10 (cubic meter/sec)

  • pressure_units (int) –

    Pressue unit ID.

    Must be one of the following EPANET constants:

    • EN_PSI = 0 (Pounds per square inch)

    • EN_KPA = 1 (Kilopascals)

    • EN_METERS = 2 (Meters)

    • EN_BAR = 3 (Bar)

    • EN_FEET = 4 (Feet)

convert_units(flow_units: int, pressure_units: int) → Any[source]

Converts this instance to a NetworkTopology instance where everything is measured in given flow and pressure units.

Parameters:

  • flow_units (int) –

    Flow unit ID.

    Must be one of the following EPANET constants:

    • EN_CFS = 0 (cubic foot/sec)

    • EN_GPM = 1 (gal/min)

    • EN_MGD = 2 (Million gal/day)

    • EN_IMGD = 3 (Imperial MGD)

    • EN_AFD = 4 (ac-foot/day)

    • EN_LPS = 5 (liter/sec)

    • EN_LPM = 6 (liter/min)

    • EN_MLD = 7 (Megaliter/day)

    • EN_CMH = 8 (cubic meter/hr)

    • EN_CMD = 9 (cubic meter/day)

    • EN_CMS = 10 (cubic meter/sec)

  • pressure_units (int) –

    Pressure unit ID.

    Must be one of the following EPANET constants:

    • EN_PSI = 0 (Pounds per square inch)

    • EN_KPA = 1 (Kilopascals)

    • EN_METERS = 2 (Meters)

    • EN_BAR = 3 (Bar)

    • EN_FEET = 4 (Feet)

Returns:

Network topology with the new measurements units.

Return type:

NetworkTopology

property curves: dict[str, tuple[int, list[tuple[float, float]]]]

Gets all curves – i.e., ID and list of points.

Returns:

All curves.

Return type:

dict[str, tuple[int, list[tuple[float, float]]]]

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

property flow_units: int

Return the flow unit ID.

Will be one of the following EPANET constants:

  • EN_CFS = 0 (cubic foot/sec)

  • EN_GPM = 1 (gal/min)

  • EN_MGD = 2 (Million gal/day)

  • EN_IMGD = 3 (Imperial MGD)

  • EN_AFD = 4 (ac-foot/day)

  • EN_LPS = 5 (liter/sec)

  • EN_LPM = 6 (liter/min)

  • EN_MLD = 7 (Megaliter/day)

  • EN_CMH = 8 (cubic meter/hr)

  • EN_CMD = 9 (cubic meter/day)

  • EN_CMS = 10 (cubic meter/sec)

Returns:

Flow unit ID.

Return type:

int

get_adj_list() → dict[str, list[str]][source]

Returns the connectivity of the nodes (node IDs) as an adjacency list.

Returns:

Adjacency list as a dictionary.

Return type:

dict[str, list[str]]

get_adj_matrix() → scipy.sparse.bsr_array[source]

Gets the adjacency matrix of this graph.

Returns:

Adjacency matrix as a sparse array.

Return type:

scipy.bsr_array

Gets all adjacent links/pipes of a given node.

Parameters:

node_id (str) – ID of the node.

Returns:

Adjacent links – i.e. (link ID, IDs of node end points).

Return type:

list[tuple[str, tuple[str, str]]]

get_all_junctions() → list[str][source]

Gets all junctions – i.e. nodes that are not tanks or reservoirs.

Returns:

List of all junctions.

Return type:

list[str]

Gets a list of all links/pipes (incl. their end points).

Returns:

List of links – (link ID, (left node ID, right node ID)).

Return type:

list[tuple[str, tuple[str, str]]]

get_all_nodes() → list[str][source]

Gets a list of all nodes.

Returns:

List of all nodes ID.

Return type:

list[str]

get_all_pairs_shortest_path(use_pipe_length_as_weight: bool = True) → dict[source]

Computes the shortest path between all pairs of nodes in this graph.

Parameters:

use_pipe_length_as_weight (bool, optional) –

If True, pipe lengths are used for the edge weights – otherwise, each edge weight is set to one.

The default is True.

Returns:

Shortest paths between all pairs of nodes as nested dictionaries – first key is the start node, second key is the end node.

Return type:

dict

get_all_pairs_shortest_path_length(use_pipe_length_as_weight: bool = True) → dict[source]

Computes the shortest path length between all pairs of nodes in this graph.

Parameters:

use_pipe_length_as_weight (bool, optional) –

If True, pipe lengths are used for the edge weights – otherwise, each edge weight is set to one.

The default is True.

Returns:

Shortest paths between all pairs of nodes as nested dictionaries – first key is the start node, second key is the end node.

Return type:

dict

get_all_pipes() → list[tuple[str, tuple[str, str]]][source]

Gets all pipes – i.e. links that not valves or pumps.

Returns:

List of all pipes – (link ID, (left node ID, right node ID)).

Return type:

list[tuple[str, tuple[str, str]]]

get_all_pumps() → list[str][source]

Gets the IDs of all pumps.

Returns:

Pump IDs.

Return type:

list[str]

get_all_reservoirs() → list[str][source]

Gets all reservoirs – i.e. nodes that are not junctions or tanks.

Returns:

List of all reservoirs.

Return type:

list[str]

get_all_tanks() → list[str][source]

Gets all tanks – i.e. nodes that are not junctions or reservoirs.

Returns:

List of all tanks.

Return type:

list[str]

get_all_valves() → list[str][source]

Gets the IDs of all valves.

Returns:

Valve IDs.

Return type:

list[str]

get_attributes() → dict[source]

Gets all attributes to be serialized – these attributes are passed to the constructor when the object is deserialized.

Returns:

Dictionary of attributes – i.e. pairs of attribute name + value.

Return type:

dict

Gets all information (e.g. diameter, length, etc.) associated with a given link.

Note that links can be pipes, pumps, or valves.

Parameters:

link_id (str) – ID of the link.

Returns:

Information associated with the given link.

Return type:

dict

get_neighbors(node_id: str) → list[str][source]

Gets all neighboring nodes of a given node.

Parameters:

node_id (str) – ID of the node.

Returns:

IDs of neighboring nodes.

Return type:

list[str]

get_node_info(node_id: str) → dict[source]

Gets all information (e.g. elevation, type, etc.) associated with a given node.

Parameters:

node_id (str) – ID of the node.

Returns:

Information associated with the given node.

Return type:

dict

get_number_of_junctions() → int[source]

Returns the number of junctions.

Returns:

Number of junctions.

Return type:

int

Returns the number of links.

Returns:

Number of links.

Return type:

int

get_number_of_nodes() → int[source]

Returns the number of nodes.

Returns:

Number of nodes.

Return type:

int

get_number_of_pipes() → int[source]

Returns the number of pipes.

Returns:

Number of pipes.

Return type:

int

get_number_of_pumps() → int[source]

Returns the number of pumps.

Returns:

Number of pumps.

Return type:

int

get_number_of_reservoirs() → int[source]

Returns the number of reservoirs.

Returns:

Number of reservoirs.

Return type:

int

get_number_of_tanks() → int[source]

Returns the number of tanks.

Returns:

Number of tanks.

Return type:

int

get_number_of_valves() → int[source]

Returns the number of valves.

Returns:

Number of valves.

Return type:

int

get_pump_info(pump_id: str) → dict[source]

Gets all information associated with a given pump.

Parameters:

pump_id (str) – ID of the pump.

Returns:

Pump information.

Return type:

dict

get_shortest_path(start_node_id: str, end_node_id: str, use_pipe_length_as_weight: bool = True) → list[str][source]

Computes the shortest path between two nodes in this graph.

Parameters:

  • start_node_id (str) – ID of start node.

  • end_node_id (str) – ID of end node.

  • use_pipe_length_as_weight (bool, optional) –

    If True, pipe lengths are used for the edge weights – otherwise, each edge weight is set to one.

    The default is True.

get_shortest_path_length(start_node_id: str, end_node_id: str, use_pipe_length_as_weight: bool = True) → list[str][source]

Computes the shortest path length between two nodes in this graph.

Parameters:

  • start_node_id (str) – ID of start node.

  • end_node_id (str) – ID of end node.

  • use_pipe_length_as_weight (bool, optional) –

    If True, pipe lengths are used for the edge weights – otherwise, each edge weight is set to one.

    The default is True.

get_valve_info(valve_id: str) → dict[source]

Gets all information associated with a given valve.

Parameters:

valve_id (str) – ID of the valve.

Returns:

Valve information.

Return type:

dict

property patterns: dict[str, list[float]]

Returns all time patterns – i.e., ID and list of multipliers.

Returns:

All time patterns.

Return type:

dict[str, list[float]]

property pressure_units: int

Returns the pressure unit ID.

Will be one of the following EPANET constants:

  • EN_PSI = 0 (Pounds per square inch)

  • EN_KPA = 1 (Kilopascals)

  • EN_METERS = 2 (Meters)

  • EN_BAR = 3 (Bar)

  • EN_FEET = 4 (Feet)

Returns:

Pressure unit ID.

Return type:

int

property pumps: dict

Gets all pumps – i.e. ID and associated information such as the pump type.

Returns:

All pumps and their associated information.

Return type:

dict

to_gis(coord_reference_system: str | None = None, pumps_as_points: bool = False, valves_as_points: bool = False) → dict[source]

Gets the network topology as a dictionary of geopandas.GeoDataFrames instances – i.e. each quantity (nodes, links/pipes, valves, etc.) is represented by a geopandas.GeoDataFrames instance.

Parameters:

  • coord_reference_system (str, optional) –

    Coordinate reference system.

    The default is None.

  • pumps_as_points (bool, optional) –

    If True, pumps are represented by points, otherwise by lines.

    The default is False.

  • valves_as_points (bool, optional) –

    If True, valves are represented by points, otherwise by lines.

    The default is False.

Returns:

Network topology as a dictionary of geopandas.GeoDataFrames instances. If a quantity does not exist, the data frame will be None.

Return type:

dict

to_inp_file(inp_file_out: str) → None[source]

Creates an .inp file with the network layout and parameters as specified in this instance. Note that no control rules are set!

Parameters:

inp_file_out (str) – Path to the .inp file.

property valves: dict

Gets all valves – i.e. ID and associated information such as the valve type.

Returns:

All valves and their associated information.

Return type:

dict

epyt_flow.topology.unitscategoryid_to_str(unit_category_id: int) → str[source]

Converts a given units category ID to the corresponding description.

Parameters:

unit_category_id (int) –

ID of the units category.

Must be one of the following constants:

  • UNITS_USCUSTOM = 0

  • UNITS_SIMETRIC = 1

Returns:

Units category description.

Return type:

str

epyt_flow.utils

Module provides helper functions.

epyt_flow.utils.areaunit_to_id(unit_desc: str) → int[source]

Converts a given area units string to the corresponding ID.

Parameters:

unit_desc (str) – Area units string.

Returns:

Corresponding area unit ID.

Return type:

int

epyt_flow.utils.areaunit_to_str(unit_id: int) → str[source]

Converts a given area measurement unit ID to the corresponding description.

Parameters:

unit_id (int) –

ID of the area unit.

Must be one of the following constants:

  • AREA_UNIT_FT2 = 1

  • AREA_UNIT_M2 = 2

  • AREA_UNIT_CM2 = 3

Returns:

Area unit description.

Return type:

str

epyt_flow.utils.create_path_if_not_exist(path_in: str) → None[source]

Creates a directory and all its parent directories if they do not already exist.

Parameters:

path_in (str) – Path to be created.

epyt_flow.utils.download_from_gdrive_if_necessary(download_path: str, url: str, verbose: bool = True) → None[source]

Downloads a file from a google drive repository if it does not already exist in a given path.

Note that if the path (folder) does not already exist, it will be created.

Parameters:

  • download_path (str) – Local path to the file – if this path does not exist, the file will be downloaded from the provided ‘url’ and stored in ‘download_dir’.

  • url (str) – Web-URL of the google drive repository.

  • verbose (bool, optional) –

    If True, a progress bar is shown while downloading the file.

    The default is True.

epyt_flow.utils.download_if_necessary(download_path: str, url: str, verbose: bool = True, backup_urls: list[str] = [], timeout: int = 30) → None[source]

Downloads a file from a given URL if it does not already exist in a given path. This function is deprecated, please use robust_download instead.

Note that if the path (folder) does not already exist, it will be created.

Parameters:

  • download_path (str) – Local path to the file – if this path does not exist, the file will be downloaded from the provided ‘url’ and stored in ‘download_dir’.

  • url (str) – Web-URL.

  • verbose (bool, optional) –

    If True, a progress bar is shown while downloading the file.

    The default is True.

  • backup_urls (list[str], optional) –

    List of alternative URLs that are being tried in the case that downloading from ‘url’ fails.

    The default is an empty list.

  • timeout (int, optional) –

    Allowed download inactivity in seconds. After this time passed without an update, the donwload is considered failed.

    The default is 30 seconds.

epyt_flow.utils.flowunit_to_str(unit_id: int) → str[source]

Converts a given flow unit ID to the corresponding description.

Parameters:

unit_id (int) –

Flow unit ID.

Must be one of the following EPANET constants:

  • EN_CFS = 0 (cubic foot/sec)

  • EN_GPM = 1 (gal/min)

  • EN_MGD = 2 (Million gal/day)

  • EN_IMGD = 3 (Imperial MGD)

  • EN_AFD = 4 (ac-foot/day)

  • EN_LPS = 5 (liter/sec)

  • EN_LPM = 6 (liter/min)

  • EN_MLD = 7 (Megaliter/day)

  • EN_CMH = 8 (cubic meter/hr)

  • EN_CMD = 9 (cubic meter/day)

  • EN_CMS = 10 (cubic meter/sec)

Returns:

Flow unit description.

Return type:

str

epyt_flow.utils.get_temp_folder() → str[source]

Gets a path to a temporary folder – i.e. a folder for storing temporary files.

Returns:

Path to a temporary folder.

Return type:

str

epyt_flow.utils.is_flowunit_simetric(unit_id: int) → bool[source]

Checks if a given flow unit belongs to SI metric units.

Parameters:

unit_id (int) –

ID of the flow unit.

Must be one of the following EPANET constants:

  • EN_CFS = 0 (cubic foot/sec)

  • EN_GPM = 1 (gal/min)

  • EN_MGD = 2 (Million gal/day)

  • EN_IMGD = 3 (Imperial MGD)

  • EN_AFD = 4 (ac-foot/day)

  • EN_LPS = 5 (liter/sec)

  • EN_LPM = 6 (liter/min)

  • EN_MLD = 7 (Megaliter/day)

  • EN_CMH = 8 (cubic meter/hr)

  • EN_CMD = 9 (cubic meter/day)

  • EN_CMS = 10 (cubic meter/sec)

Returns:

True if the fiven unit is a SI metric unit, False otherwise.

Return type:

bool

epyt_flow.utils.level_to_volume(tank_level: float, tank_diameter: float) → float[source]

Computes the volume of water in a tank given the water level in the tank.

Parameters:

  • tank_level (float) – Water level in the tank.

  • tank_diameter (float) – Diameter of the tank.

Returns:

Water volume in tank.

Return type:

float

epyt_flow.utils.massunit_to_id(unit_desc: str) → int[source]

Converts a given mass units string to the corresponding ID.

Parameters:

unit_desc (str) – Mass units string.

Returns:

Corresponding mass unit ID.

Return type:

int

epyt_flow.utils.massunit_to_str(unit_id: int) → str[source]

Converts a given mass unit ID to the corresponding description.

Parameters:

unit_id (int) –

ID of the mass unit.

Must be one of the following constant:

  • MASS_UNIT_MG = 4

  • MASS_UNIT_UG = 5

  • MASS_UNIT_MOL = 6

  • MASS_UNIT_MMOL = 7

  • MASS_UNIT_CUSTOM = 9

Returns:

Mass unit description.

Return type:

str

epyt_flow.utils.pack_zip_archive(f_in: list[str], f_out: str) → None[source]

Compresses a given list of files into a .zip archive.

Parameters:

  • f_in (list[str]) – List of files to be compressed into the .zip archive.

  • f_out (str) – Path to the final .zip file.

epyt_flow.utils.plot_timeseries_data(data: numpy.ndarray, labels: list[str] | None = None, x_axis_label: str | None = None, y_axis_label: str | None = None, y_ticks: tuple[list[float], list[str]] | None = None, show: bool = True, save_to_file: str | None = None, ax: matplotlib.axes.Axes | None = None) → matplotlib.axes.Axes[source]

Plots a single or multiple time series.

Parameters:

  • data (numpy.ndarray) – Time series data – each row in data corresponds to a complete time series.

  • labels (list[str], optional) –

    Labels for each time series in data. If None, no labels are shown.

    The default is None.

  • x_axis_label (str, optional) –

    X axis label.

    The default is None.

  • y_axis_label (str, optional) –

    Y axis label.

    The default is None.

  • y_ticks ((list[float], list[str]), optional) –

    Tuple of ticks (numbers) and labels (strings) for the y-axis.

    The default is None.

  • show (bool, optional) –

    If True, the plot/figure is shown in a window.

    Only considered when ‘ax’ is None.

    The default is True.

  • save_to_file (str, optional) –

    File to which the plot is saved.

    If specified, ‘show’ must be set to False – i.e. a plot can not be shown and saved to a file at the same time!

    The default is None.

  • ax (matplotlib.axes.Axes, optional) –

    If not None, ‘ax’ is used for plotting.

    The default is None.

Returns:

Plot.

Return type:

matplotlib.axes.Axes

epyt_flow.utils.plot_timeseries_prediction(y: numpy.ndarray, y_pred: numpy.ndarray, confidence_interval: numpy.ndarray | None = None, x_axis_label: str | None = None, y_axis_label: str | None = None, y_ticks: tuple[list[float], list[str]] | None = None, show: bool = True, save_to_file: str | None = None, ax: matplotlib.axes.Axes | None = None) → matplotlib.axes.Axes[source]

Plots the prediction (e.g. forecast) of single time series together with the ground truth time series. In addition, confidence intervals can be plotted as well.

Parameters:

  • y – Ground truth values.

  • y_pred – Predicted values.

  • confidence_interval –

    Confidence interval (upper and lower value) for each prediction in y_pred. If not None, the confidence interval is plotted as well.

    The default is None.

  • x_axis_label (str, optional) –

    X axis label.

    The default is None.

  • y_axis_label (str, optional) –

    Y axis label.

    The default is None.

  • y_ticks ((list[float], list[str]), optional) –

    Tuple of ticks (numbers) and labels (strings) for the y-axis.

    The default is None.

  • show (bool, optional) –

    If True, the plot/figure is shown in a window.

    Only considered when ‘ax’ is None.

    The default is True.

  • save_to_file (str, optional) –

    File to which the plot is saved.

    If specified, ‘show’ must be set to False – i.e. a plot can not be shown and saved to a file at the same time!

    The default is None.

  • ax –

    If not None, ‘axes’ is used for plotting.

    The default is None.

Returns:

Plot.

Return type:

matplotlib.axes.Axes

epyt_flow.utils.pressureunit_to_str(unit_id: int) → str[source]

Converts a given pressure unit ID to the corresponding description.

Parameters:

unit_id (int) –

Pressure unit ID.

Must be one of the following EPANET constants:

  • EN_PSI = 0 (Pounds per square inch)

  • EN_KPA = 1 (Kilopascals)

  • EN_METERS = 2 (Meters)

  • EN_BAR = 3 (Bar)

  • EN_FEET = 4 (Feet)

Returns:

Pressure unit description.

Return type:

str

epyt_flow.utils.qualityunit_to_id(unit_desc: str) → int[source]

Converts a given measurement unit description to the corresponding mass unit ID.

Parameters:

unit_desc (str) – Mass unit.

Returns:

Mass unit ID.

Will be either None (if no water quality analysis was set up) or one of the following constants:

  • MASS_UNIT_MG = 4 (mg/L)

  • MASS_UNIT_UG = 5 (ug/L)

  • TIME_UNIT_HRS = 8 (hrs)

Return type:

int

epyt_flow.utils.qualityunit_to_str(unit_id: int) → str[source]

Converts a given quality measurement unit ID to the corresponding description.

Parameters:

unit_id (int) –

ID of the quality unit.

Must be one of the following constants:

  • MASS_UNIT_MG = 4 (mg/L)

  • MASS_UNIT_UG = 5 (ug/L)

  • TIME_UNIT_HRS = 8 (hrs)

Returns:

Mass unit description.

Return type:

str

epyt_flow.utils.robust_download(download_path: str, urls: str | list, verbose: bool = True, timeout: int = 30) → None[source]

Downloads a file from the given urls if it does not already exist in the given path. The urls are tried in order. If a download stops or stalls, the next url is tried until one succeeds or all urls have failed.

Parameters:

  • download_path (str) – Local path to the file – if this path does not exist, the file will be downloaded from the provided ‘urls’ and stored there.

  • urls (list or str) – One url or a list of urls (where additional urls function as backup) to download the file from.

  • verbose (bool, optional) –

    If True, a progress bar is shown while downloading the file.

    The default is True.

  • timeout (int) –

    If this time passed without progress while downloading, the download is considered failed.

    The default is 30 seconds.

epyt_flow.utils.time_points_to_one_hot_encoding(time_points: list[int], total_length: int) → list[int][source]

Converts a list of time points into a one-hot-encoding.

Parameters:

  • time_points (list[int]) – Time points to be one-hot-encoded.

  • total_length (int) – Length of final one-hot-encoding.

Returns:

One-hot-encoded time points.

Return type:

list[int]

epyt_flow.utils.to_seconds(days: int | None = None, hours: int | None = None, minutes: int | None = None) → int[source]

Converts a timestamp (i.e. days, hours, minutes) into seconds.

Parameters:

  • days (int, optional) – Days.

  • hours (int, optional) – Hours.

  • minutes (int, optional) – Minutes.

Returns:

Timestamp in seconds.

Return type:

int

epyt_flow.utils.unpack_zip_archive(f_in: str, folder_out: str) → None[source]

Unpacks a .zip archive.

Parameters:

  • f_in (str) – Path to the .zip file.

  • folder_out (str) – Path to the folder where the unpacked files will be stored.

epyt_flow.utils.volume_to_level(tank_volume: float, tank_diameter: float) → float[source]

Computes the water level in a tank containing a given volume of water.

Parameters:

  • tank_volume (float) – Water volume in the tank.

  • tank_diameter (float) – Diameter of the tank.

Returns:

Water level in tank.

Return type:

float