Skip to content

hololinked.core.events.Event

Asynchronously push arbitrary messages to clients (as-in messages that cannot be properly timed) without the client requesting the data every time. Events are pushed from the server to the clients that have subscribed to them.

Source code in hololinked/hololinked/core/events.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
class Event:
    """
    Asynchronously push arbitrary messages to clients (as-in messages that cannot be properly timed) without
    the client requesting the data every time. Events are pushed from the server to the clients
    that have subscribed to them.
    """

    __slots__ = ["name", "_internal_name", "_publisher", "_observable", "doc", "schema", "label", "owner"]

    def __init__(
        self,
        doc: str | None = None,
        schema: JSON | None = None,
        label: str | None = None,
    ) -> None:
        """
        Parameters
        ----------
        doc: str
            docstring for the event
        schema: JSON
            schema of the event
        label: str
            a descriptive label for the event, to be shown in a GUI for example.
        """
        self.doc = doc
        if global_config.VALIDATE_SCHEMAS and schema:
            jsonschema.Draft7Validator.check_schema(schema)
        self.schema = schema
        self.label = label
        self._observable = False

    def __set_name__(self, owner: ParameterizedMetaclass, name: str) -> None:
        self.name = name
        self.owner = owner

    @overload
    def __get__(self, obj, objtype) -> "EventDispatcher": ...

    def __get__(self, obj: Parameterized, objtype: ParameterizedMetaclass = None):
        try:
            if not obj:
                return self
            # uncomment for type hinting
            # from .thing import Thing
            # assert isinstance(obj, Thing)
            return EventDispatcher(
                unique_identifier=f"{obj._qualified_id}/{self.name}",
                publisher=obj.rpc_server.event_publisher if obj.rpc_server else None,
                owner_inst=obj,
                descriptor=self,
            )
        except KeyError:
            raise AttributeError(
                "Event object not yet initialized, please dont access now." + " Access after Thing is running."
            )

    def to_affordance(self, owner_inst: Any = None):
        """
        Generates a `EventAffordance` TD fragment for this Event

        Parameters
        ----------
        owner_inst: Thing, optional
            The instance of the owning `Thing` object. If not supplied, the class is used.

        Returns
        -------
        EventAffordance
            the affordance TD fragment for this event
        """
        from ..td import EventAffordance

        return EventAffordance.generate(self, owner_inst or self.owner)

Functions

__init__ 

__init__(doc: str | None = None, schema: JSON | None = None, label: str | None = None) -> None

Parameters:

Name Type Description Default

doc

str | None

docstring for the event

 None

schema

JSON | None

schema of the event

 None

label

str | None

a descriptive label for the event, to be shown in a GUI for example.

 None
Source code in hololinked/hololinked/core/events.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
def __init__(
    self,
    doc: str | None = None,
    schema: JSON | None = None,
    label: str | None = None,
) -> None:
    """
    Parameters
    ----------
    doc: str
        docstring for the event
    schema: JSON
        schema of the event
    label: str
        a descriptive label for the event, to be shown in a GUI for example.
    """
    self.doc = doc
    if global_config.VALIDATE_SCHEMAS and schema:
        jsonschema.Draft7Validator.check_schema(schema)
    self.schema = schema
    self.label = label
    self._observable = False

to_affordance 

to_affordance(owner_inst: Any = None)

Generates a EventAffordance TD fragment for this Event

Parameters:

Name Type Description Default

owner_inst

Any

The instance of the owning Thing object. If not supplied, the class is used.

 None

Returns:

Type Description
EventAffordance

the affordance TD fragment for this event
Source code in hololinked/hololinked/core/events.py 
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
def to_affordance(self, owner_inst: Any = None):
    """
    Generates a `EventAffordance` TD fragment for this Event

    Parameters
    ----------
    owner_inst: Thing, optional
        The instance of the owning `Thing` object. If not supplied, the class is used.

    Returns
    -------
    EventAffordance
        the affordance TD fragment for this event
    """
    from ..td import EventAffordance

    return EventAffordance.generate(self, owner_inst or self.owner)