Scenario Loaders

Submodules

CSV Loader

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.

async assume.scenario.loader_csv.async_setup_world(world: World, scenario_data: dict[str, object], study_case: str, perform_evaluation: bool = False, terminate_learning: 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.

• scenario_data (dict) – A dictionary containing the configuration and loaded files for the scenario and study case.

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

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

• terminate_learning (bool, optional) – An automatically set flag indicating that we terminated the learning process now, either because we reach the end of the episode itteration or because we triggered an early stopping.

• 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.load_config_and_create_forecaster(inputs_path: str, scenario: str, study_case: str) → dict[str, object]

Load the configuration and files for a given scenario and study case. This function allows us to load the files and config only once when running multiple iterations of the same scenario.

Parameters:

• 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.

Returns:

dict[str, object]:: A dictionary containing the configuration and loaded files for the scenario and study case.

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_dsm_units(path: str, config: dict, file_name: str) → dict

Loads and processes a CSV file containing DSM unit data, where each unit may consist of multiple components (technologies) under the same plant name. The function groups data by plant name, processes each group to handle different technologies, and organizes the data into a structured DataFrame. It then splits the DataFrame based on unique unit_types.

Parameters:

• path (str) – The directory path where the CSV file is located.

• config (dict) – Configuration dictionary, potentially used for specifying additional options or behaviors (not used in the current implementation but provides flexibility for future enhancements).

• file_name (str) – The name of the CSV file to be loaded.

Returns:

A dictionary where each key is a unique unit_type and the value is a DataFrame containing

the corresponding DSM units of that type.

Return type:

dict

Notes

• The CSV file is expected to have columns such as ‘name’, ‘technology’, ‘unit_type’, and other operational parameters.

• The function assumes that the first non-null value in common and bidding columns is representative if multiple entries exist for the same plant.

• It is crucial that the input CSV file follows the expected structure for the function to process it correctly.

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_evaluation: bool = False, terminate_learning: 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_evaluation (bool, optional) – A flag indicating whether evaluation should be performed. Defaults to False.

• terminate_learning (bool, optional) – An automatically set flag indicating that we terminated the learning process now, either because we reach the end of the episode itteration or because we triggered an early stopping.

• 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_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_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.

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.read_grid(network_path: str | Path) → dict[str, pandas.core.frame.DataFrame]

assume.scenario.loader_csv.read_units(units_df: DataFrame, unit_type: str, forecaster: Forecaster, world_bidding_strategies: dict[str, assume.common.base.BaseStrategy]) → dict[str, list[dict]]

Read units from a dataframe and only add them to a dictionary. The dictionary contains the operator ids as keys and the list of units belonging to the operator as values.

Parameters:

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

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

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

• world_bidding_strategies (dict[str, BaseStrategy]) – The strategies available in the world

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, verbose: bool = False) → 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_csv.setup_world(world: World, scenario_data: dict[str, object], study_case: str, perform_evaluation: bool = False, terminate_learning: bool = False, episode: int = 0, eval_episode: int = 0) → None

OEDS Loader

async assume.scenario.loader_oeds.load_oeds_async(world: World, scenario: str, study_case: str, start: datetime, end: datetime, infra_uri: str, marketdesign: list[assume.common.market_objects.MarketConfig], bidding_strategies: dict[str, str], 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 [].

AMIRIS Loader

assume.scenario.loader_amiris.add_agent_to_world(agent: dict, world: World, prices: dict, contracts: list, base_path: str, markups: dict = {}, supports: dict = {})

Adds an agent from a amiris agent definition to the ASSUME world. It should be called in load_amiris_async, which loads agents in the correct order.

Parameters:

• agent (dict) – AMIRIS agent dict

• world (World) – ASSUME world to add the agent to

• prices (dict) – prices read from amiris scenario beforehand

• contracts (list) – contracts read from the amiris scenario beforehand

• base_path (str) – base path to load profile csv files from

• markups (dict, optional) – markups read from former agents. Defaults to {}.

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: int, contracts_config: list[dict])

AMIRIS contract conversion function which finds the list of ids which receive or send a message from/to the agent with agent_id.

Parameters:

• agent_id (int) – the agent id to which the contracts are found

• contracts_config (list[dict]) – whole contracts dict read from yaml

Returns:

A tuple containing the following:

• list: A list containing the ids of sending agents.

• list: A list containing the ids of receiving agents

Return type:

tuple

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)

Loads an Amiris scenario. Markups and markdowns are handled by linearly interpolating the agents volume. This mimicks the behavior of the way it is done in AMIRIS.

Parameters:

• world (World) – the ASSUME world

• scenario (str) – the scenario name

• study_case (str) – study case to define

• base_path (str) – base path from where to load the amrisi scenario

assume.scenario.loader_amiris.read_amiris_yaml(base_path)

assume.scenario.loader_amiris.read_csv(base_path, filename)

PyPSA Loader

async assume.scenario.loader_pypsa.load_pypsa_async(world: World, scenario: str, study_case: str, network: Network, marketdesign: list[assume.common.market_objects.MarketConfig], bidding_strategies: dict[str, str], save_frequency_hours: int = 4)

This initializes a scenario from the given pypsa grid. One can load a grid from pypsa import_from_csv_folder, adjust its properties and add it to this function to create an energy market scenario from it. This is also compatible with netCDF, PyPower, PandaPower and HDF5 pypsa-compatible datasets.

Parameters:

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

• scenario (str) – scenario name

• study_case (str) – study case name

• network (pypsa.Network) – pypsa Network from which the simulation properties and timeseries data is extracted

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