desmod.simulation

Simulation model with batteries included.

class desmod.simulation.SimEnvironment(config: Dict[str, Any])

Simulation Environment.

The SimEnvironment class is a simpy.Environment subclass that adds some useful features:

• Access to the configuration dictionary (config).
• Access to a seeded pseudo-random number generator (rand).
• Access to the simulation timescale (timescale).
• Access to the simulation duration (duration).

Some models may need to share additional state with all its desmod.component.Component instances. SimEnvironment may be subclassed to add additional members to achieve this sharing.

Parameters:config (dict) – A fully-initialized configuration dictionary.

config

The configuration dictionary.

rand

The pseudo-random number generator; an instance of random.Random.

timescale

Simulation timescale (magnitude, units) tuple. The current simulation time is now * timescale.

duration

The intended simulation duration, in units of timescale.

tracemgr

TraceManager instance.

now

The current simulation time.

time(t=None, unit='s')

The current simulation time scaled to specified unit.

Parameters:

• t (float) – Time in simulation units. Default is now.
• unit (str) – Unit of time to scale to. Default is ‘s’ (seconds).

Returns:

Simulation time scaled to to unit.

active_process

The currently active process of the environment.

process(generator)

Process an event yielding generator.

A generator (also known as a coroutine) can suspend its execution by yielding an event. Process will take care of resuming the generator with the value of that event once it has happened. The exception of failed events is thrown into the generator.

Process itself is an event, too. It is triggered, once the generator returns or raises an exception. The value of the process is the return value of the generator or the exception, respectively.

Processes can be interrupted during their execution by interrupt().

timeout(delay, value)

A Event that gets triggered after a delay has passed.

This event is automatically triggered when it is created.

event()

An event that may happen at some point in time.

An event

• may happen (triggered is False),
• is going to happen (triggered is True) or
• has happened (processed is True).

Every event is bound to an environment env and is initially not triggered. Events are scheduled for processing by the environment after they are triggered by either succeed(), fail() or trigger(). These methods also set the ok flag and the value of the event.

An event has a list of callbacks. A callback can be any callable. Once an event gets processed, all callbacks will be invoked with the event as the single argument. Callbacks can check if the event was successful by examining ok and do further processing with the value it has produced.

Failed events are never silently ignored and will raise an exception upon being processed. If a callback handles an exception, it must set defused to True to prevent this.

This class also implements __and__() (&) and __or__() (|). If you concatenate two events using one of these operators, a Condition event is generated that lets you wait for both or one of them.

all_of(events)

A Condition event that is triggered if all of a list of events have been successfully triggered. Fails immediately if any of events failed.

any_of(events)

A Condition event that is triggered if any of a list of events has been successfully triggered. Fails immediately if any of events failed.

schedule(event: simpy.events.Event, priority: NewType.<locals>.new_type = 1, delay: Union[int, float] = 0) → None

Schedule an event with a given priority and a delay.

peek() → Union[int, float]

Get the time of the next scheduled event. Return Infinity if there is no further event.

step() → None

Process the next event.

Raise an EmptySchedule if no further events are available.

desmod.simulation.simulate(config: Dict[str, Any], top_type: Type[Component], env_type: Type[desmod.simulation.SimEnvironment] = <class 'desmod.simulation.SimEnvironment'>, reraise: bool = True, progress_manager=<function standalone_progress_manager>) → Dict[str, Any]

Initialize, elaborate, and run a simulation.

All exceptions are caught by simulate() so they can be logged and captured in the result file. By default, any unhandled exception caught by simulate() will be re-raised. Setting reraise to False prevents exceptions from propagating to the caller. Instead, the returned result dict will indicate if an exception occurred via the ‘sim.exception’ item.

Parameters:

• config (dict) – Configuration dictionary for the simulation.
• top_type – The model’s top-level Component subclass.
• env_type – SimEnvironment subclass.
• reraise (bool) – Should unhandled exceptions propogate to the caller.

Returns:

Dictionary containing the model-specific results of the simulation.

desmod.simulation.simulate_factors(base_config: Dict[str, Any], factors: List[Tuple[List[str], List[Any]]], top_type: Type[Component], env_type: Type[desmod.simulation.SimEnvironment] = <class 'desmod.simulation.SimEnvironment'>, jobs: Optional[int] = None, config_filter: Optional[Callable[[Dict[str, Any]], bool]] = None) → List[Dict[str, Any]]

Run multi-factor simulations in separate processes.

The factors are used to compose specialized config dictionaries for the simulations.

The multiprocessing module is used run each simulation with a separate Python process. This allows multi-factor simulations to run in parallel on all available CPU cores.

Parameters:

• base_config (dict) – Base configuration dictionary to be specialized.
• factors (list) – List of factors.
• top_type – The model’s top-level Component subclass.
• env_type – SimEnvironment subclass.
• jobs (int) – User specified number of concurent processes.
• config_filter (function) – A function which will be passed a config and returns a bool to filter.

Returns:

Sequence of result dictionaries for each simulation.