— INFORMATION FOR DEVELOPERS —

Creating sequential blocks

Topics to be considered during the planning phase:

  • what fully defines the internal state?

  • how will be the internal state initialized?

  • how to calculate the output from the internal state?

  • what events will be accepted?

  • what data is expected and which value should be returned for each particular event?

Checklist for creating a new SBlock:

SBlock initialization

The goal is to set the internal state and the output.

Let’s recap the initialization order with links to corresponding sections added. Each block defines only those steps that are appropriate to its functionality.

  1. from persistent data

    see: AddonPersistence and SBlock._restore_state()

  2. asynchronous initialization routine

    see: AddonAsync and SBlock.init_async()

  3. regular initialization routine

    see SBlock.init_regular()

  4. from the initdef value

    see SBlock.init_from_value()

Important

The general rule for all initialization functions: If it is not possible to initialize the block, leave it uninitialized and return. Do not raise on errors, only log a notice.

SBlock.get_state() → Any

Return the internal state. Raise an EdzedInvalidState when called before a successful initialization.

The default implementation assumes the state is equal to the output.

This method must be redefined for more complex SBlocks to return the real internal state.

It is recommended that this method produces JSON serializable data, especially when the block supports persistent state. JSON serializable data can be stored or transferred with minimum difficulties.

SBlock.init_regular() → None

Initialize the internal state to a fixed value and set the output.

Define only if the block can be initialized this way.

SBlock.init_from_value(value: Any, /) → None

Initialize the internal state from the given value and set the output.

Define only if the block can be initialized this way.

Defining this method automatically enables SBlock's keyword argument initdef.

Event handling

Dispatching events to handlers

Taking delivery of an event by its destination block is implemented in the SBlock.event() method.

SBlock.event(etype: str | EventType, /, **data) → Any

Handle the incoming event of type etype with attached data by dispatching it to the respective handler. Return the handler’s exit value. Evaluate conditional events (see the EventCond) before further processing. Raise EdzedUnknownEvent if there is no handler for the etype.

event() may return a value of any type. It is useful as a reply to an external event, but it is unused in the case of internal events. Other circuit blocks ignore the returned value.

Warning

If an exception other than an unknown event type or a trivial parameter error is raised during the event handling, the simulation will be aborted even if the caller catches the exception with a try-except construct.

Event handlers

There are two ways to define handlers for incoming events:

  1. Add specialized event handlers.

SBlock._event_ETYPE(**data: Any) → Any

If a method with matching event type 'ETYPE' is defined, it will be called to handle that event type.

For example: _event_put() will be called for all 'put' events.

Note

This way a non-FSM event can be added to an FSM block if need be. Take care not to interfere with the FSM operations.

Customize the method signature to extract the expected event data. Always accept unused additional data (**_data in the examples below). Examples:

# event 'dec' accepts optional data item 'amount'
def _event_dec(self, *, amount=1, **_data):
   ...

# event 'put' requires data item 'value'
def _event_put(self, *, value, **_data):
   ...

  1. Utilize the default event handler.

SBlock._event(etype: str | EventType, data: Mapping[str, Any]) → Any

_event() will be called for events without a specialized event handler.

Raise EdzedUnknownEvent if the etype is not supported. event() may return a response of any type. It has a significance only when interfacing with an external system. Other circuit blocks ignore the returned vale.

Example:

def _event(self, etype, data):
    if etype == 'ying':
        # handle event ying here
        return None

    if etype == 'yang':
        # handle event yang here
        return None

    # let the parent handle everything else,
    # the base class simply raises the EdzedUnknownEvent
    return super()._event(event, data)

Note: Do not confuse this event handler _event() with the event dispatcher event(). The latter should be left untouched.

Note

Note the different ways the event data is passed to a handler (it is intentional):

def _event_ETYPE(self, **data):  # as keyword args
def _event(self, etype, data):   # as a dict

Important

A block must ignore any additional data items.

Setting the output

Event handlers and initialization functions manage the internal state and the output value. The output setter is:

SBlock.set_output(value: Any) → None

Set the output value. The value must not be UNDEF.

A block is deemed initialized when its output value changes from UNDEF to any other value. i.e. after the first set_output() call.

Generating events

The recommended way to implement events to be sent when some trigger has fired is to add an on_TRIGGER keyword argument:

class ExampleBlock(edzed.SBlock):

    def __init__(self, *args, on_bang=None, **kwargs):
        self._bang_events = edzed.event_tuple(on_bang)
        super().__init__(*args, **kwargs)

    def _send_bang_events(self):
        """Notify all recipients that a 'bang' has occurred."""
        for event in self._bang_events:
            event.send(self, trigger='bang')  # add event data as needed
edzed.event_tuple(arg: None | Event | Sequence[Event]) → tuple[Event, ...]

A helper supporting multiple ways to specify event(s).

Convert None, a single Event object or a sequence of Event objects to a tuple with zero, one or more events.

Event.send(source: Block, /, **data) → bool

Apply filters, add source item to the data and send this event.

Return True if the event was sent, False if rejected by a filter. The return value may be important as a response to external events, but the internal block-to-block events are one-way communication and the return value from the SBlock.event() is disregarded by sender blocks.

Parameter source specifies the sender block. The destination is given by the Event. The purpose of data (given as name=value keyword arguments) is to describe the event. Details depend on the particular sender and event type.

'source': <name of sender block> item is added to the event data.

Blocks provided by edzed also include 'trigger': <trigger name> where the trigger name is derived from the corresponding parameter mainly by stripping the 'on_' prefix.

Start and stop

SBlock.start() is called when the circuit simulation is about to start, before the block initialization; SBlock.stop() is called when the circuit simulation has finished.

SBlock.start() → None

Pre-simulation hook.

Set up resources necessary for proper function of the block. Do not set the internal state here.

Important

When using start(), always call the super().start().

Note

Why do we need start() when we have __init__()?

Only the start() is the right place for actions that:

  • have a side effect, or

  • are resource intensive (time, memory, CPU), or

  • require an asyncio event loop

What we want to achieve is that blocks may be created at import time, i.e. defined at the module level. Importing such module should not have any negative effects.

SBlock.stop() → None

Post-simulation hook, a counterpart of start().

stop() is dedicated for cleanup actions. It is called when the circuit simulation has finished, but only if start() was successfully called.

An exception in stop() will be logged, but otherwise ignored.

Important

When using stop(), always call the super().stop()

Add-ons

Important

In the list of new block’s bases always put the add-on classes before the SBlock:

class NewBlock(edzed.AddonPersistence, edzed.SBlock): ...

Persistent state add-on

class edzed.AddonPersistence

Inheriting from this class adds support for state persistence. The related arguments persistent, sync_state, and expiration are explained in the SBlock's documentation.

If enabled, the block’s internal state (as returned by SBlock.get_state() is saved to the persistent storage provided by the circuit in these situations:

  • when SBlock.save_persistent_state() is called

  • at the end of a simulation

  • by default also after each event; this can be disabled with sync_state keyword argument.

Saving of persistent state is disabled after an error in SBlock.event() in order to prevent saving of possibly corrupted state.

For state restoration _restore_state() must be implemented.

The simulator retrieves the saved state from the persistent storage, then it checks the expiration time and unless the state has expired, it is passed to _restore_state().

SBlock.save_persistent_state() → None

Save the internal state to persistent storage.

This method is usually called by the simulator.

abstractmethod SBlock._restore_state(state: Any, /) → None

Initialize by restoring the state (presumably created by get_state()) and the corresponding output.

Note that _restore_state() is sometimes identical with SBlock.init_from_value().

SBlock.key: str

The persistent dict key associated with this block. It equals the string representation str(self), but this is an implementation detail that may change in the future.

Async add-on

class edzed.AddonAsync

Inheriting from this class adds asynchronous support, in particular asynchronous initialization and asynchronous cleanup.

Note

If the only function to be implemented is an async initialization, consider using an InitAsync helper block instead.

This class also implements a helper for general use:

async _create_monitored_task(coro: Awaitable, is_service: bool = False, **task_kwargs) → asyncio.Task

Create a task monitored for an eventual failure.

Important

For the purpose of monitoring, the cancellation is never considered an error. Note that the CancelledError exception is not derived from the Exception class (it’s a BaseException subclass) and as such it is not caught by the monitor.

This function acts like asyncio.create_task(), but if the task exits due to an exception, the simulation will abort. _create_monitored_task() also adds the block name to the exception notes (Python >= 3.11) or to the exception message (Exception.args[0] if it is a string) for better problem identification.

Coroutines marked as services (is_service is True) are supposed to run until cancelled - even a normal exit is treated as an error.

Extra keyword arguments **task_kwargs are passed to the asyncio.create_task(); as of Python 3.11 it accepts name and context.

async SBlock.init_async() → None

Optional async initialization coroutine, define only when needed.

The async initialization is intended to interact with external systems and as such should be utilized solely by circuit inputs.

The existence of this method automatically enables the init_timeout SBlock keyword argument.

init_async() is run as a task and is waited for init_timeout seconds. When a timeout occurs, the task is cancelled and the initialization continues with the next step.

Implementation detail: The simulator may wait longer than specified if it is also concurrently initializing another AddonAsync based block with a longer init_timeout.

Important

Should an event arrive during the async initialization, the block will get a regular synchronous initialization in order to be able to process the event immediately. For this reason, when init_async() asynchronously obtains the initialization value, it should check whether the block is still uninitialized before applying the value.

async SBlock.stop_async() → None

Optional async cleanup coroutine, define only when needed.

The existence of this method automatically enables the stop_timeout SBlock keyword argument.

This coroutine is awaited after the regular stop().

stop_async() is run as a task and is waited for stop_timeout seconds. When a timeout occurs, the task is cancelled. The simulator logs the error and continues the cleanup.

Tip

Use utils.shield_cancel() to protect small critical task sections from immediate cancellation.

Main task add-on

class edzed.AddonMainTask(*args, **block_kwargs, **task_kwargs)

A subclass of AddonAsync. In addition to AddonAsync's features, inheriting from this add-on adds support for a task automatically running from simulation start to stop.

The add-on manages everything necessary including task monitoring. If the task terminates before stop, the simulation will be aborted.

Adjust the stop_timeout (SBlock's argument) if necessary. Note that the init_timeout argument does not apply, because task creation is a regular function.

All keyword arguments starting with "task_" are passed to asyncio.create_task() with the "task_" prefix removed. Example "task_name="data acquisition task".

Note: the illustrative signature in the heading is not syntactically correct.

abstractmethod async AddonMainTask._maintask()

The task coroutine.

Async initialization add-on

class edzed.AddonAsyncInit

There are blocks like the ValuePoll lacking any initialization code, because they are collecting data as their main job and the first value they obtain automatically becomes the initialization value.

This tiny add-on adds a SBlock.init_async() implementation that ensures proper interfacing with the simulator in such cases. Its only function is to inform the simulator that the initialization took place.

Presence of this add-on modifies the behavior of AddonMainTask.

Note about eager tasks in asynchronous add-ons

Python 3.12 introduced eagerly started asyncio tasks. This feature changes the task execution order and that sometimes matters during the circuit start.

Normally, the developer decides whether eager tasks will be used and will write the code accordingly, but sometimes an edzed application needs to support both cases. Following functions may be affected:

These measures can be taken if necessary:

  • Python 3.14 added a new option eager_start that you can add to task_kwargs in and AddonMainTask (as task_eager_start in the latter case) and thus override the default task factory setting.

  • Consider using a statement that re-orders the task execution. Put it at the top of the coroutine. This is what AddonMainTask does when used without AddonAsyncInit:

    circuit = edzed.get_circuit()
await circuit.wait_init()       # wait until initialization is completed

    The moment when await returns from wait_init() does not depend on the method the task was started.

Helper methods

When creating new blocks, you may find these methods useful:

Block.log_msg(msg: str, *args, level: int, **kwargs) → None

Log a message.

The block name is prepended to the msg and then the arguments are passed to logging.log() with the given level.

Block.log_debug(*args, **kwargs) → None
Block.log_info(*args, **kwargs) → None
Block.log_warning(*args, **kwargs) → None
Block.log_error(*args, **kwargs) → None

Log a debug/info/warning/error message respectively.

A debug message is logged only if debug messages are enabled for the block.

These are simple Block.log_msg() wrappers.

Example (Input)

An input block like Input, but without data validation:

class Input(edzed.AddonPersistence, edzed.SBlock):
    def init_from_value(self, value):
        self.event('put', value=value)

    def _event_put(self, *, value, **_data):
        self.set_output(value)
        return True

    _restore_state = init_from_value