Scenario Loaders

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.

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

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.

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

Scenario Loaders

Contents

Scenario Loaders#

Submodules#

CSV Loader#

OEDS Loader#

AMIRIS Loader#

PyPSA Loader#