epyt_control.signal_processing.state_forecasting

epyt_control.signal_processing.state_forecasting.surrogates

This module contains different state transition surrogate models.

class epyt_control.signal_processing.state_forecasting.surrogates.HydraulicStateTransitionSurrogate(wdn_topology: NetworkTopology, n_actuators: int, state_transition_model: StateTransitionModel)[source]

Bases: StateTransitionSurrogate

Surrogate for the hydraulic state transition function.

Paramaters

wdn_topologyepyt_flow.topology.NetworkTopology

Information about the topology of the WDN.

n_actuatorsint

Dimensionality of the control signal.

state_transition_modelStateTransitionModel

State transition model which is used as an approximation of the true state transition function. Usually, a neural network is used.

fit_to_scada(scada_data: ScadaData, control_actions: ndarray | None = None) → None[source]

Fits the state transition surrogate to given SCADA data.

Parameters:
predict(cur_pressure: ndarray, cur_flow: ndarray, next_demand: ndarray, control_actions: ndarray) → ndarray[source]

Predcts how the current state evolves for the next time step.

cur_pressurenumpy.ndarray

Current pressure at every node.

cur_flownumpy.ndarray

Current flow rate at every link.

next_demandnumpy.ndarray

Demand at every node for the next time step.

control_actionsnumpy.ndarray

Control signal at the current time step.

Returns:

Next state.

Return type:

numpy.ndarray

class epyt_control.signal_processing.state_forecasting.surrogates.StateTransitionModel[source]

Bases: object

Abstract base class of state transition models used in a surrogte – i.e. a deep neural network approximating the state transition functions.

abstract fit(cur_state: ndarray, next_time_varying_quantity: ndarray, next_state: ndarray) → None[source]

Fits the neural network to given state transition data.

Parameters:
abstract init(wdn_topology: NetworkTopology, input_size: int, state_size: int) → None[source]

Initializes the model.

Parameters:

  • wdn_topology – Information about the topology of the WDN.

  • input_size (int) – Dimensionality of the input – i.e. current state + time varying inputs that are relevant for the state transition (incl. control inputs).

  • state_size (int) – Dimensionality of the state to be predicted.

abstract partial_fit(cur_state: ndarray, next_time_varying_quantity: ndarray, next_state: ndarray) → None[source]

Performs a partial fit of the state transition surrogate to given data.

Parameters:

  • cur_state – Current state of the system.

  • next_time_varying_quantity – Time varying events (incl. control signals) that are relevant for evolving the state.

  • next_state – Next state – is to be predicted based on the other two arguments.

abstract predict(cur_state: ndarray, next_time_varying_quantity: ndarray) → ndarray[source]

Predicts the next state based on the current state and time varying events such as control signals.

Parameters:

  • cur_state – Current state.

  • next_time_varying_quantity – Time varying events (incl. control signals) that are relevant for evolving the state.

Returns:

Next state.

Return type:

numpy.ndarray

class epyt_control.signal_processing.state_forecasting.surrogates.StateTransitionSurrogate(wdn_topology: NetworkTopology, n_actuators: int)[source]

Bases: object

Base class of state transition surrogates.

Parameters:

  • wdn_topology – Information about the topology of the WDN.

  • n_actuators (int) – Number of actuators – i.e. control inputs.

fit_to_env(env: RlEnv, n_max_iter: int | None = None, policy: Callable[[ndarray], ndarray] | None = None) → None[source]

Fits the state transition surrogate to a given control environment.

Parameters:

  • env (RlEnv) – Control environment.

  • n_max_iter (int) – Maximum numbe of iterations used for data collection. Note that data collection stops if the environment terminates.

  • policy (Callable[[numpy.ndarray], numpy.ndarray]) –

    A policy for mapping observations to actions (i.e. control signals) – will be applied at each time step. If None, random actions are sampled from the action space.

    The default is None.

abstract fit_to_scada(scada_data: ScadaData, control_actions: ndarray) → None[source]

Fits the state transition surrogate to given SCADA data.

Parameters:

  • scada_data – SCADA data.

  • control_actions – Control signals at every time step.

class epyt_control.signal_processing.state_forecasting.surrogates.WaterQualityStateTransitionSurrogate(wdn_topology: NetworkTopology, n_actuators: int, state_transition_model: StateTransitionModel)[source]

Bases: StateTransitionSurrogate

Surrogate for the quality (e.g. water age, chemical concentration) state transition function.

Paramaters

wdn_topologyepyt_flow.topology.NetworkTopology

Information about the topology of the WDN.

n_actuatorsint

Dimensionality of the control signal.

state_transition_modelStateTransitionModel

State transition model which is used as an approximation of the true state transition function. Usually, a neural network is used.

fit_to_scada(scada_data: ScadaData, control_actions: ndarray | None = None) → None[source]

Fits the state transition surrogate to given SCADA data.

Parameters:

  • scada_data – SCADA data.

  • control_actions – Control signals at every time step.

predict(cur_node_quality: ndarray, cur_link_quality: ndarray, next_flow: ndarray, control_actions: ndarray) → ndarray[source]

Predicts the next state (i.e. quality everywhere) based on the current state of the system, the next flow, and control signals.

Parameters:

  • cur_flow – Current flow rate at each link.

  • cur_node_quality – Current quality at every node.

  • cur_link_quality – Current quality at every link.

  • next_demand – Demand at every node for the next time step.

  • control_actions – Control signal at the current time step.

Returns:

Next state.

Return type:

numpy.ndarray

epyt_control.signal_processing.state_forecasting.mlp_surrogate

This module contains Multi Layer Perceptrons (MLP), also known as feedforward artifical neural networks, and deep neural networks (DNNs) for estimating the state transition function.

class epyt_control.signal_processing.state_forecasting.mlp_surrogate.DnnStateTransitionModel(hidden_layers_size: list[int] = [128], activation: str = 'tanh', last_layer_activation: str | None = None, max_iter: int = 200, device: str = 'cpu', normalization_layer: bool = True, normalize_input_output: bool = False, dropout: float = 0.0, batch_size: int = 128, **kwds)[source]

Bases: StateTransitionModel

Neural network state transition model. Implemented in PyTorch.

Parameters:

  • hidden_layers_size (list[int], optional) –

    Dimensionality of the hidden layers.

    The default is [128].

  • activation (str, optional) –

    Activation function for the hidden layers.

    The default is ‘tanh’

  • last_layer_activation (str, optional) –

    Activation function of the last layer. If None, no acitvation function will be applied in the last layer.

    The default is None.

  • max_iter (int, optional) –

    Maximum number of training itertions.

    The default is 200.

  • device (str, optional) –

    Device used for the computation.

    The default is ‘cpu’

  • normalization_layer (bool, optional) –

    If True, the first layer is a normalization layer.

    The default is True.

  • normalize_input_output (bool, optional) –

    If True, input is scaled and the target as well – i.e. the scaled state is predicted. Can not be used in conjunction with ‘normalization_layer’.

    The default is false.

  • dropout (float, optional) –

    Specifies the dropout probability of in the input layer. If 0, no dropout will be used.

    The default is 0.

  • batch_size (int, optional) –

    Batch size for training. Be aware that the batch size might have an influence on the normalization layer.

    The default is 128

compute_jacobian(cur_state: ndarray, next_time_varying_quantity: ndarray) → ndarray[source]

Computes the Jacobian w.r.t. a given state (incl. control signals).

Parameters:
Returns:

numpy.ndarray <https – Jacobian.

Return type:

//numpy.org/doc/stable/reference/generated/numpy.ndarray.html>`_

fit(cur_state: ndarray, next_time_varying_quantity: ndarray, next_state: ndarray) → None[source]

Fits the neural network to given state transition data.

Parameters:
init(wdn_topology: NetworkTopology, input_size: int, state_size: int) → None[source]

Initializes the model.

Parameters:

  • wdn_topology – Information about the topology of the WDN.

  • input_size (int) – Dimensionality of the input – i.e. current state + time varying inputs that are relevant for the state transition (incl. control inputs).

  • state_size (int) – Dimensionality of the state to be predicted.

invert_output_scaling(Y_pred: ndarray) → ndarray[source]

Inverts the scaling of the output (i.e. predicted state).

Parameters:

Y_pred (numpy.ndarray <https://numpy.org/doc/stable/reference/generated/numpy.ndarray.html>`_) – Predicted state.

Returns:

numpy.ndarray <https – Unscaled predicted state.

Return type:

//numpy.org/doc/stable/reference/generated/numpy.ndarray.html>`_

load_from_file(file_path: str) → None[source]

Loads model’s weights and the standard scaler from a given file.

Parameters:

file_path (str) – File path.

partial_fit(cur_state: ndarray, next_time_varying_quantity: ndarray, next_state: ndarray) → None[source]

Performs a partial fit of the state transition surrogate to given data.

Parameters:

  • cur_state – Current state of the system.

  • next_time_varying_quantity – Time varying events (incl. control signals) that are relevant for evolving the state.

  • next_state – Next state – is to be predicted based on the other two arguments.

predict(cur_state: ndarray, next_time_varying_quantity: ndarray, invert_output_scaling: bool = False) → ndarray[source]

Predicts the next state based on the current state and time varying events such as control signals.

Parameters:

  • cur_state – Current state.

  • next_time_varying_quantity – Time varying events (incl. control signals) that are relevant for evolving the state.

Returns:

Next state.

Return type:

numpy.ndarray

save_to_file(file_path: str) → None[source]

Saves model’s weights and the standard scaler to a file.

Parameters:

file_path (str) – File path.

class epyt_control.signal_processing.state_forecasting.mlp_surrogate.SimpleMlpStateTransitionModel(hidden_layer_sizes: list[int] = [128], activation: str = 'tanh', max_iter: int = 500, normalize: bool = True, **kwds)[source]

Bases: StateTransitionModel

Multi-layer perceptron state transition model. Implemented in scikit-learn.

Parameters:

  • hidden_layers_size (list[int], optional) –

    Dimensionality of the hidden layers.

    The default is [128].

  • activation (str, optional) –

    Activation function for the hidden layers.

    The default is ‘tanh’

  • max_iter (int, optional) –

    Maximum number of training itertions.

    The default is 500.

fit(cur_state: ndarray, next_time_varying_quantity: ndarray, next_state: ndarray) → None[source]

Fits the neural network to given state transition data.

Parameters:
init(wdn_topology: NetworkTopology, input_size: int, state_size: int) → None[source]

Initializes the model.

Parameters:

  • wdn_topology – Information about the topology of the WDN.

  • input_size (int) – Dimensionality of the input – i.e. current state + time varying inputs that are relevant for the state transition (incl. control inputs).

  • state_size (int) – Dimensionality of the state to be predicted.

partial_fit(cur_state: ndarray, next_time_varying_quantity: ndarray, next_state: ndarray) → None[source]

Performs a partial fit of the state transition surrogate to given data.

Parameters:

  • cur_state – Current state of the system.

  • next_time_varying_quantity – Time varying events (incl. control signals) that are relevant for evolving the state.

  • next_state – Next state – is to be predicted based on the other two arguments.

predict(cur_state: ndarray, next_time_varying_quantity: ndarray) → ndarray[source]

Predicts the next state based on the current state and time varying events such as control signals.

Parameters:

  • cur_state – Current state.

  • next_time_varying_quantity – Time varying events (incl. control signals) that are relevant for evolving the state.

Returns:

Next state.

Return type:

numpy.ndarray