hololinked.core.events.Event

class Event:
    """
    Asynchronously push arbitrary messages to clients without the client requesting the data every time.

    Asynchronous as-in messages that cannot be properly timed, not necessary `async`. Events are pushed from the server
    to the clients that have subscribed to them.
    """

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

    def __init__(
        self,
        doc: str | None = None,
        schema: JSON | type[BaseModel] | None = None,
        label: str | None = None,
    ) -> None:
        """
        Initialize an event.

        Parameters
        ----------
        doc: str
            docstring for the event
        schema: JSON | type[BaseModel]
            schema of the event data, either a JSON schema or a pydantic model. Any other kind of schema is
            accepted as well, as long as a validator that matches it is registered with `SchemaValidators`.
        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:
            SchemaValidators.check_schema(schema)
        self.schema = schema
        self.label = label
        self._observable = False

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

    @overload
    def __get__(self, obj: None, objtype: ThingMeta | None = None) -> Event: ...

    @overload
    def __get__(self, obj: Thing, objtype: ThingMeta | None = None) -> EventDispatcher: ...

    def __get__(self, obj: Thing | None, objtype: ThingMeta | None = None):
        try:
            if not obj:
                return self
            return EventDispatcher(
                unique_identifier=self.get_unique_identifier(obj),
                publisher=obj.eventloop.event_bus if obj.eventloop 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 get_unique_identifier(self, owner_inst: Thing) -> str:
        """
        An instance bound unique identifier string of the event.

        Parameters
        ----------
        owner_inst: Thing
            the instance whose copy of this event is meant

        Returns
        -------
        str
            the identifier, as `<thing id>/<event name>`
        """
        return f"{owner_inst._qualified_id}/{self.name}"

    def to_metadata(self, owner_inst: Thing | None = None, format: str = "wot") -> EventMetadata:
        """
        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 hololinked import MetadataFormats

        return MetadataFormats.get(format).event.from_descriptor(self, owner_inst or self.owner)