agentlib.modules.simulation package

Submodules

agentlib.modules.simulation.csv_data_source module

class agentlib.modules.simulation.csv_data_source.CSVDataSource(config: dict, agent: Agent)[source]

Bases: BaseModule

backwards_iterator()[source]

Iterator for backwards extrapolation

create_iterator()[source]

Create a custom iterator based on the extrapolation method

process()[source]

Write the current data values into data_broker every t_sample

register_callbacks()[source]

Don’t do anything as this module is not event-triggered

pydantic model agentlib.modules.simulation.csv_data_source.CSVDataSourceConfig[source]

Bases: BaseModuleConfig

Config:

  • arbitrary_types_allowed: bool = True

  • validate_assignment: bool = True

  • extra: str = forbid

  • frozen: bool = True

Fields:
Validators:
field data: DataFrame | Path = Empty DataFrame Columns: [] Index: []

Data that should be communicated during execution. Index should be either numeric or Datetime, numeric values are interpreted as seconds.

Validated by:
field data_offset: Timedelta | float | None = 0

Offset will be subtracted from index, allowing you to start at any point in your data. I.e. if your environment starts at 0, and you want your data-source to start at 1000 seconds, you should set this to 1000.

Validated by:
field extrapolation: Literal['constant', 'repeat', 'backwards'] = 'constant'

Determines what to do, when the data source runs out. ‘constant’ returns the last value, ‘repeat’ repeats the data from the start, and ‘backwards’ goes through the data backwards, bouncing indefinitely.

Validated by:
field outputs: List[AgentVariable] = []

Optional list of columns of data frame that should be sent. If ommited, all datapoint in frame are sent.

Validated by:
field shared_variable_fields: List[str] = ['outputs']
Validated by:
field t_sample: float | int = 1

Sampling time. Data source sends an interpolated value from the data every <t_sample> seconds. Default is 1 s.

Constraints:

  • ge = 0

Validated by:
validator check_data  »  data[source]

Makes sure data is a data frame, and loads it if required.

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialise private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

  • self – The BaseModel instance.

  • context – The context.

transform_to_numeric_index(data: DataFrame) → DataFrame[source]

Handles the index and ensures it is numeric, with correct offset

validator validate_data  »  all fields[source]

Checks if outputs and data columns match, and ensures a numeric index.

agentlib.modules.simulation.simulator module

Module contains the Simulator, used to simulate any model.

class agentlib.modules.simulation.simulator.Simulator(*, config: dict, agent: Agent)[source]

Bases: BaseModule

The Simulator is the interface between simulation models and further other implementations. It contains all interface functions for interacting with the standard model class.

do_step()[source]

Generator function to perform a simulation step, update inputs, outputs and model results.

In a simulation step following happens: 1. Update inputs (only necessary if self.update_inputs_on_callback = False) 2. Specify the end time of the simulation from the agents perspective. Important note: The agents use unix-time as a timestamp and start the simulation with the current datetime (represented by self.env.time), the model starts at 0 seconds (represented by self.env.now). 3. Directly after the simulation we send the updated output values to other modules and agents by setting them the data_broker. Even though the environment time is not already at the end time specified above, we explicitly add the timestamp to the variables. This way other agents and communication has the maximum time possible to process the outputs and send input signals to the simulation. 4. Call the timeout in the environment, hence actually increase the environment time.

get_results() → DataFrame | None[source]

Return the current results.

Returns:

The results DataFrame.

Return type:

pd.DataFrame

property model: Model

Getter for current simulation model

Returns:

Current simulation model

Return type:

agentlib.core.model.Model

process()[source]

This function creates a endless loop for the single simulation step event. The do_step() function needs to return a generator.

register_callbacks()[source]
run(until=None)[source]

Runs the simulator in stand-alone mode if needed Attention: If the environment is connected to another environment all scheduled process will be started in this environment.

terminate()[source]

Terminate the model

update_model_inputs()[source]

Internal method to write current data_broker to simulation model. Only update values, not other module_types.

update_module_vars()[source]

Method to write current model output and states values to the module outputs and states.

pydantic model agentlib.modules.simulation.simulator.SimulatorConfig[source]

Bases: BaseModuleConfig

Pydantic data model for simulator configuration parser

Config:

  • arbitrary_types_allowed: bool = True

  • validate_assignment: bool = True

  • extra: str = forbid

  • frozen: bool = True

Fields:
Validators:
field inputs: List[AgentVariable] = []
field measurement_uncertainty: Dict[str, float] | float = 0

Either pass a float and add the percentage uncertainty to all measurements from the model.Or pass a Dict and specify the model variable name as keyand the associated uncertainty as a float

field model: Dict [Required]
Validated by:
field outputs: List[AgentVariable] = []
field overwrite_result_file: bool = False

If True, and the result file already exists, the file is overwritten.

field parameters: List[AgentVariable] = []
field result_causalities: List[Causality] = [<Causality.input: 'input'>, <Causality.output: 'output'>]

List of causalities to store. Default stores only inputs and outputs

field result_filename: str | None = None

If not None, results are stored in that filename.Needs to be a .csv file

Validated by:
field result_sep: str = ','

Separator in the .csv file. Only relevant if result_filename is passed

field save_results: bool = False

If True, results are created and stored

field shared_variable_fields: List[str] = ['outputs']
Validated by:
field states: List[AgentVariable] = []
field t_sample: float | int = 1

Simulation sample time

Constraints:

  • ge = 0

Validated by:
field t_start: float | int = 0.0

Simulation start time

Constraints:

  • ge = 0

field t_stop: float | int = inf

Simulation stop time

Constraints:

  • ge = 0

Validated by:
field update_inputs_on_callback: bool = True

If True, model inputs are updated if they are updated in data_broker.Else, the model inputs are updated before each simulation.

field validate_incoming_values: bool | None = False

If true, the validator of the AgentVariable value is called when receiving a new value from the DataBroker. In the simulator, this is False by default, as we expect to receive a lot of measurements and want to be efficient.

field write_results_delay: float | None = None

Sampling interval for which the results are written to disc in seconds.

Constraints:

  • gt = 0

Validated by:
validator check_model  »  model[source]

Validate the model input

validator check_nonexisting_csv  »  result_filename[source]

Check if the result_filename is a .csv file or an hf and assert that it does not exist.

validator check_t_sample  »  t_sample[source]

Check if t_sample is smaller than stop-start time

validator check_t_stop  »  t_stop[source]

Check if stop is greater than start time

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialise private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

  • self – The BaseModel instance.

  • context – The context.

validator set_default_t_sample  »  write_results_delay[source]
class agentlib.modules.simulation.simulator.SimulatorResults(variables: List[ModelVariable])[source]

Bases: object

Class to organize in-memory simulator results.

columns: MultiIndex
data: List[List[float]]
df() → DataFrame[source]

Returns the current results as a dataframe.

index: List[float]
initialize(time: float)[source]

Adds the first row to the data

write_results(file: str)[source]

Dumps results which are currently in memory to a file. On creation of the file, the header columns are dumped, as well.

agentlib.modules.simulation.simulator.convert_agent_vars_to_list_of_dicts(var: List[AgentVariable]) → List[Dict][source]

Function to convert AgentVariables to a list of dictionaries containing information for ModelVariables.

agentlib.modules.simulation.simulator.read_simulator_results(file: str)[source]

Reads results from file with correct multi-column format.