Scenario Loaders#

Submodules#

assume.scenario.loader_csv module#

assume.scenario.loader_csv.add_units(units_df: DataFrame, unit_type: str, world: World, forecaster: Forecaster) None#

Add units to the world from a given dataframe. The callback is used to adjust unit_params depending on the unit_type, before adding the unit to the world.

Parameters:
  • units_df (pandas.DataFrame) – The dataframe containing the units.

  • unit_type (str) – The type of the unit.

  • world (World) – The world to which the units will be added.

  • forecaster (Forecaster) – The forecaster used for adding the units.

async assume.scenario.loader_csv.async_load_custom_units(world: World, inputs_path: str, scenario: str, file_name: str, unit_type: str) None#

Load custom units from a given path.

This function loads custom units of a specified type from a given path within a scenario, adding them to the world environment for simulation.

Parameters:
  • world (World) – An instance of the World class representing the simulation environment.

  • inputs_path (str) – The path to the folder containing input files necessary for the custom units.

  • scenario (str) – The name of the scenario from which the custom units are to be loaded.

  • file_name (str) – The name of the file containing the custom units.

  • unit_type (str) – The type of the custom units to be loaded.

assume.scenario.loader_csv.convert_to_rrule_freq(string: str) tuple[int, int]#

Convert a string to a rrule frequency and interval.

Parameters:

string (str) – The string to be converted. Should be in the format of “1h” or “1d” or “1w”.

Returns:

The rrule frequency and interval.

Return type:

tuple[int, int]

assume.scenario.loader_csv.load_custom_units(world: World, inputs_path: str, scenario: str, file_name: str, unit_type: str) None#

Load custom units from a given path.

This function loads custom units of a specified type from a given path within a scenario, adding them to the world environment for simulation.

Parameters:
  • world (World) – An instance of the World class representing the simulation environment.

  • inputs_path (str) – The path to the folder containing input files necessary for the custom units.

  • scenario (str) – The name of the scenario from which the custom units are to be loaded.

  • file_name (str) – The name of the file containing the custom units.

  • unit_type (str) – The type of the custom units to be loaded.

Example

>>> load_custom_units(
    world=world,
    inputs_path="/path/to/inputs",
    scenario="scenario_name",
    file_name="custom_units.csv",
    unit_type="custom_type"
)

Notes

  • The function loads custom units from the specified file within the given scenario and adds them to the world environment for simulation.

  • If the specified custom units file is not found, a warning is logged.

  • Each unique unit operator in the custom units is added to the world’s unit operators.

  • The custom units are added to the world environment based on their type for use in simulations.

assume.scenario.loader_csv.load_file(path: str, config: dict, file_name: str, index: DatetimeIndex | None = None) DataFrame#

Loads a csv file from the given path and returns a dataframe.

The config file is used to check if the file name is specified in the config file, otherwise defaults to the file name.

If the index is specified, the dataframe is resampled to the index, if possible. If not, None is returned.

Parameters:
  • path (str) – The path to the csv file.

  • config (dict) – The config file containing file mappings.

  • file_name (str) – The name of the csv file.

  • index (pd.DatetimeIndex, optional) – The index of the dataframe. Defaults to None.

Returns:

The dataframe containing the loaded data.

Return type:

pandas.DataFrame

Raises:

FileNotFoundError – If the specified file is not found, returns None.

assume.scenario.loader_csv.load_scenario_folder(world: World, inputs_path: str, scenario: str, study_case: str, perform_learning: bool = True, perform_evaluation: bool = False, episode: int = 1, eval_episode: int = 1)#

Load a scenario from a given path.

This function loads a scenario within a specified study case from a given path, setting up the world environment for simulation and learning.

Parameters:
  • world (World) – An instance of the World class representing the simulation environment.

  • inputs_path (str) – The path to the folder containing input files necessary for the scenario.

  • scenario (str) – The name of the scenario to be loaded.

  • study_case (str) – The specific study case within the scenario to be loaded.

  • perform_learning (bool, optional) – A flag indicating whether learning should be performed. Defaults to True.

  • perform_evaluation (bool, optional) – A flag indicating whether evaluation should be performed. Defaults to False.

  • episode (int, optional) – The episode number for learning. Defaults to 0.

  • eval_episode (int, optional) – The episode number for evaluation. Defaults to 0.

Raises:

ValueError – If the specified scenario or study case is not found in the provided inputs.

Example

>>> load_scenario_folder(
    world=world,
    inputs_path="/path/to/inputs",
    scenario="scenario_name",
    study_case="study_case_name",
    perform_learning=True,
    perform_evaluation=False,
    episode=1,
    eval_episode=1,
    trained_policies_save_path="",
)

Notes

  • The function sets up the world environment based on the provided inputs and configuration files.

  • If perform_learning is set to True and learning_mode is set, the function initializes the learning mode with the specified episode number.

  • If perform_evaluation is set to True, the function performs evaluation using the specified evaluation episode number.

  • The function utilizes the specified inputs to configure the simulation environment, including market parameters, unit operators, and forecasting data.

  • After calling this function, the world environment is prepared for further simulation and analysis.

async assume.scenario.loader_csv.load_scenario_folder_async(world: World, inputs_path: str, scenario: str, study_case: str, perform_learning: bool = True, perform_evaluation: bool = False, episode: int = 0, eval_episode: int = 0) None#

Load a scenario from a given path.

This function loads a scenario within a specified study case from a given path, setting up the world environment for simulation and learning.

Parameters:
  • world (World) – An instance of the World class representing the simulation environment.

  • inputs_path (str) – The path to the folder containing input files necessary for the scenario.

  • scenario (str) – The name of the scenario to be loaded.

  • study_case (str) – The specific study case within the scenario to be loaded.

  • perform_learning (bool, optional) – A flag indicating whether learning should be performed. Defaults to True.

  • perform_evaluation (bool, optional) – A flag indicating whether evaluation should be performed. Defaults to False.

  • episode (int, optional) – The episode number for learning. Defaults to 0.

  • eval_episode (int, optional) – The episode number for evaluation. Defaults to 0.

Raises:

ValueError – If the specified scenario or study case is not found in the provided inputs.

assume.scenario.loader_csv.make_market_config(id: str, market_params: dict, world_start: datetime, world_end: datetime) MarketConfig#

Create a market config from a given dictionary.

Args: id (str): The id of the market. market_params (dict): The market parameters. world_start (datetime.datetime): The start time of the world. world_end (datetime.datetime): The end time of the world.

Returns: MarketConfig: The market config.

assume.scenario.loader_csv.replace_paths(config: dict, inputs_path: str)#

This function replaces all config items which end with “_path” to one starting with the given inputs_path. So that paths in the config are relative to the inputs_path where the config is read from.

Parameters:
  • config (dict) – the config dict read from yaml

  • inputs_path (str) – the base path from the config

Returns:

the adjusted config dict

Return type:

dict

assume.scenario.loader_csv.run_learning(world: World, inputs_path: str, scenario: str, study_case: str) None#

Train Deep Reinforcement Learning (DRL) agents to act in a simulated market environment.

This function runs multiple episodes of simulation to train DRL agents, performs evaluation, and saves the best runs. It maintains the buffer and learned agents in memory to avoid resetting them with each new run.

Parameters:
  • world (World) – An instance of the World class representing the simulation environment.

  • inputs_path (str) – The path to the folder containing input files necessary for the simulation.

  • scenario (str) – The name of the scenario for the simulation.

  • study_case (str) – The specific study case for the simulation.

Note

  • The function uses a ReplayBuffer to store experiences for training the DRL agents.

  • It iterates through training episodes, updating the agents and evaluating their performance at regular intervals.

  • Initial exploration is active at the beginning and is disabled after a certain number of episodes to improve the performance of DRL algorithms.

  • Upon completion of training, the function performs an evaluation run using the best policy learned during training.

  • The best policies are chosen based on the average reward obtained during the evaluation runs, and they are saved for future use.

assume.scenario.loader_oeds module#

assume.scenario.loader_oeds.load_oeds(world: World, scenario: str, study_case: str, infra_uri: str, marketdesign: list[assume.common.market_objects.MarketConfig], nuts_config: list[str] = [])#

Load a scenario from a given path.

Parameters:
  • world (World) – the world to add the oeds scenario to

  • scenario (str) – the scenario name of the simulation

  • study_case (str) – the study case name of the simulation

  • infra_uri (str) – database uri for the infrastructure

  • marketdesign (list[MarketConfig]) – the list of marketconfigs, which form the market design

  • nuts_config (list[str], optional) – The list of NUTS areas which are loaded in the scenario. Defaults to [].

async assume.scenario.loader_oeds.load_oeds_async(world: World, scenario: str, study_case: str, infra_uri: str, marketdesign: list[assume.common.market_objects.MarketConfig], nuts_config: list[str] = [])#

This initializes a scenario using the open-energy-data-server NOWUM/open-energy-data-server

Scenarios can use any NUTS area in Germany and use any year with appropriate weather available

Parameters:
  • world (World) – the world to add this scenario to

  • scenario (str) – scenario name

  • study_case (str) – study case name

  • infra_uri (str) – database uri to connect to the OEDS

  • marketdesign (list[MarketConfig]) – description of the market design which will be used with the scenario

  • nuts_config (list[str], optional) – list of NUTS areas from which the simulation data is taken. Defaults to [].

assume.scenario.loader_amiris module#

assume.scenario.loader_amiris.add_agent_to_world(agent: dict, world: World, prices: dict, contracts: list, base_path: str, markups: dict = {})#
assume.scenario.loader_amiris.get_matching_send_one_or_multi(agent_id: int, contract: dict)#
assume.scenario.loader_amiris.get_send_receive_msgs_per_id(agent_id, contracts_config: list)#
assume.scenario.loader_amiris.interpolate_blocksizes(installed_power: float, block_size_in_mw: float, min_eff: float, max_eff: float, min_markup: float, max_markup: float)#

This method interpolates efficiencies and markups for a given powerplant The fist powerplant receives the highest markup and lowest efficiency. The last one has lowest markup and highest efficiency.

async assume.scenario.loader_amiris.load_amiris_async(world: World, scenario: str, study_case: str, base_path: str)#
assume.scenario.loader_amiris.read_amiris_yaml(base_path)#
assume.scenario.loader_amiris.read_csv(base_path, filename)#