Components, And All That

Overview: Components, Facets, Receptacles

Components are objects that implement functionality. Access to that funtionality is provided via stubs, so-called facets.

A component may use functionality of another component via its facets. For this, you connect together both components by plugging one component’s facet into a compatible receptacle of the using component.

from endless.framework.component import Component
from endless.framework.facet import facet
from endless.framework.receptacle import receptacle, ONE

import pytest
import abc


def test_connected_components():
    class SimpleCalculator(abc.ABC):
        @abc.abstractmethod
        def add(self, lhs, rhs):
            raise NotImplementedError
        @abc.abstractmethod
        def sub(self, lhs, rhs):
            raise NotImplementedError

    class SuperStatistics(abc.ABC):
        @abc.abstractmethod
        def add(self, value):
            raise NotImplementedError
        @abc.abstractmethod
        def get_accumulated_value(self):
            raise NotImplementedError
        @abc.abstractmethod
        def get_average_value(self):
            raise NotImplementedError

    @facet('calculator', SimpleCalculator,
           (('add', '_add'),    # SimpleCalculator.add -> MyCalculatingThing._add
            ('sub', '_sub'),    # SimpleCalculator.sub -> MyCalculatingThing._sub
            ))  
    @facet('statistics', SuperStatistics, 
           (('add', '_present_value'),                      # SuperStatistics.add -> MyCalculatingThing._present_value
            ('get_accumulated_value', '_get_accumulated'),  # SuperStatistics.get_accumulated_value -> MyCalculatingThing._get_accumulated
            ('get_average_value', '_get_running_average'),  # SuperStatistics.get_average_value -> MyCalculatingThing._get_running_average
            ))
    class MyCalculatingThing(Component):
        def __init__(self):
            super().__init__()
            self._accumulated_value = 0
            self._running_average = 0
            self._nvalues = 0

        def _add(self, lhs, rhs):
            return lhs + rhs
        def _sub(self, lhs, rhs):
            return lhs - rhs

        def _present_value(self, value):
            self._accumulated_value += value
            if self._nvalues == 0:
                self._running_average = value
            else:
                self._running_average = (self._running_average * self._nvalues + value) / (self._nvalues + 1)
            self._nvalues += 1

        def _get_accumulated(self):
            return self._accumulated_value
        def _get_running_average(self):
            return self._running_average

    @receptacle('statistical_outlet', SuperStatistics, multiplicity=ONE)
    class ValueProducer(Component):
        '''Produces values from an iterable, and injects it into
        another component that is connected via the
        ``statistical_outlet`` receptacle'''

        def __init__(self, values):
            super().__init__()
            self._iter = iter(values)
        def produce_next_value(self):
            '''Produce next value, and inject it into receptacle'''
            self._statistical_outlet.add(next(self._iter))

    my_values = ValueProducer([1,5,3,4])
    my_calc_thing = MyCalculatingThing()

    # connect both together (using only statistics functionality,
    # disregarding calculator)

    my_values.statistical_outlet.connect(my_calc_thing.statistics)

    # component "pipeline": produce values, and check flow into
    # statistics
    my_values.produce_next_value()      # <-- 1
    assert my_calc_thing.statistics.get_accumulated_value() == 1
    assert my_calc_thing.statistics.get_average_value() == 1

    my_values.produce_next_value()      # <-- 5
    assert my_calc_thing.statistics.get_accumulated_value() == 6
    assert my_calc_thing.statistics.get_average_value() == pytest.approx(3)

    my_values.produce_next_value()      # <-- 3
    assert my_calc_thing.statistics.get_accumulated_value() == 9
    assert my_calc_thing.statistics.get_average_value() == pytest.approx(3)

    my_values.produce_next_value()      # <-- 4
    assert my_calc_thing.statistics.get_accumulated_value() == 13
    assert my_calc_thing.statistics.get_average_value() == pytest.approx(3.25)

    # use calculator facet directly
    sum = my_calc_thing.calculator.add(1, 2)
    diff = my_calc_thing.calculator.sub(1, 2)

    assert sum == 3
    assert diff == -1
            

Component Base Class

class endless.framework.component.Component

Base class for components.

Provides basic functionality that components inherit.

• Components are guaranteed to have an errorhandler, via the errorhandler property

• Components usually have facets and receptacles

__init__()

__weakref__

list of weak references to the object

Defining Facets And Receptacles

class endless.framework.facet.facet(name, basetype, methodspec)

Class decorator for component classes.

@facet('facetname', FacetInterface, (('method1', '_implementation_method1'), ('method2', '_implementation_method2')))
class MyComponent(Component):
    def _implementation_method1(self, ...):
        # do something
        pass
    def _implementation_method2(self, ...):
        # do something
        pass

This creates a property facetname on MyComponent instances, of base type FacetInterface. FacetInterface.method1 is implemented as a simple trampoline that calls into the associated component object’s _implementation_method1.

With this, the component can be used as follows,

my_comp = MyComponent()
my_comp.facetname.method1('blah')

__call__(component_class)

Call self as a function.

__init__(name, basetype, methodspec)

__weakref__

list of weak references to the object

class endless.framework.receptacle.receptacle(name, basetype, multiplicity)

__call__(component_class)

Call self as a function.

__init__(name, basetype, multiplicity)

__weakref__

list of weak references to the object

Running Components: Lifetime and Error Management

class endless.framework.component.LifetimeComponent(func)

Base class for components whose lifetime needs to be managed.

__init__(func)

start(taskgroup)

Create task

stop()

Cancel task

class endless.framework.runner.Runner(components, errorhandler=None)

class endless.framework.errorhandler.ErrorHandler

Abstract base class for error handlers

__weakref__

list of weak references to the object

abstract async report_exception(exc)

Report an exception

Parameters:

exc – Exception