nexosim

The root module.

This module defines the Simulation type, which acts as a front-end to a NeXosim gRPC simulation server.

Example usage

ClientServer

from dataclasses import dataclass
from nexosim import Simulation
from nexosim.time import Duration

# We could read simulation events as dictionaries, but it is often more
# convenient to use classes that mirror their Rust counterpart.
@dataclass
class OutputEvent:
    foo: int
    bar: str

# Connect to a local server listening on the 41633 port.
with Simulation(address='localhost:41633') as sim:

    # Initialize the simulation.
    sim.build()
    sim.init()

    # Schedule an event on the "input" event source
    sim.schedule_event(Duration(1), "input", 1)

    # Advance the simulation to the next scheduled timestamp.
    sim.step()

    # Read a list of `OutputEvent` objects from the "output" event sink.
    outputs = sim.try_read_events("output", OutputEvent)
    print(outputs)

    # Advance the simulation by 3s and read the final simulation time.
    t = sim.step_until(Duration(3))

    print(t)

use std::error::Error;

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"))
}

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

EventKey

A handle to a scheduled event.

Event keys are opaque objects. They are meant to be created by the Simulation.schedule_event method and consumed by the Simulation.cancel_event method.

Simulation

A 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 a context manager. If not used in a with statement, the close() method should be called after use.

Parameters:

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

close()

Closes the gRPC channel.

build(cfg=None)

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)

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)

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()

Terminates a simulation.

halt()

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()

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)

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)

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()

Returns the current simulation time.

Returns:

Type	Description
MonotonicTime	The current simulation time.

Raises:

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

step()

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

This method blocks 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()

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

The request blocks until all scheduled events are processed or the simulation is halted.

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)

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

This method blocks 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()

Lists available event sources.

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)

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()

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)

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()

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)

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)

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)

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)

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)

Broadcasts an event from an event source immediately, blocking 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)

Broadcasts a query from a query source immediately, blocking 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)

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)

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)

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)

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