Engine

Engine runs the simulation.

class vivarium.core.engine.Defer(defer: Any, f: Callable, args: Tuple)

Bases: object

Allows for delayed application of a function to an update.

The object simply holds the provided arguments until it’s time for the computation to be performed. Then, the function is called.

Parameters

• defer – An object with a .get_command_result() method whose output will be passed to the function. For example, the object could be an vivarium.core.process.Process object whose .get_command_result() method will return the process update.

• function – The function. For example, invert_topology() to transform the returned update.

• args – Passed as the second argument to the function.

get() → Dict[str, Any]

Perform the deferred computation.

Returns

The result of calling the function.

class vivarium.core.engine.EmptyDefer

Bases: vivarium.core.engine.Defer

get() → Dict[str, Any]

class vivarium.core.engine.Engine(composite: Optional[vivarium.core.composer.Composite] = None, processes: Optional[Dict[str, Any]] = None, steps: Optional[Dict[str, Any]] = None, flow: Optional[Dict[str, Sequence[Tuple[str, …]]]] = None, topology: Optional[Dict[str, Union[Tuple[str, …], dict, object]]] = None, store: Optional[vivarium.core.store.Store] = None, initial_state: Optional[Dict[str, Any]] = None, experiment_id: Optional[str] = None, experiment_name: Optional[str] = None, metadata: Optional[dict] = None, description: str = '', emitter: Union[str, dict] = 'timeseries', store_schema: Optional[dict] = None, emit_topology: bool = True, emit_processes: bool = False, emit_config: bool = False, emit_step: float = 1, display_info: bool = True, progress_bar: bool = False, global_time_precision: Optional[int] = None, profile: bool = False, initial_global_time: float = 0)

Bases: object

Defines simulations

Parameters

• composite – A Composite, which specifies the processes, steps, flow, and topology. This is an alternative to passing in processes and topology dict, which can not be loaded at the same time.

• processes – A dictionary that maps process names to process objects. You will usually get this from the processes key of the dictionary from vivarium.core.composer.Composer.generate().

• steps – A dictionary that maps step names to step objects. You will usually get this from the steps key of the dictionary from vivarium.core.composer.Composer.generate().

• flow – A dictionary that maps step names to sequences of paths to the steps that the step depends on. You will usually get this from the flow key of the dictionary from vivarium.core.composer.Composer.generate().

• topology – A dictionary that maps process names to sub-dictionaries. These sub-dictionaries map the process’s port names to tuples that specify a path through the tree from the compartment root to the store that will be passed to the process for that port.

• store – A pre-loaded Store. This is an alternative to passing in processes and topology dict, which can not be loaded at the same time. Note that if you provide this argument, you must ensure that all parallel processes (i.e. vivarium.core.process.Process objects with the parallel attribute set to True) are instances of vivarium.core.process.ParallelProcess. This constructor converts parallel processes to ParallelProcess objects automatically if you do not provide this store argument.

• initial_state – By default an empty dictionary, this is the initial state of the simulation.

• experiment_id – A unique identifier for the experiment. A UUID will be generated if none is provided.

• metadata – A dictionary with additional data about the experiment, which is saved by the emitter with the configuration.

• description – A description of the experiment. A blank string by default.

• emitter – An emitter configuration which must conform to the specification in the documentation for vivarium.core.emitter.get_emitter(). The experiment ID will be added to the dictionary you provide as the value for the key experiment_id.

• display_info – prints experiment info

• progress_bar – shows a progress bar

• global_time_precision – an optional int that sets the decimal precision of global_time. This is useful for remove floating- point rounding errors for the time keys of saved states.

• store_schema – An optional dictionary to expand the store hierarchy configuration, and also to turn emits on or off. The dictionary needs to be structured as a hierarchy, which will expand the existing store hierarchy. Setting an emit value for a branch node will set the emits of all the leaves to that value.

• emit_topology – If True, this will emit the topology with the configuration data.

• emit_processes – If True, this will emit the serialized processes with the configuration data.

• emit_config – If True, this will emit the serialized initial state with the configuration data.

• profile – Whether to profile the simulation with cProfile.

• initial_global_time – The initial time for the simulation. Useful when this Engine is part of a larger, older simulation.

apply_update(update: Dict[str, Any], state: vivarium.core.store.Store) → bool

Apply an update to the simulation state.

Parameters

• update – The update to apply. Must be relative to state.

• state – The store to which the update is relative (usually root of simulation state. We need this so to preserve the “perspective” from which the update was generated.

Returns

a bool indicating whether the topology_views expired.

end() → None

Terminate all processes running in parallel.

This MUST be called at the end of any simulation with parallel processes. This function also ends profiling and computes profiling stats, including stats from parallel sub-processes. These stats are stored in self.stats.

run_for(interval: float, force_complete: bool = False) → None

Run each process within the given interval and update their states.

run_for() gives the caller more control over the simulation loop than update(). In particular, it may be called repeatedly within a caller-managed simulation loop without forcing processes to complete after each call (as would be the case with update()). It is the responsibility of the caller to ensure that when run_for() is called for the last time in the caller-managed simulation loop, force_complete=True so that all the processes finish at the end of the simulation.

Parameters

• interval – the amount of time to simulate the composite.

• force_complete – a bool indicating whether to force processes to complete at the end of the interval.

run_steps() → None

Run all the steps in the simulation.

update(interval: float) → None

Run each process for the given interval and force them to complete at the end of the interval. See run_for for the keyword args.

vivarium.core.engine.empty_front(t: float) → Dict[str, Union[float, dict]]

vivarium.core.engine.invert_topology(update: Dict[str, Any], args: Tuple[Tuple[str, …], Dict[str, Union[Tuple[str, …], dict, object]]]) → Dict[str, Any]

Wrapper function around inverse_topology.

Wraps vivarium.library.topology.inverse_topology().

Updates are produced relative to the process that produced them. To transform them such that they are relative to the root of the simulation hierarchy, this function “inverts” a topology.

Parameters

• update – The update.

• args – Tuple of the path to which the update is relative and the topology.

Returns

The update, relative to the root of path.

vivarium.core.engine.pf(x: Any) → str

Format x for display.

vivarium.core.engine.pp(x: Any) → None

Print x in a pretty format.

vivarium.core.engine.print_progress_bar(iteration: float, total: float, decimals: float = 1, length: int = 50) → None

Create terminal progress bar

Parameters

• iteration – Current iteration

• total – Total iterations

• decimals – Positive number of decimals in percent complete

• length – Character length of bar

vivarium.core.engine.starts_with(a_list: Tuple[str, …], sub: Tuple[str, …]) → bool

Check whether one path is a prefix of another.

Parameters

• a_list – Path to look for prefix in.

• sub – Prefix.

Returns

True if sub is a prefix of a_list; False otherwise.

vivarium.core.engine.timestamp(dt: Optional[Any] = None) → str

Get a timestamp of the form YYYYMMDD.HHMMSS.

Parameters

dt – Datetime object to generate timestamp from. If not specified, the current time will be used.

Returns

Timestamp.