Module omnipy.util.publisher
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
DataPublisher
DataPublisher()
Methods
eq
Return self==value.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
other |
setattr
Implement setattr(self, name, value).
Parameters:
Name | Type | Description | Default |
---|---|---|---|
key |
|||
value |
View Source
def __setattr__(self, key, value):
super().__setattr__(key, value)
if key in self._subscriptions:
for callback_fun in self._subscriptions[key]:
callback_fun(value)
subscribe
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
Returns:
Type | Description |
---|---|
NoneType |
View Source
def unsubscribe_all(self) -> None:
self._subscriptions = _subscribers_factory()