Routine

A routine is responsible for performing one of the component’s main tasks. It can be run either as a thread or as a process. First it runs a setup function, then it runs its main logic function in a continuous loop (until it is told to terminate), and finally it runs a cleanup function. The routines of a component use queues in order to pass the data between them.

Routines can also register events (and event handlers) which can be triggered at any point. By default, each routine registers 2 events which are triggered at the beginning and at the end of each iteration of the routine’s main logic loop. Each routine can implement its own handlers for the events.

class pipert.core.routine.Routine(logger, name='', component_name='', extensions=None, metrics_collector=<pipert.core.metrics_collector.NullCollector object>, *args, **kwargs)

add_event_handler(event_name, handler, first=False, last=False, *args, **kwargs)

Add an event handler to be executed when the specified event is fired.

Args:

event_name: An event to attach the handler to. Valid events are from Events or any event_name added by

register_events().

handler (callable): the callable event handler that should be invoked first: specify ‘true’ if the event handler should be called first last: specify ‘true’ if the event handler should be called last *args: additional args to be passed to handler. **kwargs: additional keyword args to be passed to handler.

Notes:

The handler function’s first argument will be self, the Routine object it was bound to.

Note that other arguments can be passed to the handler in addition to the *args and **kwargs passed here, for example

during EXCEPTION_RAISED.

Example usage:

routine = Routine(process_function)

def print_epoch(routine):
    print("Epoch: {}".format(routine.state.epoch))

routine.add_event_handler(Events.EPOCH_COMPLETED, print_epoch)

abstract does_routine_use_queue(queue_name)

Returns True whether the routine uses the given queue_name. Args:

queue_name: the name of the queue

abstract static get_constructor_parameters()

Returns a dictionary of the constructor’s parameters built as key for name and value for type name

get_creation_dictionary()

Returns a dictionary containing the routine parameters name as keys and their values as values. The method return queue objects instead of queue names when encountering them.

has_event_handler(handler, event_name=None)

Check if the specified event has the specified handler.

Args:

handler (callable): the callable event handler. event_name: The event the handler attached to. Set this

to None to search all events.

on(event_name, *args, **kwargs)

Decorator shortcut for add_event_handler.

Args:

event_name: An event to attach the handler to. Valid events are from Events or any event_name added by

register_events().

*args: additional args to be passed to handler. **kwargs: additional keyword args to be passed to handler.

register_events(*event_names)

Add events that can be fired.

Registering an event will let the user fire these events at any point. This opens the door to make the run() loop even more configurable.

By default, the events from Events are registerd.

Args:

*event_names: An object (ideally a string or int) to define the

name of the event being supported.

Example usage:

from enum import Enum

class Custom_Events(Enum):
    FOO_EVENT = "foo_event"
    BAR_EVENT = "bar_event"

routine = Routine(process_function)
routine.register_events(*Custom_Events)

remove_event_handler(handler, event_name)

Remove event handler handler from registered handlers of the routine

Args:

handler (callable): the callable event handler that should be removed event_name: The event the handler attached to.