nexosim.aio

Asyncio version of the simulation API.

This module defines an asynchronous version of the Simulation class.

Example usage

ClientServer

import asyncio
from nexosim.aio import Simulation
from nexosim.time import MonotonicTime, Duration
from nexosim.exceptions import SimulationNotStartedError

async def run():
    async with Simulation("0.0.0.0:41633") as sim:
        await sim.build()
        await sim.init()

        await sim.schedule_event(MonotonicTime(1), "input", 1)
        await sim.schedule_event(MonotonicTime(3), "input", 2)
        try:
            await sim.step_until(Duration(5))
        except SimulationNotStartedError:
            time = await sim.time()
            print(f"Simulation halted at {time}")
            print(await sim.try_read_events("output"))

async def halt():
    async with Simulation("0.0.0.0:41633") as sim:
        await asyncio.sleep(2)
        await sim.halt()

async def main():
    await asyncio.gather(run(), halt())

asyncio.run(main())

use std::error::Error;
use std::time::Duration;

use nexosim::time::{AutoSystemClock, PeriodicTicker};
use serde::{Deserialize, Serialize};

use nexosim::model::Model;
use nexosim::ports::{EventSource, Output, SinkState, event_queue_endpoint};
use nexosim::simulation::{Mailbox, SimInit};
use nexosim::{Message, server};

#[derive(Clone, Message, Serialize, Deserialize)]
pub(crate) struct OutputEvent {
    pub(crate) foo: u16,
    pub(crate) bar: String,
}

#[derive(Default, Serialize, Deserialize)]
pub(crate) struct MyModel {
    pub(crate) output: Output<OutputEvent>,
}

#[Model]
impl MyModel {
    pub async fn my_input(&mut self, value: u16) {
        let event = OutputEvent {
            foo: value,
            bar: String::from("string"),
        };
        self.output.send(event).await;
    }
}

fn bench(_cfg: ()) -> Result<SimInit, Box<dyn Error>> {
    let mut model = MyModel::default();
    let mbox = Mailbox::new();

    let mut bench = SimInit::new();

    let sink = event_queue_endpoint(&mut bench, SinkState::Enabled, "output")?;
    model.output.connect_sink(sink);

    EventSource::new()
        .connect(MyModel::my_input, &mbox)
        .bind_endpoint(&mut bench, "input")?;

    // Assembly.
    Ok(bench.add_model(model, mbox, "model").with_clock(
        AutoSystemClock::new(),
        PeriodicTicker::new(Duration::from_millis(10)),
    ))
}

fn main() {
    server::run(bench, "0.0.0.0:41633".parse().unwrap()).unwrap();
}

Simulation

An asynchronous handle to the remote simulation bench.

Creates a handle to the remote simulation bench running at the specified address.

A gRPC NeXosim server must be running at the specified address.

For a regular remote gRPC connection via HTTP/2, the address should omit the URL scheme and the double-slash prefix (e.g. localhost:41633).

For a local Unix Domain Socket connection, the address is the socket path prefixed with the unix: scheme (e.g. unix:relative/path/to/socket, unix:/absolute/path/to/socket or alternatively unix:///absolute/path/to/socket).

Simulation is an asynchronous context manager. If not used in a async with statement, the aclose() method should be called after use.

Note

While most requests are processed concurrently, step* and process* requests are mutually blocking.

Parameters:

Name	Type	Description	Default
address	str	the address at which the NeXosim server is running.	required

aclose() async

Closes the gRPC channel.

build(cfg=None) async

Builds a simulation bench.

Parameters:

Name	Type	Description	Default
cfg	Any	A bench configuration object which can be serialized/deserialized to the expected bench configuration type. The None default is appropriate if the bench builder expects the Rust type (), accepts an Option::None.	None
Raises		exceptions.SimulationError: One of the exceptions derived from SimulationError may be raised, such as: - [`BenchPanicError`][nexosim.exceptions.BenchPanicError] - [`BenchError`][nexosim.exceptions.BenchError] - [`DuplicateEventSourceError`][nexosim.exceptions.DuplicateEventSourceError] - [`DuplicateQuerySourceError`][nexosim.exceptions.DuplicateQuerySourceError] - [`DuplicateEventSinkError`][nexosim.exceptions.DuplicateEventSinkError] - [`InvalidBenchConfigError`][nexosim.exceptions.InvalidBenchConfigError]	required

init(time=MonotonicTime.EPOCH) async

Initializes a simulation bench.

The simulation must be built using the Simulation.build method before it can be initialized.

If a simulation bench is already running, it is replaced by the newly initialized bench. In such case, events that have not yet been retrieved from the sinks will be lost and the sinks are reset to their default enabled/disabled state.

Parameters:

Name	Type	Description	Default
time	MonotonicTime	The time at which the simulation will be initialized.	EPOCH

Raises:

Type	Description
SimulationError	One of the exceptions derived from SimulationError may be raised, such as: SimulationOutOfSyncError BenchNotBuiltError

init_and_run(time=MonotonicTime.EPOCH) async

Initializes and runs a simulation.

This is functionally equivalent to calling Simulation.init followed by Simulation.run, but is implemented as a single gRPC request. This method should be preferred in latency-sensitive simulations to mitigate the risk of loss of synchronization between the invocations of Simulation.init and Simulation.run.

The simulation must be built using the Simulation.build method before it can be initialized.

If a simulation bench is already running, it is replaced by the newly initialized bench. In such case, events that have not yet been retrieved from the sinks will be lost and the sinks are reset to their default enabled/disabled state.

Parameters:

Name	Type	Description	Default
time	MonotonicTime	The time at which the simulation will be initialized.	EPOCH

Raises:

Type	Description
SimulationError	One of the exceptions derived from SimulationError may be raised, such as: SimulationOutOfSyncError BenchNotBuiltError SimulationTerminatedError SimulationDeadlockError SimulationMessageLossError SimulationNoRecipientError SimulationPanicError SimulationTimeoutError

terminate() async

Terminates a simulation.

halt() async

Requests the simulation to stop at the earliest opportunity.

If a stepping method such as Simulation.step or Simulation.run is concurrently being executed, this will cause such method to raise a SimulationNotStartedError before it steps to next scheduler deadline or simulation tick.

Otherwise, this will cause the next call to a Simulation.step_* or Simulation.process_* method to return immediately with SimulationNotStartedError.

Once halted, the simulation can be resumed with another stepping method or a Simulation.process_* call.

Note that the request will only become effective on the next attempt by the simulator to advance the simulation time.

Raises:

Type	Description
SimulationError	One of the exceptions derived from SimulationError may be raised, such as: SimulationNotStartedError

save() async

Saves and returns the current simulation state in a serialized form.

Returns:

Type	Description
bytes	The serialized simulation state.

Raises:

Type	Description
SimulationError	One of the exceptions derived from SimulationError may be raised, such as: SimulationOutOfSyncError BenchNotBuiltError SaveError SimulationNotStartedError

restore(state) async

Restores the simulation from a serialized state.

The simulation must be built using the Simulation.build method before it can be initialized.

If a simulation bench is already running, it is replaced by the newly initialized bench in the restored state. In such case, events that have not yet been retrieved from the sinks will be lost and the sinks are reset to their default enabled/disabled state.

Parameters:

Name	Type	Description	Default
state	bytes	The serialized state of the bench.	required

Raises:

Type	Description
SimulationError	One of the exceptions derived from SimulationError may be raised, such as: SimulationOutOfSyncError BenchNotBuiltError RestoreError

restore_and_run(state) async

Restores the simulation from a serialized state and runs the simulation.

This is functionally equivalent to calling Simulation.restore followed by Simulation.run, but is implemented as a single gRPC request. This method should be preferred in latency-sensitive simulations to mitigate the risk of loss of synchronization between the invocations of Simulation.restore and Simulation.run.

The simulation must be built using the Simulation.build method before it can be initialized.

If a simulation bench is already running, it is replaced by the newly initialized bench in the restored state. In such case, events that have not yet been retrieved from the sinks will be lost and the sinks are reset to their default enabled/disabled state.

Parameters:

Name	Type	Description	Default
state	bytes	The serialized state of the bench.	required

Raises:

Type	Description
SimulationError	One of the exceptions derived from SimulationError may be raised, such as: SimulationOutOfSyncError BenchNotBuiltError RestoreError SimulationTerminatedError SimulationDeadlockError SimulationMessageLossError SimulationNoRecipientError SimulationPanicError SimulationTimeoutError

time() async

Returns the current simulation time.

Returns:

Type	Description
MonotonicTime	The current simulation time.

exceptions.SimulationError: One of the exceptions derived from SimulationError may be raised, such as:

- [`SimulationNotStartedError`][nexosim.exceptions.SimulationNotStartedError]

step() async

Advances simulation time to that of the next scheduled event, processing that event as well as all other events scheduled for the same time and returns the final simulation time.

This method blocks other step* and process* requests until all newly processed events have completed and returns the final simulation time.

Returns:

Type	Description
MonotonicTime	The final simulation time.

Raises:

Type	Description
SimulationError	One of the exceptions derived from SimulationError may be raised, such as: SimulationNotStartedError SimulationTerminatedError SimulationDeadlockError SimulationMessageLossError SimulationNoRecipientError SimulationPanicError SimulationTimeoutError SimulationOutOfSyncError

run() async

Iteratively advances the simulation time until the simulation end, as if by calling step() repeatedly.

The request will complete when all scheduled events are processed or the simulation is halted.

This method blocks other step* and process* requests until completed. The simulation time upon completion is returned.

Returns:

Type	Description
MonotonicTime	The final simulation time.

Raises:

Type	Description
SimulationError	One of the exceptions derived from SimulationError may be raised, such as: SimulationNotStartedError SimulationTerminatedError SimulationDeadlockError SimulationMessageLossError SimulationNoRecipientError SimulationPanicError SimulationTimeoutError SimulationOutOfSyncError

step_until(deadline) async

Iteratively advances the simulation time until the specified deadline, as if by calling Simulation.step repeatedly.

This method blocks other step* and process* requests until all events scheduled up to the specified target time have completed. The simulation time upon completion is returned and is always equal to the specified target time, whether or not an event was scheduled for that time.

Parameters:

Name	Type	Description	Default
deadline	MonotonicTime | Duration	The target time, specified either as an absolute time reference or as a positive duration relative to the current simulation time.	required

Returns:

Type	Description
MonotonicTime	The final simulation time.

Raises:

Type	Description
SimulationError	One of the exceptions derived from SimulationError may be raised, such as: InvalidDeadlineError SimulationNotStartedError SimulationTerminatedError SimulationDeadlockError SimulationMessageLossError SimulationNoRecipientError SimulationPanicError SimulationTimeoutError SimulationOutOfSyncError

list_event_sources() async

Lists available event sources.

This method requires the simulation bench to be built with the Simulation.build method beforehand, but can be used without initialization.

Returns:

Type	Description
list[list[str]]	A list of event source names.

Raises:

Type	Description
SimulationError	One of the exceptions derived from SimulationError may be raised, such as: BenchNotBuiltError

get_event_source_schemas(sources) async

Retrieves the schema of the event sources specified in sources.

This method requires the simulation bench to be built with the Simulation.build method beforehand, but can be used without initialization.

Parameters:

Name	Type	Description	Default
source_names		The names of the event sources.	required

Returns:

Type	Description
dict[tuple[str], dict]	A mapping of event source schemas to their names.

Raises:

Type	Description
SimulationError	One of the exceptions derived from SimulationError may be raised, such as: BenchNotBuiltError

list_query_sources() async

Lists available query sources.

This method requires the simulation bench to be built with the Simulation.build method beforehand, but can be used without initialization.

Returns:

Type	Description
list[list[str]]	A list of query source names.

Raises:

Type	Description
SimulationError	One of the exceptions derived from SimulationError may be raised, such as: BenchNotBuiltError

get_query_source_schemas(sources) async

Retrieves the schema of the query sources specified in sources.

This method requires the simulation bench to be built with the Simulation.build method beforehand, but can be used without initialization.

Parameters:

Name	Type	Description	Default
source_names		The names of the query sources.	required

Returns:

Type	Description
dict[str, dict]	A mapping of query source schemas to their names.

Raises:

Type	Description
SimulationError	One of the exceptions derived from SimulationError may be raised, such as: BenchNotBuiltError

list_event_sinks() async

Lists available event sinks.

This method requires the simulation bench to be built with the Simulation.build method beforehand, but can be used without initialization.

Returns:

Type	Description
list[list[str]]	A list of event sink names.

Raises:

Type	Description
SimulationError	One of the exceptions derived from SimulationError may be raised, such as: BenchNotBuiltError

get_event_sink_schemas(sinks) async

Retrieves the schema of the event sinks specified in sinks.

This method requires the simulation bench to be built with the Simulation.build method beforehand, but can be used without initialization.

Parameters:

Name	Type	Description	Default
sink_names		The names of the event sinks.	required

Returns:

Type	Description
	A mapping of event sinks schemas to their names.

Raises:

Type	Description
SimulationError	One of the exceptions derived from SimulationError may be raised, such as: BenchNotBuiltError

inject_event(source, event=None) async

Injects an event to be processed as soon as possible.

If a stepping method such as Simulation.step or Simulation.run is executed concurrently, the event will be processed at the deadline set by the scheduler event or simulation tick that directly follows the one that is being stepped into.

If the event is injected while the simulation is at rest, the event will be processed at the lapse of the next simulation step (next scheduler event or simulation tick).

Parameters:

Name	Type	Description	Default
source	str | Iterable[str]	The path of the event source.	required
event	Any	an object that can be serialized/deserialized to the expected event type. The None default may be used if the Rust event type is () or Option.	None

Raises:

Type	Description
SimulationError	One of the exceptions derived from SimulationError may be raised, such as: InvalidMessageError EventSourceNotFoundError InvalidEventTypeError SimulationNotStartedError SimulationTerminatedError SimulationDeadlockError SimulationMessageLossError SimulationNoRecipientError SimulationPanicError SimulationTimeoutError

schedule_event(deadline, source, event=None, period=None, with_key=False) async

schedule_event(deadline: MonotonicTime | Duration, source: str | typing.Iterable[str], event: object = None, period: None | Duration = None, with_key: typing.Literal[False] = False) -> None

schedule_event(deadline: MonotonicTime | Duration, source: str | typing.Iterable[str], event: object, period: None | Duration, with_key: typing.Literal[True]) -> EventKey

Schedules an event at a future time.

Events scheduled for the same time and targeting the same model are guaranteed to be processed according to the scheduling order.

Parameters:

Name	Type	Description	Default
deadline	MonotonicTime | Duration	The target time, specified either as an absolute time set in the future of the current simulation time or as a strictly positive duration relative to the current simulation time.	required
source	str | Iterable[str]	The path of the event source.	required
event	object	an object that can be serialized/deserialized to the expected event type. The None default may be used if the Rust event type is () or Option.	None
period	None | Duration	An optional, strictly positive duration expressing the repetition period of the event. If left to None, the event is scheduled only once. Otherwise, the first event is scheduled at the specified deadline and repeated periodically from then on until it is cancelled.	None
with_key	bool	Optionally requests an event key to be returned, which may be used to cancel the event with Simulation.cancel_event.	False

Returns:

Type	Description
EventKey | None	If with_key is set then a key for the scheduled event is returned.

Raises:

Type	Description
SimulationError	One of the exceptions derived from SimulationError may be raised, such as: InvalidDeadlineError InvalidPeriodError EventSourceNotFoundError InvalidEventTypeError InvalidMessageError SimulationNotStartedError

cancel_event(key) async

Cancels a previously schedule event.

Parameters:

Name	Type	Description	Default
key	EventKey	The key for an event that is currently scheduled.	required

Raises:

Type	Description
SimulationError	One of the exceptions derived from SimulationError may be raised, such as: InvalidKeyError SimulationNotStartedError

process_event(source, event=None) async

Broadcasts an event from an event source immediately, blocking other step* and process* requests until completion.

Simulation time remains unchanged.

Parameters:

Name	Type	Description	Default
source	str | Iterable[str]	The path of the event source.	required
event	Any	an object that can be serialized/deserialized to the expected event type. The None default may be used if the Rust event type is () or Option.	None

Raises:

Type	Description
SimulationError	One of the exceptions derived from SimulationError may be raised, such as: InvalidMessageError EventSourceNotFoundError InvalidEventTypeError SimulationNotStartedError SimulationTerminatedError SimulationDeadlockError SimulationMessageLossError SimulationNoRecipientError SimulationPanicError SimulationTimeoutError

process_query(source, request=None, reply_type=object) async

Broadcasts a query from a query source immediately, , blocking other step* and process* requests until completion.

Simulation time remains unchanged.

Parameters:

Name	Type	Description	Default
source	str | Iterable[str]	The path of the query source.	required
request	Any	An object that can be serialized/deserialized to the expected request type. The None default may be used if the Rust request type is () or Option.	None
reply_type	Type[T]	The Python type to which replies to the query should be mapped. If left unspecified, replies are mapped to their canonical representation in terms of built-in Python types such as bool, int, float, str, bytes, dict and list.	object

Returns:

Type	Description
list[T]	An ordered list of replies to the query.

Raises:

Type	Description
SimulationError	One of the exceptions derived from SimulationError may be raised, such as: InvalidMessageError QuerySourceNotFoundError InvalidQueryTypeError SimulationNotStartedError SimulationTerminatedError SimulationDeadlockError SimulationMessageLossError SimulationNoRecipientError SimulationPanicError SimulationTimeoutError SimulationBadQueryError

try_read_events(sink, event_type=object) async

Reads all events from an event sink.

Parameters:

Name	Type	Description	Default
sink	str | Iterable[str]	The path of the event sink.	required
event_type	Type[T]	The Python type to which events should be mapped. If left unspecified, events are mapped to their canonical representation in terms of built-in Python types such as bool, int, float, str, bytes, dict and list.	object

Returns:

Type	Description
list[T]	An ordered list of events.

Raises:

Type	Description
SimulationError	One of the exceptions derived from SimulationError may be raised, such as: InvalidMessageError SinkNotFoundError SinkTerminatedError SinkReadRaceError SinkReadTimeoutError InvalidEventTypeError SimulationNotStartedError

read_event(sink, timeout, event_type=object) async

Waits for the next event from an event sink.

The call is blocking.

Parameters:

Name	Type	Description	Default
sink	str | Iterable[str]	The path of the event sink.	required
event_type	Type[T]	The Python type to which events should be mapped. If left unspecified, events are mapped to their canonical representation in terms of built-in Python types such as bool, int, float, str, bytes, dict and list.	object

Returns:

Type	Description
T	An event.

Raises:

Type	Description
SimulationError	One of the exceptions derived from SimulationError may be raised, such as: InvalidMessageError SinkNotFoundError SinkTerminatedError SinkReadRaceError SinkReadTimeoutError InvalidEventTypeError SimulationNotStartedError

enable_sink(sink) async

Enables the reception of new events by the specified sink.

Note that the initial state of a sink may be either enabled or disabled depending on the bench initializer.

Parameters:

Name	Type	Description	Default
sink	str | Iterable[str]	The path of the event sink.	required

Raises:

Type	Description
SimulationError	One of the exceptions derived from SimulationError may be raised, such as: SinkNotFoundError SinkReadRaceError MissingArgumentError SimulationNotStartedError

disable_sink(sink) async

Disables the reception of new events by the specified sink.

Note that the initial state of a sink may be either enabled or disabled depending on the bench initializer.

Parameters:

Name	Type	Description	Default
sink	str | Iterable[str]	The path of the event sink.	required

Raises:

Type	Description
SimulationError	One of the exceptions derived from SimulationError may be raised, such as: SinkNotFoundError SinkReadRaceError MissingArgumentError SimulationNotStartedError