commonpower.control.observation_handling.ObservationHandler

class ObservationHandler(num_forecasts: int = 0, observation_types: List[ElementTypes] = [ElementTypes.DATA, ElementTypes.STATE], num_past_observations: int = 0)[source]

Bases: object

Base class for handling observations of RL agents. Extracts information from the underlying controlled system and processes it for RL observations.

Parameters:
  • num_forecasts (int) – The number of forecasted steps to include in the observations. Default is 0.

  • observation_types (List[ElementTypes]) – The types of elements to include in the observations. Default includes DATA and STATE.

  • num_past_observations (int) – The number of past observations to stack for the agent. Default is 0.

Methods

get_adjusted_obs

Adjusts the system observations by stacking past observations as needed.

get_obs_mask

Retrieves the current observation mask.

get_observation_space

Builds and retrieves the observation space for the RL controller based on the controlled nodes.

reset

Resets the ObservationHandler by clearing all stored past observations.

set_obs_mask

Sets the observation mask for the elements observed by the RL controller.

get_adjusted_obs(system_obs: dict) dict[source]

Adjusts the system observations by stacking past observations as needed.

Parameters:

system_obs (dict) – A dictionary containing the current system observations.

Returns:

dict

A dictionary containing the adjusted observations, including stacked past observations if

configured.

get_obs_mask() Tuple[dict, int][source]

Retrieves the current observation mask.

Returns:

Tuple[dict, int]

A tuple containing the observation mask (mapping of node IDs to observed elements)

and the number of forecasted steps.

get_observation_space(nodes_controller: List[None]) Dict[source]

Builds and retrieves the observation space for the RL controller based on the controlled nodes.

Parameters:

nodes_controller (List[None]) – A list of nodes controlled by the RL agent. Each node provides its observation space using the observation_space method.

Returns:

gym.spaces.Dict

A dictionary representing the observation space for the controller, including stacking

for past observations if configured.

reset() None[source]

Resets the ObservationHandler by clearing all stored past observations.

Returns:

None

set_obs_mask(nodes_controller: List[None]) None[source]

Sets the observation mask for the elements observed by the RL controller.

Parameters:
  • nodes_controller (List[None]) – A list of nodes representing the system components controlled by

  • agent. (the RL)

Returns:

None