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.

exception xronos.DuplicateNameError

Exception that is thrown when creating an element with a name that is already in use.

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.

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_: 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 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 trigger_shutdown() on a ShutdownEffect.

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.

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

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 (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.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 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

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

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

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

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 (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 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]

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

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.

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 – that is, 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 violated. A negative value means the deadline has 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.ReactionInterface

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 @reaction or @reaction_with_deadline.

Deprecated since version v0.12.0: ReactionInterface will be removed in an upcoming release. Use ReactionContext instead, which additionally provides the reaction-scoped timing API. ReactionInterface remains the base class of ReactionContext, so existing annotations keep working.

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]

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_: 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.

get_lag() → timedelta

Get the current lag.

Deprecated since version v0.12.0: The current time is only well-defined while a reaction handler executes. This method will be removed in an upcoming release; use the reaction-scoped ReactionContext.lag instead.

The lag is the difference between wall-clock time and the current time (wall-clock now minus the current time). While a reaction handler executes, the current time does not advance, but the wall clock does; the lag therefore measures how far the wall clock has run ahead of the internal clock – that is, how far the execution of reactions lags behind the events it processes.

Returns:: The current lag as a wall-clock duration.
Return type:: timedelta

get_time() → datetime

Get the current time.

Deprecated since version v0.12.0: The current time is only well-defined while a reaction handler executes. This method will be removed in an upcoming release; use the reaction-scoped ReactionContext.current_time instead.

Note

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

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

get_time_since_startup() → timedelta

Get how far the internal clock has advanced since startup.

Deprecated since version v0.12.0: The current time is only well-defined while a reaction handler executes. This method will be removed in an upcoming release; use the reaction-scoped ReactionContext.elapsed_time instead.

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

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.

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

class xronos.Startup

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, 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. Annotating the argument as the deprecated ReactionInterface is still accepted.

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