API Documentation

class xronos.Element(sdk_instance: Element)

A reactor element.

This is used as a base class for all elements that can be contained by a reactor, including the Reactor class itself.

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

Annotate the element with an attribute.

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.

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

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.

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) – When set, the connection waits for delay before delivering the messages to to. (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 the collection of telemetry data during program execution.

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.

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 program started with execute() at the next convenience. This triggers shutdown after completing all currently active reactions, and stops program execution after processing all reactions triggered by shutdown.

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.

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.

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

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

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

A port that receives values from other reactors.

The input port does not provide direct access to received values. A reaction may declare an input port as a Trigger or Effect to read or write its value.

Type Parameters:: T – The type of values carried by the port.

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

Annotate the element with an attribute.

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.

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

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 type of values carried by the port.

class xronos.Metric(sdk_instance: Element)

Allows recording values to an external data base from reaction handlers.

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.

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.

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

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

Provides write access to a metric.

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

record(value: int | float) → None

Record a value.

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

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

A port that sends values to other reactors.

The output port does not provide direct access to received values. A reaction may declare an output port as a Trigger or Effect to read or write its value.

Type Parameters:: T – The type of values carried by the port.

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

Annotate the element with an attribute.

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.

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

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

An event source that emits events in regular intervals.

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

Annotate the element with an attribute.

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.

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

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

If period is 0 (the default), then the timer will trigger only once with a delay of offset after startup triggers. If offset is also set to 0, the timer triggers simultaneously to startup.

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

An element for triggering events from outside the xronos program (e.g. sensors).

Physical events may be used to schedule new events from outside of the scope of the reactor program. These could be external event handlers that respond to sensor inputs. Physical events do not provide direct access to their values. A reaction may declare a Trigger to access the value associated with an active event.

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

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

Annotate the element with an attribute.

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.

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

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

trigger(value: T) → None

Create a new event with an automatically assigned timestamp.

Parameters:: value (T) – A value to be associated with the newly created 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 any triggered events.

class xronos.PortEffect[T]

Provides write access to a port.

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

get() → T

Get the current value of the referenced port.

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

is_present() → bool

Check if an event is present at the current Timestamp.

set(value: T) → None

Set the port value and send a message to connected ports.

Can be called multiple times, but at each time 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 schedulable event source.

Programmable timers may be used by reactions to schedule new events in the future. Similar to ports, programmable timers may not be accessed directly. Instead, reaction’s may declare a ProgrammableTimerEffect to schedule new events, or a Trigger to access the value associated with an active event.

Type Parameters:: T – The type of values associated with events scheduled via the programmable timer.

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

Annotate the element with an attribute.

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.

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

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 type of values associated with events scheduled via the programmable timer.

class xronos.ProgrammableTimerEffect(sdk_effect: ProgrammableTimerEffect)

Provides write access to a programmable timer.

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

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 produces 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.

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.

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.

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) – When set, the connection waits for delay before delivering the messages to to. (optional)

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

Raises:

ValidationError – If an invalid connections is created.

property fqn: str: Fully qualified name of the element (read-only).

get_lag() → timedelta

Get the current lag.

Since time in the xronos runtime 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.

get_time() → datetime

Get the current point in time.

Note

This does not read wall-clock time. The xronos runtime uses an internal clock to control how a program advances. get_time() reads the current time as measured by the internal clock.

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 program started with execute() at the next convenience. This triggers shutdown after completing all currently active reactions, and stops program execution after processing all reactions triggered by shutdown.

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

Emits an event once when the program shuts down.

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

Annotate the element with an attribute.

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.

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

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

class xronos.Startup(sdk_instance: Element)

Emits an event once when the program starts executing.

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

Annotate the element with an attribute.

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.

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

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 current value of the referenced reactor element.

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

is_present() → bool

Check if the referenced event source has produced an event.

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.