Module omnipy.hub.entry

Overview

View Source

from dataclasses import dataclass, field

from omnipy.api.protocols.public.hub import IsRuntime

from omnipy.util.publisher import DataPublisher

@dataclass

class RuntimeEntryPublisher(DataPublisher):

    _back: IsRuntime | None = field(default=None, init=False, repr=False)

    def __setattr__(self, key, value):

        super().__setattr__(key, value)

        if hasattr(self, key) and not key.startswith('_') and self._back is not None:

            self._back.reset_subscriptions()

Classes

RuntimeEntryPublisher

class RuntimeEntryPublisher(

)

RuntimeEntryPublisher()

View Source

@dataclass

class RuntimeEntryPublisher(DataPublisher):

    _back: IsRuntime | None = field(default=None, init=False, repr=False)

    def __setattr__(self, key, value):

        super().__setattr__(key, value)

        if hasattr(self, key) and not key.startswith('_') and self._back is not None:

            self._back.reset_subscriptions()

---

Methods

eq

def __eq__(
    self,
    other
)

Return self==value.

Parameters:

Name	Type	Description	Default
other

setattr

def __setattr__(
    self,
    key,
    value
)

Implement setattr(self, name, value).

Parameters:

Name	Type	Description	Default
key
value

View Source

    def __setattr__(self, key, value):

        super().__setattr__(key, value)

        if hasattr(self, key) and not key.startswith('_') and self._back is not None:

            self._back.reset_subscriptions()

subscribe

def subscribe(
    self,
    config_item: str,
    callback_fun: Callable[[Any], NoneType]
)

Parameters:

Name	Type	Description	Default
config_item	str
callback_fun	Callable[[Any], NoneType]

View Source

    def subscribe(self, config_item: str, callback_fun: Callable[[Any], None]):

        if not hasattr(self, config_item):

            raise AttributeError(f'No config items named "{config_item}"')

        elif config_item.startswith('_'):

            raise AttributeError(f'Subscribing to private member "{config_item}" not allowed')

        else:

            self._subscriptions[config_item].append(callback_fun)

            callback_fun(getattr(self, config_item))

unsubscribe_all

def unsubscribe_all(
    self
) -> None

Returns:

Type	Description
NoneType

View Source

    def unsubscribe_all(self) -> None:

        self._subscriptions = _subscribers_factory()