API Documentation

class xronos.Element(sdk_instance: Element)

A reactor element.

Reactor elements are objects that can be contained by reactors and that have special meaning to the Xronos SDK.

Reactor elements should not be instantiated directly. Use the corresponding <Element>Declaration classes instead.

add_attribute(key: str, value: str | bool | int | float) → None

Annotate the element with an attribute.

Adding the attribute only succeeds, if the given key has not been set before on the same element.

See Attributes for more information.

Parameters:

key (str) – The name of the attribute to add
value (str | bool | int | float) – The value of the attribute to add

Raises:

KeyError – If the attribute key was added before.

add_attributes(attributes: dict[str, str | bool | int | float]) → None

Annotate the element with multiple attributes.

Adding the attributes only succeeds, if the given key has not been set before on the same element.

See Attributes for more information.

Parameters:: attributes (dict[str, str | bool | int | float]) – A map of attribute names and their values to be added.
Raises:: KeyError – If the attribute key was added before.

property fqn: str

Fully qualified name of the element (read-only).

The fully qualified name (FQN) represents the containment hierarchy. It consists of the containing reactor’s FQN plus the element’s name separated by a .. For top-level reactors (those owned by the environment), the FQN is equal to the name.

property name: str: Name of the element (read-only).

class xronos.Environment(fast: bool = False, timeout: timedelta | None = None)

The entry point for assembling and executing reactor programs.

The environment acts as an execution context for reactor programs. It manages both the creation of reactors and the execution of reactor programs.

Parameters:

fast (bool) – If set to True, enables a special mode of execution that skips waiting between executing events and instead processes events as fast as possible. This is relevant for testing. (optional)
timeout (timedelta | None) – Specifies a maximum duration for which the program executes. When the program reaches the timeout it terminates and triggers shutdown. This is mostly useful for testing. (optional)

connect(from_: InputPort[T] | OutputPort[T], to: InputPort[T] | OutputPort[T], delay: timedelta | None = None) → None

Connect two ports.

Creates a new connection from the port given in from_ to the port given in to.

If delay is None messages are delivered without a delay. This means that the timestamp at which the message is received is the same as the timestamp at which it was sent.

If delay is set messages are delivered with a delay. This means that the timestamp at which the message is received is the timestamp at which it was sent plus delay.

Parameters:

from (InputPort[T] | OutputPort[T]) – The port to draw the connection from.
to (InputPort[T] | OutputPort[T]) – The port to draw the connection to.
delay (timedelta | None) – The delay to apply to messages. (optional)

Raises:

ValidationError – If an invalid connections is created.

create_reactor(name: str, reactor_class: type, *args, **kwargs)

Create a new reactor.

A factory method for instantiating and registering a new reactor.

Parameters:

name (str) – Name of the reactor to be instantiated.
reactor_class (type) – The reactor class to be instantiated. Must be a subclass of Reactor.
args (Any) – Arguments to be passed to the __init__ method of reactor_class.
kwargs (Any) – Keyword arguments to be passed to the __init__ method of reactor_class.

Returns:

An instance of the class provided in the reactor_class argument.

Return type:

reactor_class

enable_telemetry(application_name: str = 'xronos', endpoint: str = 'localhost:4317') → None

Enable collecting and sending telemetry data from the application.

See Telemetry and Dashboard for more information on producing, collecting and visualizing telemetry data.

Parameters:

application_name (str) – The name of the application as it should appear in the telemetry metadata.
endpoint (str) – The network endpoint to send telemetry data to. This is typically port 4137 on the host running the Dashboard.

execute() → None

Execute the reactor program.

Initiates the execution of a reactor program. This triggers startup and instructs the runtime to start processing reactions.

Returns when the reactor program terminates. The reactor program terminates when there are no more events, or after calling request_shutdown().

Raises:: ValidationError – When the program is invalid or reaches an invalid state.

request_shutdown() → None

Request the termination of the currently running reactor program.

Terminates a running program at the next convenience. After completing all currently active reactions, this triggers shutdown. Once all reactions triggered by shutdown are processed, the program terminates.

class xronos.EventSource[T](sdk_instance: Element)

An element that may emit events and act as reaction trigger.

add_attribute(key: str, value: str | bool | int | float) → None

Annotate the element with an attribute.

Adding the attribute only succeeds, if the given key has not been set before on the same element.

See Attributes for more information.

Parameters:

key (str) – The name of the attribute to add
value (str | bool | int | float) – The value of the attribute to add

Raises:

KeyError – If the attribute key was added before.

add_attributes(attributes: dict[str, str | bool | int | float]) → None

Annotate the element with multiple attributes.

Adding the attributes only succeeds, if the given key has not been set before on the same element.

See Attributes for more information.

Parameters:: attributes (dict[str, str | bool | int | float]) – A map of attribute names and their values to be added.
Raises:: KeyError – If the attribute key was added before.

property fqn: str

Fully qualified name of the element (read-only).

The fully qualified name (FQN) represents the containment hierarchy. It consists of the containing reactor’s FQN plus the element’s name separated by a .. For top-level reactors (those owned by the environment), the FQN is equal to the name.

property name: str: Name of the element (read-only).

class xronos.InputPort[T](sdk_instance: Element)

A reactor element for receiving messages from other reactors.

Input ports can be used as a reaction Trigger and provide an interface for reactors to receive messages from other reactors.

Input ports may be connected to other ports so that messages are forwarded automatically (see Environment.connect() and Reactor.connect()).

Reactions of other reactors may also use input ports as a PortEffect allowing an external reaction handler to send messages directly to the port.

This class is not intended to be instantiated directly. Use InputPortDeclaration instead.

Type Parameters:: T – The value type associated with messages.

add_attribute(key: str, value: str | bool | int | float) → None

Annotate the element with an attribute.

Adding the attribute only succeeds, if the given key has not been set before on the same element.

See Attributes for more information.

Parameters:

key (str) – The name of the attribute to add
value (str | bool | int | float) – The value of the attribute to add

Raises:

KeyError – If the attribute key was added before.

add_attributes(attributes: dict[str, str | bool | int | float]) → None

Annotate the element with multiple attributes.

Adding the attributes only succeeds, if the given key has not been set before on the same element.

See Attributes for more information.

Parameters:: attributes (dict[str, str | bool | int | float]) – A map of attribute names and their values to be added.
Raises:: KeyError – If the attribute key was added before.

property fqn: str

Fully qualified name of the element (read-only).

The fully qualified name (FQN) represents the containment hierarchy. It consists of the containing reactor’s FQN plus the element’s name separated by a .. For top-level reactors (those owned by the environment), the FQN is equal to the name.

property name: str: Name of the element (read-only).

class xronos.InputPortDeclaration[T](attributes: dict[str, str | bool | int | float] | None = None)

A declaration for an InputPort[T].

Parameters:: attributes (optional) – A dict of attributes characterizing the input port.
Type Parameters:: T – The value type associated with messages.

class xronos.Metric(sdk_instance: Element)

A reactor element for recording metric data to an external data base.

Can be used by a reaction as a MetricEffect allowing the reaction handler to record values using the metric.

This class is not intended to be instantiated directly. Use MetricDeclaration instead.

add_attribute(key: str, value: str | bool | int | float) → None

Annotate the element with an attribute.

Adding the attribute only succeeds, if the given key has not been set before on the same element.

See Attributes for more information.

Parameters:

key (str) – The name of the attribute to add
value (str | bool | int | float) – The value of the attribute to add

Raises:

KeyError – If the attribute key was added before.

add_attributes(attributes: dict[str, str | bool | int | float]) → None

Annotate the element with multiple attributes.

Adding the attributes only succeeds, if the given key has not been set before on the same element.

See Attributes for more information.

Parameters:: attributes (dict[str, str | bool | int | float]) – A map of attribute names and their values to be added.
Raises:: KeyError – If the attribute key was added before.

property description: str: Description of the metric (read-only).

property fqn: str

Fully qualified name of the element (read-only).

The fully qualified name (FQN) represents the containment hierarchy. It consists of the containing reactor’s FQN plus the element’s name separated by a .. For top-level reactors (those owned by the environment), the FQN is equal to the name.

property name: str: Name of the element (read-only).

property unit: str: Unit of the metric (read-only).

A declaration for a Metric.

Parameters:

description (str) – A description of the metric.
unit (str | None) – The unit of the metric.
attributes (dict[str, str | bool | int | float] | None) – A dict of attributes characterizing the metric.

class xronos.MetricEffect(sdk_metric: MetricEffect)

Allows a reaction to record telemetry data using a given Metric.

This class is not intended to be instantiated directly. Use add_effect() instead.

record(value: int | float) → None

Record a value at the current timestamp.

Parameters:: value (int | float) – The value to be recorded.

class xronos.OutputPort[T](sdk_instance: Element)

A reactor element for sending messages to other reactors.

Output ports can be used as a PortEffect and provide an interface for reactors to send messages to other reactors.

Output ports may be connected to other ports so that messages are forwarded automatically (see Environment.connect() and Reactor.connect()).

Other reactors may also use output ports as a reaction Trigger allowing an external reaction handler to receive messages directly from the port.

This class is not intended to be instantiated directly. Use OutputPortDeclaration instead.

Type Parameters:: T – The value type associated with messages.

add_attribute(key: str, value: str | bool | int | float) → None

Annotate the element with an attribute.

Adding the attribute only succeeds, if the given key has not been set before on the same element.

See Attributes for more information.

Parameters:

key (str) – The name of the attribute to add
value (str | bool | int | float) – The value of the attribute to add

Raises:

KeyError – If the attribute key was added before.

add_attributes(attributes: dict[str, str | bool | int | float]) → None

Annotate the element with multiple attributes.

Adding the attributes only succeeds, if the given key has not been set before on the same element.

See Attributes for more information.

Parameters:: attributes (dict[str, str | bool | int | float]) – A map of attribute names and their values to be added.
Raises:: KeyError – If the attribute key was added before.

property fqn: str

Fully qualified name of the element (read-only).

The fully qualified name (FQN) represents the containment hierarchy. It consists of the containing reactor’s FQN plus the element’s name separated by a .. For top-level reactors (those owned by the environment), the FQN is equal to the name.

property name: str: Name of the element (read-only).

class xronos.OutputPortDeclaration[T](attributes: dict[str, str | bool | int | float] | None = None)

A declaration for an OutputPort[T].

Parameters:: attributes (optional) – A dict of attributes characterizing the output port.
Type Parameters:: T – The type of values carried by the port.

class xronos.PeriodicTimer(sdk_instance: Element)

A reactor element that emits events in regular intervals.

Can be used as a reaction Trigger.

This class is not intended to be instantiated directly. Use PeriodicTimerDeclaration instead.

add_attribute(key: str, value: str | bool | int | float) → None

Annotate the element with an attribute.

Adding the attribute only succeeds, if the given key has not been set before on the same element.

See Attributes for more information.

Parameters:

key (str) – The name of the attribute to add
value (str | bool | int | float) – The value of the attribute to add

Raises:

KeyError – If the attribute key was added before.

add_attributes(attributes: dict[str, str | bool | int | float]) → None

Annotate the element with multiple attributes.

Adding the attributes only succeeds, if the given key has not been set before on the same element.

See Attributes for more information.

Parameters:: attributes (dict[str, str | bool | int | float]) – A map of attribute names and their values to be added.
Raises:: KeyError – If the attribute key was added before.

property fqn: str

Fully qualified name of the element (read-only).

The fully qualified name (FQN) represents the containment hierarchy. It consists of the containing reactor’s FQN plus the element’s name separated by a .. For top-level reactors (those owned by the environment), the FQN is equal to the name.

property name: str: Name of the element (read-only).

property offset: timedelta

The delay between startup and the first event emitted by the timer (read-write).

Raises:: RuntimeError – When set during program execution.

property period: timedelta

The delay in between two events emitted by the timer (read-write).

Raises:: RuntimeError – When set during program execution.

class xronos.PeriodicTimerDeclaration(period: timedelta = datetime.timedelta(0), offset: timedelta = datetime.timedelta(0), attributes: dict[str, str | bool | int | float] | None = None)

A declaration for a PeriodicTimer.

Parameters:

period (timedelta) – The delay in between two events emitted by the timer (optional).
offset (timedelta) – The delay between the startup event and the first event emitted by the timer (optional).
attributes (optional) – A dict of attributes characterizing the timer.

class xronos.PhysicalEvent[T](sdk_instance: Element)

A reactor element for receiving events from external sources.

Physical events may be used to trigger reactions from a context outside of the scope of the reactor program. These could be external event handlers that respond to sensor inputs.

Can be used as a reaction Trigger allowing the reaction handler to read the associated value.

Type Parameters:: T – The type of values carried by emitted events.

add_attribute(key: str, value: str | bool | int | float) → None

Annotate the element with an attribute.

Adding the attribute only succeeds, if the given key has not been set before on the same element.

See Attributes for more information.

Parameters:

key (str) – The name of the attribute to add
value (str | bool | int | float) – The value of the attribute to add

Raises:

KeyError – If the attribute key was added before.

add_attributes(attributes: dict[str, str | bool | int | float]) → None

Annotate the element with multiple attributes.

Adding the attributes only succeeds, if the given key has not been set before on the same element.

See Attributes for more information.

Parameters:: attributes (dict[str, str | bool | int | float]) – A map of attribute names and their values to be added.
Raises:: KeyError – If the attribute key was added before.

property fqn: str

Fully qualified name of the element (read-only).

The fully qualified name (FQN) represents the containment hierarchy. It consists of the containing reactor’s FQN plus the element’s name separated by a .. For top-level reactors (those owned by the environment), the FQN is equal to the name.

property name: str: Name of the element (read-only).

trigger(value: T) → None

Emit a new event with an associated value and trigger reactions.

The event will be assigned a timestamp equal to the current wall-clock time.

Parameters:: value (T) – The value to be associated with the emitted event.

class xronos.PhysicalEventDeclaration[T](attributes: dict[str, str | bool | int | float] | None = None)

A declaration for a PhysicalEvent[T].

Parameters:: attributes (optional) – A dict of attributes characterizing the physical event.
Type Parameters:: T – The type of values carried by emitted events.

class xronos.PortEffect[T]

Allows a reaction to write data to a given InputPort or OutputPort.

This class is not intended to be instantiated directly. Use add_effect() instead.

Type Args:: T: The value type associated with the port.

get() → T

Get a previously set value.

Returns:: The current value if an event is present.
Raises:: AbsentError – If called and is_present() returns False.
Return type:: T

is_present() → bool

Check if an event is present at the current Timestamp.

set(value: T) → None

Write a value to the port sending a message to connected ports.

May be called multiple times, but at most one value is sent to connected ports. When called repeatedly at a given timestamp, the previous value is overwritten.

Parameters:: value (T) – The value to be written to the referenced port.

class xronos.ProgrammableTimer[T](sdk_instance: Element)

A reactor element for scheduling new events.

Programmable timers may be used by reactions to schedule new events that will be emitted in the future. They can be used both as a reaction Trigger as and as a ProgrammableTimerEffect.

Type Parameters:: T – The value type associated with events emitted by the programmable timer.

add_attribute(key: str, value: str | bool | int | float) → None

Annotate the element with an attribute.

Adding the attribute only succeeds, if the given key has not been set before on the same element.

See Attributes for more information.

Parameters:

key (str) – The name of the attribute to add
value (str | bool | int | float) – The value of the attribute to add

Raises:

KeyError – If the attribute key was added before.

add_attributes(attributes: dict[str, str | bool | int | float]) → None

Annotate the element with multiple attributes.

Adding the attributes only succeeds, if the given key has not been set before on the same element.

See Attributes for more information.

Parameters:: attributes (dict[str, str | bool | int | float]) – A map of attribute names and their values to be added.
Raises:: KeyError – If the attribute key was added before.

property fqn: str

Fully qualified name of the element (read-only).

The fully qualified name (FQN) represents the containment hierarchy. It consists of the containing reactor’s FQN plus the element’s name separated by a .. For top-level reactors (those owned by the environment), the FQN is equal to the name.

property name: str: Name of the element (read-only).

class xronos.ProgrammableTimerDeclaration[T](attributes: dict[str, str | bool | int | float] | None = None)

A declaration for a ProgrammableTimer[T].

Parameters:

attributes (optional) – A dict of attributes characterizing the
timer. (programmable)

Type Parameters:

T – The value type associated with events emitted by the programmable timer.

class xronos.ProgrammableTimerEffect(sdk_effect: ProgrammableTimerEffect)

Allows a reaction to schedule future events using a ProgrammableTimer.

This class is not intended to be instantiated directly. Use add_effect() instead.

Type Args:: T: The value type associated with the programmable timer.

schedule(value: T, delay: timedelta = datetime.timedelta(0)) → None

Schedule a future event.

Parameters:

value (T) – The value to be associated with the event.
delay (timedelta) – The time to wait until the new event is processed.

class xronos.ReactionInterface(context: ReactionContext)

Helper class for defining the interfaces of a reaction.

This class is not intended to be instantiated directly. An instance of this class is passed automatically to any method decorated with {attr}`~xronos.reaction`.

Declare a reaction effect.

An effect provides read and write access to the referenced element, but does not trigger the reaction.

Parameters:: target (InputPort[T] | OutputPort[T] | ProgrammableTimer[T] | Metric) – The reactor element to declare as the reaction effect.
Returns:: A new effect object that can be used by the reaction handler to write to ports or schedule events.
Return type:: PortEffect[T] | ProgrammableTimerEffect[T] | MetricEffect

add_trigger(event_source: EventSource[T]) → Trigger[T]

Declare a reaction trigger.

When the triggering event source emits an event, the reaction that declares the trigger will be invoked automatically.

Parameters:: event_source (EventSource[T]) – The event source to declare as the reaction trigger.
Returns:: A new trigger object that can be used by the reaction handler to check presence and read values.
Return type:: Trigger[T]

class xronos.Reactor

An abstract reactor that can be subclassed to define new reactors.

add_attribute(key: str, value: str | bool | int | float) → None

Annotate the element with an attribute.

Adding the attribute only succeeds, if the given key has not been set before on the same element.

See Attributes for more information.

Parameters:

key (str) – The name of the attribute to add
value (str | bool | int | float) – The value of the attribute to add

Raises:

KeyError – If the attribute key was added before.

add_attributes(attributes: dict[str, str | bool | int | float]) → None

Annotate the element with multiple attributes.

Adding the attributes only succeeds, if the given key has not been set before on the same element.

See Attributes for more information.

Parameters:: attributes (dict[str, str | bool | int | float]) – A map of attribute names and their values to be added.
Raises:: KeyError – If the attribute key was added before.

connect(from_: InputPort[T] | OutputPort[T], to: InputPort[T] | OutputPort[T], delay: timedelta | None = None) → None

Connect two ports.

Creates a new connection from the port given in from_ to the port given in to.

If delay is None messages are delivered without a delay. This means that the timestamp at which the message is received is the same as the timestamp at which it was sent.

If delay is set messages are delivered with a delay. This means that the timestamp at which the message is received is the timestamp at which it was sent plus delay.

Also see Environment.connect.

Parameters:

from (InputPort[T] | OutputPort[T]) – The port to draw the connection from.
to (InputPort[T] | OutputPort[T]) – The port to draw the connection to.
delay (timedelta | None) – The delay to apply to messages. (optional)

Raises:

ValidationError – If an invalid connections is created.

create_reactor(name: str, reactor_class: type, *args, **kwargs)

Create a nested reactor.

A factory method for instantiating and registering a new nested reactor. In contrast to Environment.create_reactor(), the newly created reactor is contained by self.

Parameters:

name (str) – Name of the reactor to be instantiated.
reactor_class (type) – The reactor class to be instantiated. Must be a subclass of Reactor.
args (Any) – Arguments to be passed to the __init__ method of reactor_class.
kwargs (Any) – Keyword arguments to be passed to the __init__ method of reactor_class.

Returns:

An instance of the class provided in the reactor_class argument.

Return type:

reactor_class

property fqn: str

Fully qualified name of the element (read-only).

The fully qualified name (FQN) represents the containment hierarchy. It consists of the containing reactor’s FQN plus the element’s name separated by a .. For top-level reactors (those owned by the environment), the FQN is equal to the name.

get_lag() → timedelta

Get the current lag.

Since in the Xronos SDK time does not advance while reactions execute, the internal clock may advance slower than a wall clock would. The lag denotes the difference between the wall clock and the internal clock. It is a measure of how far the execution of reactions lags behind events in the physical world.

Returns:: The current lag.
Return type:: timedelta

get_time() → datetime

Get the current timestamp.

Note

This does not read wall-clock time. The Xronos runtime uses an internal clock to control how a program advances.

Returns:: The current timestamp as provided by the internal clock.
Return type:: datetime

get_time_since_startup() → timedelta

Get the time that passed since the startup event.

Returns:: The difference between the current time point given by get_time() and the time at which the program started.
Return type:: timedelta

property name: str: Name of the element (read-only).

request_shutdown() → None

Request the termination of the currently running reactor program.

Terminates a running program at the next convenience. After completing all currently active reactions, this triggers shutdown. Once all reactions triggered by shutdown are processed, the program terminates.

property shutdown: Shutdown

Shutdown event source.

Triggers once right before the program execution ends.

property startup: Startup

Startup event source.

Triggers once when the program execution starts.

class xronos.Shutdown(sdk_instance: Element)

A reactor element that emits an event right before the program shuts down.

Can be used as a reaction Trigger.

add_attribute(key: str, value: str | bool | int | float) → None

Annotate the element with an attribute.

Adding the attribute only succeeds, if the given key has not been set before on the same element.

See Attributes for more information.

Parameters:

key (str) – The name of the attribute to add
value (str | bool | int | float) – The value of the attribute to add

Raises:

KeyError – If the attribute key was added before.

add_attributes(attributes: dict[str, str | bool | int | float]) → None

Annotate the element with multiple attributes.

Adding the attributes only succeeds, if the given key has not been set before on the same element.

See Attributes for more information.

Parameters:: attributes (dict[str, str | bool | int | float]) – A map of attribute names and their values to be added.
Raises:: KeyError – If the attribute key was added before.

property fqn: str

Fully qualified name of the element (read-only).

The fully qualified name (FQN) represents the containment hierarchy. It consists of the containing reactor’s FQN plus the element’s name separated by a .. For top-level reactors (those owned by the environment), the FQN is equal to the name.

property name: str: Name of the element (read-only).

class xronos.Startup(sdk_instance: Element)

A reactor element that emits an event program starts executing.

Can be used as a reaction Trigger.

add_attribute(key: str, value: str | bool | int | float) → None

Annotate the element with an attribute.

Adding the attribute only succeeds, if the given key has not been set before on the same element.

See Attributes for more information.

Parameters:

key (str) – The name of the attribute to add
value (str | bool | int | float) – The value of the attribute to add

Raises:

KeyError – If the attribute key was added before.

add_attributes(attributes: dict[str, str | bool | int | float]) → None

Annotate the element with multiple attributes.

Adding the attributes only succeeds, if the given key has not been set before on the same element.

See Attributes for more information.

Parameters:: attributes (dict[str, str | bool | int | float]) – A map of attribute names and their values to be added.
Raises:: KeyError – If the attribute key was added before.

property fqn: str

Fully qualified name of the element (read-only).

The fully qualified name (FQN) represents the containment hierarchy. It consists of the containing reactor’s FQN plus the element’s name separated by a .. For top-level reactors (those owned by the environment), the FQN is equal to the name.

property name: str: Name of the element (read-only).

class xronos.Trigger[T]

Provides read access to an event source that a reaction is triggered by.

This class is not intended to be instantiated directly. Use add_trigger() instead.

get() → T

Get the value of a currently present event.

Raises:: AbsentError – If called and is_present() returns False.

is_present() → bool

Check if an event is present at the current timestamp.

exception xronos.ValidationError

Exception that is thrown when a program reaches an invalid state.

add_note(): Exception.add_note(note) – add a note to the exception

args

with_traceback(): Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

xronos.reaction(declaration: Callable[[R, ReactionInterface], Callable[[], None]]) → ReactionDescriptor[R]

Decorator that is used to declare reactions.

Parameters:: declaration (Callable[[R, ReactionInterface], Callable[[], None]]) – The decorated method. Must accept an ReactionInterface as its first argument and return a reaction handler. Failing to return a handler will result in an exception when the reactor containing the reaction is initialized.