Facility Model¶

class morl4water.core.models.facility.ControlledFacility(name: str, observation_space: ~gymnasium.spaces.space.Space, action_space: ~gymnasium.core.ActType, max_capacity: float, max_action: float, objective_function=<function Objective.no_objective>, objective_name: str = '')¶

Bases: ABC

determine_info() → dict¶: Returns information about the reservoir.

abstractmethod determine_observation() → ObsType¶

abstractmethod determine_outflow() → float¶

abstractmethod determine_reward() → float¶

get_inflow(timestep: int) → float¶

get_outflow(timestep: int) → float¶

abstractmethod is_terminated() → bool¶

is_truncated() → bool¶

reset() → None¶

set_inflow(timestep: int, inflow: float) → None¶

step(action: ActType) → tuple[ObsType, SupportsFloat, bool, bool, dict]¶

class morl4water.core.models.facility.Facility(name: str, objective_function=<function Objective.no_objective>, objective_name: str = '', normalize_objective=0.0)¶

Bases: ABC

Abstract base class representing a facility with inflow and outflow management, along with reward determination.

name¶

Identifier for the facility.

Type:: str

all_inflow¶

Historical inflow values recorded over time.

Type:: list[float]

all_outflow¶

Historical outflow values recorded over time.

Type:: list[float]

objective_function¶

Function to evaluate the facility’s performance based on defined objectives.

Type:: Callable

objective_name¶

Name of the objective function used.

Type:: str

current_date¶

Date associated with the current timestep of the facility.

Type:: Optional[datetime]

timestep_size¶

Size of the timestep for simulation. Usually 1 month.

Type:: Optional[relativedelta]

timestep¶

Current timestep index for the facility simulation.

Type:: int

split_release¶

Placeholder for managing release strategies (usage to be defined).

Type:: Optional

normalize_objective¶

Normalization factor for the objective reward.

Type:: float

abstractmethod determine_consumption() → float¶

Abstract method to calculate the facility’s water consumption.

Returns:: The calculated water consumption for the current timestep.
Return type:: float
Raises:: NotImplementedError – If this method is not implemented in a subclass.

determine_info() → dict¶

Method to gather information about the facility’s state.

Returns:: A dictionary containing information about the facility.
Return type:: dict
Raises:: NotImplementedError – If this method is not implemented in a subclass.

determine_outflow() → float¶

Calculates the outflow based on the current inflow and consumption.

Returns:: The calculated outflow for the current timestep.
Return type:: float

abstractmethod determine_reward() → float¶

Abstract method to compute the reward based on the facility’s performance.

Returns:: The computed reward for the current timestep.
Return type:: float
Raises:: NotImplementedError – If this method is not implemented in a subclass.

get_inflow(timestep: int) → float¶

Retrieves the inflow value for a specific timestep.

Parameters:: timestep (int) – The timestep index for which to retrieve the inflow.
Returns:: The inflow value for the specified timestep.
Return type:: float
Raises:: IndexError – If the timestep is out of bounds.

get_outflow(timestep: int) → float¶

Retrieves the outflow value for a specific timestep.

Parameters:: timestep (int) – The timestep index for which to retrieve the outflow.
Returns:: The outflow value for the specified timestep.
Return type:: float
Raises:: IndexError – If the timestep is out of bounds.

is_terminated() → bool¶

Checks if the facility simulation has reached a terminating condition.

Returns:: Always returns False for the base Facility class; override in subclasses as needed.
Return type:: bool

is_truncated() → bool¶

Checks if the facility simulation has been truncated.

Returns:: Always returns False for the base Facility class; override in subclasses as needed.
Return type:: bool

reset() → None¶

Resets the facility to its initial state for a new simulation run.

Returns:: None

set_inflow(timestep: int, inflow: float) → None¶

Sets the inflow value for a specific timestep, adjusting if necessary.

Parameters:

timestep (int) – The timestep index at which to set the inflow.
inflow (float) – The inflow value to set.

Raises:

IndexError – If the timestep index is invalid.

step() → tuple[ObsType, float, bool, bool, dict]¶

Advances the simulation by one timestep, calculating outflows and rewards.

Returns:

A tuple containing:

ObsType: Placeholder for observation type (to be defined).
float: The reward for the current timestep.
bool: Indicates if the simulation has terminated.
bool: Indicates if the simulation has been truncated.
dict: Additional information about the facility’s state.

Return type:

tuple[ObsType, float, bool, bool, dict]