hololinked.core.meta.EventSource

Base class to add event functionality to an object, it is not meant to be subclassed directly by the end-user.

UML Diagram

Source code in hololinked\core\meta.py

class EventSource:
    """
    Base class to add event functionality to an object, 
    it is not meant to be subclassed directly by the end-user.

    [UML Diagram](http://localhost:8000/UML/PDF/Thing.pdf)
    """

    id : str

    def __init__(self) -> None:
        self._event_publisher = None # type: typing.Optional["EventPublisher"]
        self.create_events_registry()

    # creating name without underscore causes clash with the metaclass method 
    # with same name
    def create_events_registry(self) -> None:
        """creates a registry for available `Events` based on `EventsRegistry`"""
        self._events_registry = EventsRegistry(self.__class__, self)

    @property
    def events(self) -> EventsRegistry:
        """container for the event descriptors of the object."""
        return self._events_registry

    @property
    def event_publisher(self) -> "EventPublisher":
        """
        event publishing object `EventPublisher` that owns the zmq.PUB socket, valid only after 
        creating an RPC server or calling a `run()` method on the `Thing` instance.
        """
        return self._event_publisher 

    @event_publisher.setter           
    def event_publisher(self, value: "EventPublisher") -> None:
        from .thing import Thing

        if self._event_publisher is not None:
            raise AttributeError("Can set event publisher only once")
        if value is None:
            return 

        def recusively_set_event_publisher(obj : Thing, publisher : "EventPublisher") -> None:
            for name, evt in inspect._getmembers(obj, lambda o: isinstance(o, Event), getattr_without_descriptor_read):
                assert isinstance(evt, Event), "object is not an event"
                # above is type definition
                evt._publisher = publisher
            for name, subobj in inspect._getmembers(obj, lambda o: isinstance(o, Thing), getattr_without_descriptor_read):
                if name == '_owner':
                    continue 
                recusively_set_event_publisher(subobj, publisher)
            obj._event_publisher = publisher            

        recusively_set_event_publisher(self, value)

Attributes

event_publisher property writable

event_publisher: EventPublisher

event publishing object EventPublisher that owns the zmq.PUB socket, valid only after creating an RPC server or calling a run() method on the Thing instance.

Functions

create_events_registry

create_events_registry() -> None

creates a registry for available Events based on EventsRegistry

Source code in hololinked\core\meta.py

def create_events_registry(self) -> None:
    """creates a registry for available `Events` based on `EventsRegistry`"""
    self._events_registry = EventsRegistry(self.__class__, self)