API Documentation

exception xronos.AbsentError

Indicates an attempt to read a value that is absent.

See Trigger.get().

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.

class xronos.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.

Attributes apply per key. Each attribute is added only if its key has not been set before on the same element. Attributes with rejected keys do not prevent the remaining attributes from being added.

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 any 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_: OutputPort[T], to: InputPort[T], delay: timedelta | None = None) → None

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

connect(from_: InputPort[T], to: InputPort[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 4317 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, after calling trigger_shutdown() on a ShutdownEffect, or when it reaches the timeout configured on the environment.

An exception raised by a reaction handler also terminates the program. The runtime then stops invoking reaction handlers; a handler that is about to start may still run. Shutdown reactions run in any case, and execute() reraises the first exception.

Raises:

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

class xronos.EventSource[T]

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.

Attributes apply per key. Each attribute is added only if its key has not been set before on the same element. Attributes with rejected keys do not prevent the remaining attributes from being added.

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 any 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]

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.

Attributes apply per key. Each attribute is added only if its key has not been set before on the same element. Attributes with rejected keys do not prevent the remaining attributes from being added.

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 any 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 (dict[str, str | bool | int | float] | None) – A dict of attributes characterizing the input port (optional).

Type Parameters:

T – The value type associated with messages.

exception xronos.InvalidNameError

Exception that is thrown when creating an element with an invalid name: one that is empty, contains whitespace or any of .,/*$?#@, or is already in use within the same parent.

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.

exception xronos.InvalidReactionHandler

Exception that is thrown when a reaction returns an invalid handler.

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.

class xronos.Metric

A reactor element for recording numeric values to the Xronos Dashboard.

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.

Attributes apply per key. Each attribute is added only if its key has not been set before on the same element. Attributes with rejected keys do not prevent the remaining attributes from being added.

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

class xronos.MetricDeclaration(description: str, unit: str | None = None, attributes: dict[str, str | bool | int | float] | None = None)

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

Allows a reaction to record values 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.

Raises:

ValidationError – If called before execution has started.

class xronos.OutputPort[T]

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.

Attributes apply per key. Each attribute is added only if its key has not been set before on the same element. Attributes with rejected keys do not prevent the remaining attributes from being added.

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 any 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 (dict[str, str | bool | int | float] | None) – A dict of attributes characterizing the output port (optional).

Type Parameters:

T – The type of values carried by the port.

class xronos.PeriodicTimer

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.

Attributes apply per key. Each attribute is added only if its key has not been set before on the same element. Attributes with rejected keys do not prevent the remaining attributes from being added.

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 any 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 (dict[str, str | bool | int | float] | None) – A dict of attributes characterizing the timer (optional).

class xronos.PhysicalEvent[T]

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.

Attributes apply per key. Each attribute is added only if its key has not been set before on the same element. Attributes with rejected keys do not prevent the remaining attributes from being added.

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 any 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) → TriggerStatus

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.

Returns:

The outcome of the attempt. TriggerStatus.ACCEPTED on success.

Return type:

TriggerStatus

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

A declaration for a PhysicalEvent[T].

Parameters:

attributes (dict[str, str | bool | int | float] | None) – A dict of attributes characterizing the physical event (optional).

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 currently present.

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.

Raises:

ValidationError – If called before execution has started.

class xronos.ProgrammableTimer[T]

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.

Attributes apply per key. Each attribute is added only if its key has not been set before on the same element. Attributes with rejected keys do not prevent the remaining attributes from being added.

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 any 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 (dict[str, str | bool | int | float] | None) – A dict of attributes characterizing the programmable timer (optional).

Type Parameters:

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

class xronos.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.

Raises:

ValidationError – If called before execution has started.

class xronos.ReactionContext

The context passed to a reaction declaration.

An instance is passed automatically to any method decorated with @reaction or @reaction_with_deadline. Use it to declare the reaction’s triggers and effects, and to read the reaction-scoped timing API. The reaction handler may capture the context object in its closure to access the timing API during execution.

Note

The timing attributes are only meaningful while the reaction handler executes. They remain accessible during the reaction declaration, but there they return defined placeholders that must not be relied upon.

add_effect(target: InputPort[T] | OutputPort[T]) → PortEffect[T]

add_effect(target: ProgrammableTimer[T]) → ProgrammableTimerEffect[T]

add_effect(target: Metric) → MetricEffect

add_effect(target: Shutdown) → ShutdownEffect

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 | Shutdown) – 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 | ShutdownEffect

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]

property current_time: datetime

The current time as provided by the internal clock (read-only).

This does not read wall-clock time. The Xronos runtime uses an internal clock to control how a program advances. The internal clock does not advance while a reaction handler executes, so this value does not change while the handler runs: any two reads within the same handler return the same value.

This is a reaction-scoped accessor and is only meaningful while the reaction handler executes; values read outside a handler must not be relied upon.

If the program is not yet executing (for example, when accessed during reaction declaration), this returns a default value: the epoch.

property deadline: datetime | None

The deadline for the current handler invocation (read-only).

If the reaction was declared with a deadline duration D (see @reaction_with_deadline), this returns the wall-clock instant by which the handler must complete, equal to current_time + D. It is therefore anchored to current_time and stays fixed for the duration of the handler. The handler meets its deadline if it completes before the wall clock reaches this instant.

This is a reaction-scoped accessor and is only meaningful while the reaction handler executes; values read outside a handler must not be relied upon.

If no deadline was declared, or if the program is not yet executing (for example, when accessed during reaction declaration), this is None.

property elapsed_time: timedelta

How far the internal clock has advanced since startup (read-only).

This is the difference between current_time and the time at which the program started. It is measured on the internal clock and does not depend on wall-clock time. Like current_time, it does not change while the handler runs.

This is a reaction-scoped accessor and is only meaningful while the reaction handler executes; values read outside a handler must not be relied upon.

If the program is not yet executing (for example, when accessed during reaction declaration), this returns a default value: zero.

property is_before_deadline: bool

Whether the handler is still within its deadline (read-only).

This is a reaction-scoped accessor and is only meaningful while the reaction handler executes; values read outside a handler must not be relied upon.

Returns True while the wall clock has not yet reached the deadline (the slack is positive), and False once the deadline has been missed. If no deadline was declared, this is always True.

property lag: timedelta

The current lag (read-only).

The lag is the difference between wall-clock time and the current time (wall-clock now minus current_time). It relates the internal clock to the advancing wall clock and therefore changes while the handler runs: the current time does not advance, but the wall clock does, so the lag measures how far the wall clock has run ahead of the internal clock. It shows how far the execution of reactions lags behind the events it processes.

This is a reaction-scoped accessor and is only meaningful while the reaction handler executes; values read outside a handler must not be relied upon.

If the program is not yet executing (for example, when accessed during reaction declaration), this returns a default value: zero.

property slack: timedelta

The remaining slack before the deadline (read-only).

This is the remaining wall-clock duration before the deadline (the deadline minus wall-clock now). Equivalently, for a declared deadline duration D it is D minus lag: the lag and the slack always sum to D, so as the lag grows during the handler the slack shrinks by the same amount. The slack denotes how much further the lag may grow before the deadline is missed. A negative value means the deadline has already been missed: the wall clock has passed the deadline.

This is a reaction-scoped accessor and is only meaningful while the reaction handler executes; values read outside a handler must not be relied upon.

If no deadline was declared, or if the program is not executing (for example, when accessed during reaction declaration), datetime.timedelta.max is returned.

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.

Attributes apply per key. Each attribute is added only if its key has not been set before on the same element. Attributes with rejected keys do not prevent the remaining attributes from being added.

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 any attribute key was added before.

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

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

connect(from_: InputPort[T], to: InputPort[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.

property name: str

Name of the element (read-only).

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

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.

Attributes apply per key. Each attribute is added only if its key has not been set before on the same element. Attributes with rejected keys do not prevent the remaining attributes from being added.

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 any 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.ShutdownEffect

Allows a reaction to terminate the program.

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

trigger_shutdown() → None

Terminate the currently running reactor program.

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

Raises:

ValidationError – If called before execution has started.

class xronos.Startup

A reactor element that emits an event when the 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.

Attributes apply per key. Each attribute is added only if its key has not been set before on the same element. Attributes with rejected keys do not prevent the remaining attributes from being added.

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 any 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 currently present.

class xronos.TriggerStatus(*values)

The outcome of a PhysicalEvent.trigger() attempt.

ACCEPTED = 0

The runtime accepted the event and it is queued for processing.

NOT_STARTED = 1

The program has not yet started and the event was dropped.

STOPPED = 2

The program has stopped executing and the event was dropped.

UNKNOWN = 3

The runtime reported a status that this SDK version does not recognize; the event was dropped.

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, ReactionContext], Callable[[], None]]) → ReactionDescriptor[R]

Decorator that is used to declare reactions.

Parameters:

declaration (Callable[[R, ReactionContext], Callable[[], None]]) – The decorated method. Must accept a ReactionContext 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.

To declare a reaction with a deadline, use @reaction_with_deadline instead.

xronos.reaction_with_deadline(*, deadline: timedelta) → Callable[[Callable[[R, ReactionContext], Callable[[], None]]], ReactionDescriptor[R]]

Decorator that is used to declare reactions with a deadline.

Parameters:

deadline (timedelta) – A deadline relative to the current time of the triggering event. The deadline is missed if the handler does not complete within this duration.

Raises:

ValueError – If deadline is negative.

To declare a reaction without a deadline, use @reaction instead. The decorated method has the same requirements as for @reaction.