Skip to content

omnipy.util.publisher

CLASS DESCRIPTION
DataPublisher
RuntimeEntryPublisher

DataPublisher

Bases: pyd.BaseModel

CLASS DESCRIPTION
Config
METHOD DESCRIPTION
deepcopy

Create a deep copy of the DataPublisher instance, resetting subscriptions in the copy.
subscribe
subscribe_attr
unsubscribe_all
Source code in src/omnipy/util/publisher.py 
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
84
85
86
87
88
89
90
91
92
class DataPublisher(pyd.BaseModel):
    class Config:
        arbitrary_types_allowed = True
        validate_assignment = True

    _self_subscriptions: list[Callable[..., None]] = pyd.PrivateAttr(default_factory=list)
    _attr_subscriptions: DefaultDict[str, list[Callable[..., None]]] = \
        pyd.PrivateAttr(default_factory=_subscribers_factory)

    def subscribe_attr(self, attr_name: str, callback_fun: Callable[..., None]):
        if not hasattr(self, attr_name):
            raise AttributeError(f'No attribute named "{attr_name}"')
        elif attr_name.startswith('_'):
            raise AttributeError(f'Subscribing to private member "{attr_name}" not allowed')
        else:
            self._attr_subscriptions[attr_name].append(callback_fun)
            attr = getattr(self, attr_name)
            callback_fun(attr)

            if isinstance(attr, DataPublisher):
                attr.subscribe(callback_fun, do_callback=False)

    def subscribe(self, callback_fun: Callable[..., None], do_callback: bool = True) -> None:
        self._self_subscriptions.append(callback_fun)

        def _get_attr_callback_for_publisher_child(
                attr_name: str) -> Callable[[DataPublisher], None]:
            def _attr_callback_for_publisher_child(value: object) -> None:
                callback_fun(self)

            return _attr_callback_for_publisher_child

        for attr_name in self.__class__.__fields__.keys():
            if not attr_name.startswith('_'):
                attr = getattr(self, attr_name)
                if isinstance(attr, DataPublisher):
                    attr.subscribe(
                        _get_attr_callback_for_publisher_child(attr_name), do_callback=False)

        if do_callback:
            callback_fun(self)

    def unsubscribe_all(self) -> None:
        self._self_subscriptions.clear()
        self._attr_subscriptions.clear()

        for attr_name in self.__class__.__fields__.keys():
            if not attr_name.startswith('_'):
                attr = getattr(self, attr_name)
                if isinstance(attr, DataPublisher):
                    attr.unsubscribe_all()

    def _call_subscribers(self, attr_name: str, value: object) -> None:
        if attr_name in self._attr_subscriptions:
            for callback_fun in self._attr_subscriptions[attr_name]:
                callback_fun(value)

    def _call_self_subscribers(self) -> None:
        for callback_fun in self._self_subscriptions:
            callback_fun(self)

    def _call_all_subscribers(self, attr_name: str, value: object) -> None:
        self._call_subscribers(attr_name, value)
        self._call_self_subscribers()

    def __setattr__(self, attr_name: str, value: object) -> None:
        super().__setattr__(attr_name, value)

        if not attr_name.startswith('_'):
            self._call_all_subscribers(attr_name, value)

    def deepcopy(self) -> Self:
        """
        Create a deep copy of the DataPublisher instance, resetting subscriptions in the copy.
        """
        self_copy = self.copy()
        self_copy._self_subscriptions = []
        self_copy._attr_subscriptions = _subscribers_factory()
        return self_copy.copy(deep=True)

Config

ATTRIBUTE DESCRIPTION
arbitrary_types_allowed

validate_assignment

Source code in src/omnipy/util/publisher.py 
15
16
17
class Config:
    arbitrary_types_allowed = True
    validate_assignment = True

arbitrary_types_allowed class-attribute instance-attribute 

arbitrary_types_allowed = True

validate_assignment class-attribute instance-attribute 

validate_assignment = True

deepcopy 

deepcopy() -> Self

Create a deep copy of the DataPublisher instance, resetting subscriptions in the copy.

Source code in src/omnipy/util/publisher.py 
85
86
87
88
89
90
91
92
def deepcopy(self) -> Self:
    """
    Create a deep copy of the DataPublisher instance, resetting subscriptions in the copy.
    """
    self_copy = self.copy()
    self_copy._self_subscriptions = []
    self_copy._attr_subscriptions = _subscribers_factory()
    return self_copy.copy(deep=True)

subscribe 

subscribe(callback_fun: Callable[..., None], do_callback: bool = True) -> None
Source code in src/omnipy/util/publisher.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
def subscribe(self, callback_fun: Callable[..., None], do_callback: bool = True) -> None:
    self._self_subscriptions.append(callback_fun)

    def _get_attr_callback_for_publisher_child(
            attr_name: str) -> Callable[[DataPublisher], None]:
        def _attr_callback_for_publisher_child(value: object) -> None:
            callback_fun(self)

        return _attr_callback_for_publisher_child

    for attr_name in self.__class__.__fields__.keys():
        if not attr_name.startswith('_'):
            attr = getattr(self, attr_name)
            if isinstance(attr, DataPublisher):
                attr.subscribe(
                    _get_attr_callback_for_publisher_child(attr_name), do_callback=False)

    if do_callback:
        callback_fun(self)

subscribe_attr 

subscribe_attr(attr_name: str, callback_fun: Callable[..., None])
Source code in src/omnipy/util/publisher.py 
23
24
25
26
27
28
29
30
31
32
33
34
def subscribe_attr(self, attr_name: str, callback_fun: Callable[..., None]):
    if not hasattr(self, attr_name):
        raise AttributeError(f'No attribute named "{attr_name}"')
    elif attr_name.startswith('_'):
        raise AttributeError(f'Subscribing to private member "{attr_name}" not allowed')
    else:
        self._attr_subscriptions[attr_name].append(callback_fun)
        attr = getattr(self, attr_name)
        callback_fun(attr)

        if isinstance(attr, DataPublisher):
            attr.subscribe(callback_fun, do_callback=False)

unsubscribe_all 

unsubscribe_all() -> None
Source code in src/omnipy/util/publisher.py 
56
57
58
59
60
61
62
63
64
def unsubscribe_all(self) -> None:
    self._self_subscriptions.clear()
    self._attr_subscriptions.clear()

    for attr_name in self.__class__.__fields__.keys():
        if not attr_name.startswith('_'):
            attr = getattr(self, attr_name)
            if isinstance(attr, DataPublisher):
                attr.unsubscribe_all()

RuntimeEntryPublisher

Bases: DataPublisher

Source code in src/omnipy/util/publisher.py 
 95
 96
 97
 98
 99
100
101
102
103
104
class RuntimeEntryPublisher(DataPublisher):
    _back: IsRuntime | None = pyd.PrivateAttr(default=None)

    def __setattr__(self, attr_name: str, value: object) -> None:
        new_value = hasattr(self, attr_name) and getattr(self, attr_name) is not value

        super().__setattr__(attr_name, value)

        if new_value and not attr_name.startswith('_') and self._back is not None:
            self._back.reset_subscriptions()