pipert.core.routine

Module Contents

Classes

Events	Events that are fired by the RoutineInterface during
State	An object that is used to pass internal and user-defined state between
RoutineTypes	Every routine will have a type
Routine	Helper class that provides a standard way to create an ABC using

class pipert.core.routine.Events

Bases: enum.Enum

Events that are fired by the RoutineInterface during execution.

BEFORE_LOGIC = before_logic

AFTER_LOGIC = after_logic

EXCEPTION_RAISED = exception_raised

class pipert.core.routine.State

Bases: object

An object that is used to pass internal and user-defined state between event handlers.

class pipert.core.routine.RoutineTypes

Bases: enum.Enum

Every routine will have a type

NO_TYPE

INPUT = 0

PROCESSING = 1

OUTPUT = 2

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

Bases: abc.ABC

Helper class that provides a standard way to create an ABC using inheritance.

routine_type

_setup_extensions(self, extensions)

register_events(self, *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)

add_event_handler(self, 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)

has_event_handler(self, 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.

remove_event_handler(self, 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.

_extension_log_fps(self, fps_time_interval)

Log the fps of routine every fps_time_interval seconds

Args:

fps_time_interval: Interval time between each log.

_extension_pace(self, fps)

Pace the routine to work at a wanted fps

Args:

fps: The wanted fps for the routine

on(self, 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.

_fire_event(self, event_name, *event_args, **event_kwargs)

Execute all the handlers associated with given event.

This method executes all handlers associated with the event event_name. Optional positional and keyword arguments can be used to pass arguments to all handlers added with this event. These aguments updates arguments passed using add_event_handler().

Args:

event_name: event for which the handlers should be executed. Valid

events are from Events or any event_name added by

register_events().

*event_args: additional args to be passed to all handlers. **event_kwargs: additional keyword args to be passed to all handlers.

abstract main_logic(self, *args, **kwargs)

abstract setup(self, *args, **kwargs)

abstract cleanup(self, *args, **kwargs)

_extended_run(self)

Returns:

as_thread(self)

as_process(self)

start(self)

abstract static get_constructor_parameters()

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

abstract does_routine_use_queue(self, queue_name)

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

queue_name: the name of the queue

get_creation_dictionary(self)

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.