eta_ctrl.timeseries.scenario_manager module

class eta_ctrl.timeseries.scenario_manager.ConfigCsvScenario(*, path: str, interpolation_method: Literal['ffill', 'bfill', 'interpolate', 'asfreq'] | None = None, scale_factors: dict[str, float] | None = None, prefix: str | None = None, infer_datetime_cols: Literal['string', 'dates'] | tuple[int, int] = 'dates', time_conversion_str: str = '%Y-%m-%d %H:%M', rename_cols: dict[str, str] | None = None, scenarios_path: Path | None = None)

Bases: BaseModel

model_config = {'extra': 'forbid', 'frozen': True, 'use_attribute_docstrings': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

path: str

Relative path to the scenario.

interpolation_method: FillMethod | None

Pandas method to use for filling missing data [“ffill”, “bfill”, “interpolate”, “asfreq”].

scale_factors: dict[str, float] | None

Scale factors for each column.

prefix: str | None

Prefix for all column names.

infer_datetime_cols: InferDatetimeType | tuple[int, int]

Setting how the datetime values should be converted.

When set to string it uses the format from time_conversion_str, when set to ‘dates’ it will use pandas to determine the datetime. If a two-tuple (row, col) is given, data from the specified field in the data files will be used to determine the date format.

time_conversion_str: str

Time conversion string used when infer_datetime_cols is set to ‘string’.

Should specify the format for Python strptime.

rename_cols: dict[str, str] | None

Dictionary for renaming column names.

Note

The column names are stripped of illegal characters and underscores are added in place of spaces. “Water Temperature #2 [°C]” becomes “Water_Temperature_2_C”. If you want to rename the column, you need to specify the processed name, for example: {“Water_Temperature_2_C”: “T_W”}.

scenarios_path: Path | None

Directory for the scenarios. Not included in config declaration, passed by main Config object.

model_post_init(_: Any) → None

Ensure that the CSV file exists.

Raises:

FileNotFoundError – If file does not exist.

property abs_path: Path

Absolute file path of the scenario.

class eta_ctrl.timeseries.scenario_manager.ScenarioManager

Bases: ABC

compute_episode_offset(rng: np.random.Generator) → int

Compute the row offset into the scenario data for the next episode.

Returns 0 by default (no random slicing). Override in subclasses that support random time slicing.

Parameters:

rng – Random number generator from the environment.

Returns:

Integer row offset into the scenario data.

get_scenario_state_var(n_step: int, state_var: StateVar) → np.ndarray

Get scenario values for a single state variable at the given (absolute) step.

Parameters:

• n_step – Absolute row index into the scenario data (env step + episode offset).

• state_var – State variable configuration.

Returns:

Array of scenario values.

class eta_ctrl.timeseries.scenario_manager.CsvScenarioManager(scenario_configs: list[ConfigCsvScenario], start_time: datetime, end_time: datetime, total_time: float, resample_time: float, use_random_time_slice: bool = False)

Bases: ScenarioManager

ScenarioManager class for loading scenario data from CSV files.

scenario_configs: list[ConfigCsvScenario]

compute_episode_offset(rng: np.random.Generator) → int

Compute the row offset into the scenario dataframe for the next episode.

Parameters:

rng – Random number generator used to pick a random starting position.

Returns:

Integer row index into self.scenarios representing the episode start.

load_data() → None

Load scenario data by calling ‘scenario_from_csv’ with the ConfigCsvScenario objects