mosaik.util — Utility classes and functions

This module contains some utility functions and classes.

connect_many_to_one(world: ~mosaik.scenario.World | ~mosaik.async_scenario.AsyncWorld, src_set: ~typing.Iterable[~mosaik.async_scenario.Entity], dest: ~mosaik.async_scenario.Entity, *attrs: str | ~typing.Tuple[str, str], async_requests: bool = False, transform: ~typing.Callable[[~typing.Any], ~typing.Any] = <function <lambda>>)

connect each entity in src_set to dest.

See the connect for more details.

Parameters:

• world (World | AsyncWorld)

• src_set (Iterable[Entity])

• dest (Entity)

• attrs (str | Tuple[str, str])

• async_requests (bool)

• transform (Callable[[Any], Any])

connect_randomly(world: World | AsyncWorld, src_set: MutableSequence[Entity], dest_set: MutableSequence[Entity], *attrs: str | Tuple[str, str], evenly: bool = True, max_connects: int = inf)

Randomly connect the entities from src_set to the entities from dest_set and return a subset of dest_set containing all entities with a connection.

Parameters:

• world (World | AsyncWorld) – the instance of the World to which the entities belong.

• src_set (MutableSequence[Entity]) – a MutableSequence (potentially empty) containing Entity instances. Each of these entities will be connected to an entity of dest_set.

• dest_set (MutableSequence[Entity]) – a non-empty MutableSequence of Entity instances. Not every of these entities is necessarily connected (if src_set contains too few entities)

• evenly (bool) –

  How to distribute the entities:

  If True, entity connections will be distributed as evenly as possible. That means if you connect a set of three entities to a set of three entities, there will be three 1:1 connections; if you connect four entities to three entities, there will be one 2:1 and two 1:1 connections.

  If False, connections will be truly random. That means if you connect three entities to three entities, you may either have three 1:1 connections, one 2:1 and two 1:1 connections or just one 3:1 connection.

• max_connects (int) – the maximum number of connections that an entity of dest_set may receive. This argument is only taken into account if evenly is set to False.

• attrs (str | Tuple[str, str])

Params attrs:

the attribute names to connect as in connect.

Returns:

The list of entities from dest_set to which entities from src_set were actually connected.

connect_zip(world: World | AsyncWorld, src_set: Collection[Entity], dest_set: Collection[Entity], *attrs: str | Tuple[str, str], **kwargs) → None

Connect entities in parallel. This works analogously to the built-in zip function: Each entity in src_set is connected to the entity in dest_set at the corresponding index.

Parameters:

• world (World | AsyncWorld) – the world for this simulation

• src_set (Collection[Entity]) – the collection of source entities

• dest_set (Collection[Entity]) – the collection of destination entities

• attrs (str | Tuple[str, str]) – the attributes to connect, as in world.connect

• kwargs – the connection kwargs as in world.connect

Raises:

ValueError – if src_set and dest_set don’t have the same number of elements

Return type:

None

plot_execution_time(world: World, folder: str = 'figures', hdf5path: str | None = None, dpi: int = 600, format: Literal['png', 'pdf', 'svg'] = 'png', show_plot: bool = True, slice: Tuple[int, int] | None = None)

Creates an image visualizing the execution time of the different simulators of a mosaik scenario.

Parameters:

• world (World) – mosaik world object

• folder (str) – folder to store the image (only if no hdf5path is provided)

• hdf5path (str | None) – Path to HDF5 file, which will be used as path for the created image

• dpi (int) – DPI for created images

• format (Literal['png', 'pdf', 'svg']) – format for created image

• show_plot (bool) – whether to open a window to show the plot

• slice (Tuple[int, int] | None) – reduce the timeframe that you show in the plot. Usage as in Python list slicing, i.e., negative values are possible to start from the end of the list. Jumps are not possible. slice needs to be a two-element integer list, e.g. (0, 5).

Returns:

None but image file will be written to file system

plot_dataflow_graph(world: World, folder: str = 'figures', hdf5path: str | None = None, dpi: int = 600, format: Literal['png', 'pdf', 'svg'] = 'png', show_plot: bool = True)

Creates an image visualizing the data flow graph of a mosaik scenario. Using the spring layout from Matplotlib (Fruchterman- Reingold force-directed algorithm) to position the nodes.

Parameters:

• world (World) – mosaik world object

• folder (str) – folder to store the image (only if no hdf5path is provided)

• hdf5path (str | None) – Path to HDF5 file, which will be used as path for the created image

• dpi (int) – DPI for created images

• format (Literal['png', 'pdf', 'svg']) – format for created image

• show_plot (bool) – whether open a window to show the plot

Returns:

None but image file will be written to file system

quadratic_bezier(p0: Tuple[float, float], p1: Tuple[float, float], p2: Tuple[float, float], num: int = 20)

Return the curve points for a quadratic Bézier segment.

Parameters:

• p0 (Tuple[float, float])

• p1 (Tuple[float, float])

• p2 (Tuple[float, float])

• num (int)

plot_df_graph_groups(world: World | AsyncWorld, show_plot: bool = False, *, html_folder: str | None = None, max_layout_tries: int = 25, accept_incorrectly_placed_simulators: bool = False) → go.Figure

Return a Plotly figure of the dataflow graph with group overlays.

The layout is retried until no simulator is placed inside an unrelated group. Set accept_incorrectly_placed_simulators to True to keep the last attempt even if misplacements remain after max_layout_tries.

Parameters:

• world (World | AsyncWorld)

• show_plot (bool)

• html_folder (str | None)

• max_layout_tries (int)

• accept_incorrectly_placed_simulators (bool)

Return type:

go.Figure

plot_execution_graph(world: World, title: str = '', folder: str = 'figures', hdf5path: str | None = None, dpi: int = 600, format: Literal['png', 'pdf', 'svg'] = 'png', show_plot: bool = True, save_plot: bool = True, slice: Tuple[int, int] | None = None)

Creates an image visualizing the execution graph of a mosaik scenario.

Parameters:

• world (World) – mosaik world object

• title (str) – the title of the graph

• folder (str) – folder to store the image (only if no hdf5path is provided)

• hdf5path (str | None) – Path to HDF5 file, which will be used as path for the created image

• dpi (int) – DPI for created images

• format (Literal['png', 'pdf', 'svg']) – format for created image

• show_plot (bool) – whether to open a window to show the plot

• slice (Tuple[int, int] | None) – reduce the timeframe that you show in the plot. Usage as in Python list slicing, i.e., negative values are possible to start from the end of the list. Jumps are not possible. slice needs to be a two-element integer tuple, e.g. (0, 5).

• save_plot (bool)

Returns:

None but image file will be written to file system

plot_execution_time_per_simulator(world: World, folder: str = 'figures', hdf5path: str | None = None, dpi: int = 600, format: Literal['png', 'pdf', 'svg'] = 'png', show_plot: bool = True, plot_per_simulator: bool = False, slice: Tuple[int, int] | None = None)

Creates images visualizing the execution time of each of the different simulators of a mosaik scenario.

Parameters:

• world (World) – mosaik world object

• folder (str) – folder to store the image (only if no hdf5path is provided)

• hdf5path (str | None) – Path to HDF5 file, which will be used as path for the created image

• dpi (int) – DPI for created images

• format (Literal['png', 'pdf', 'svg']) – format for created image

• show_plot (bool) – whether to open a window to show the plot

• plot_per_simulator (bool) – whether to create a separated plot per simulator. This is especially useful if the step sizes of the simulators are very different.

• slice (Tuple[int, int] | None) – reduce the timeframe that you show in the plot. Usage as in Python list slicing, i.e., negative values are possible to start from the end of the list. Jumps are not possible. slice needs to be a two-element integer tuple, e.g. (0, 5).

Returns:

None but image file will be written to file system

plot_dataflow(world: World, file_name: str | None = None, dpi: int = 600, format: Literal['png', 'pdf', 'svg'] = 'png', show_plot: bool = True, return_figure: bool = True, seed: int | None = None, **kwargs: Any) → None | Tuple[Figure, Axes]

Creates an image visualizing the data flow graph of a mosaik scenario. Using the spring layout from Matplotlib (Fruchterman- Reingold force-directed algorithm) to position the nodes.

Parameters:

• world (World) – mosaik world object

• file_name (str | None) – a full file name including a folder to store the image

• dpi (int) – DPI for created images

• format (Literal['png', 'pdf', 'svg']) – format for created image

• show_plot (bool) – whether open a window to show the plot

• return_figure (bool) – return figure and axis

• seed (int | None) – needed to fix graph layout

• **kwargs (Any) –

  extra parameters will be passed to fig.savefig()

Returns:

None but image file will be written to file name if given. It returns tuple with figure and axis instead if return_figure is True

Return type:

None | Tuple[Figure, Axes]