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)

Bases: BaseModel

path: str

Relative path to the scenario

interpolation_method: FillMethod | None

Method of interpolation

scale_factors: dict[str, float] | None

Scale factors for each columns

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

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

abstract get_scenario_state(n_steps: int, columns: list[str] | None = None) → dict[str, ndarray]

Get scenario values for the current time step.

Parameters:

• n_steps – Current time step.

• columns – Optional list of column names to return. If None, returns all columns.

Returns:

Dictionary with scenario data for the requested columns.

abstract get_scenario_state_with_duration(n_step: int, duration: int, columns: list[str] | None = None) → dict[str, ndarray]

Get scenario values for the interval [n_step, n_step+duration].

Parameters:

• n_step – Current time step.

• duration – Additional amount of steps in interval.

• columns – Optional list of column names to return. If None, returns all columns.

Returns:

Dictionary with scenario data for the requested columns.

class eta_ctrl.timeseries.scenario_manager.CsvScenarioManager(scenario_configs: list[ConfigCsvScenario], start_time: datetime, end_time: datetime, total_time: float, resample_time: float = 0.0, seed: int | None = None)

Bases: ScenarioManager

ScenarioManager class for loading scenario data from CSV files.

load_data() → None

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

get_scenario_state(n_steps: int, columns: list[str] | None = None) → dict[str, ndarray]

Get scenario values for the current time step.

Parameters:

• n_steps – Current time step.

• columns – Optional list of column names to return. If None, returns all columns.

Returns:

Dictionary with scenario data for the requested columns.

get_scenario_state_with_duration(n_step: int, duration: int, columns: list[str] | None = None) → dict[str, ndarray]

Get scenario values for the interval [n_step, n_step+duration].

Parameters:

• n_step – Current time step.

• duration – Additional amount of steps in interval.

• columns – Optional list of column names to return. If None, returns all columns.

Returns:

Dictionary with scenario data for the requested columns.