Skip to content

hololinked.core.events.Event

Asynchronously push arbitrary messages to clients. Apart from default events created by the package (like state change event, observable properties etc.), events are supposed to be created at class level or at __init__ as a instance attribute, otherwise their publishing socket is unbound and will lead to AttributeError.

Parameters:

Name Type Description Default

name

name of the event, specified name may contain dashes and can be used on client side to subscribe to this event.

 required

doc

Optional[str]

docstring for the event

 None

schema

Optional[JSON]

schema of the event, if the event is JSON complaint. HTTP clients can validate the data with this schema. There is no validation on server side.

 None
Source code in .venv/lib/python3.13/site-packages/hololinked/core/events.py 
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
class Event:
    """
    Asynchronously push arbitrary messages to clients. Apart from default events created by the package (like state
    change event, observable properties etc.), events are supposed to be created at class level or at `__init__` 
    as a instance attribute, otherwise their publishing socket is unbound and will lead to `AttributeError`.  

    Parameters
    ----------
    name: str
        name of the event, specified name may contain dashes and can be used on client side to subscribe to this event.
    doc: str
        docstring for the event
    schema: JSON
        schema of the event, if the event is JSON complaint. HTTP clients can validate the data with this schema. There
        is no validation on server side.
    """
    # security: Any
    #     security necessary to access this event.

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


    def __init__(self, 
                doc : typing.Optional[str] = None, 
                schema : typing.Optional[JSON] = None, # security : typing.Optional[BaseSecurityDefinition] = None,
                label : typing.Optional[str] = None
            ) -> None:
        self.doc = doc 
        if global_config.VALIDATE_SCHEMAS and schema:
            jsonschema.Draft7Validator.check_schema(schema)
        self.schema = schema
        # self.security = security
        self.label = label
        self._observable = False

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

    @typing.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._internal_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 = None):
        from ..td import EventAffordance
        return EventAffordance.generate(self, owner_inst or self.owner)