Skip to content

sunflare.virtual

VirtualBus

Bases: Loggable

Data exchange layer.

The VirtualBus is a mechanism to exchange data between different parts of the system. Communication can happen between plugins on the same layer as well as between different layers of the system.

It can be used to emit notifications and carry information to other plugins, or to register document callbacks that process documents generated during data acquisition.

Source code in src/sunflare/virtual/_bus.py 
 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
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
class VirtualBus(Loggable):
    """Data exchange layer.

    The ``VirtualBus`` is a mechanism to exchange
    data between different parts of the system. Communication
    can happen between plugins on the same layer as
    well as between different layers of the system.

    It can be used to emit notifications and carry information
    to other plugins, or to register document callbacks
    that process documents generated during data acquisition.
    """

    def __init__(self) -> None:
        self._signals: dict[str, SignalCache] = {}
        self._callbacks: dict[str, CallbackType] = {}

    def register_signals(
        self, owner: object, only: Iterable[str] | None = None
    ) -> None:
        """
        Register the signals of an object in the virtual bus.

        Parameters
        ----------
        owner : object
            The instance whose class's signals are to be cached.
        only : Iterable[str], optional
            A list of signal names to cache. If not provided, all
            signals in the class will be cached automatically by inspecting
            the class attributes.

        Notes
        -----
        This method inspects the attributes of the owner's class to find
        [`psygnal.Signal`][] descriptors. For each such descriptor, it retrieves
        the [`psygnal.SignalInstance`][] from the owner using the descriptor protocol and
        stores it in the registry.
        """
        owner_class = type(owner)  # Get the class of the object
        class_name = owner_class.__name__  # Name of the class

        if only is None:
            # Automatically detect all attributes of the class that are psygnal.Signal descriptors
            only = [
                name
                for name in dir(owner_class)
                if isinstance(getattr(owner_class, name, None), Signal)
            ]

        # Initialize the registry for this class if not already present
        if class_name not in self._signals:
            self._signals[class_name] = SignalCache()

        # Iterate over the specified signal names and cache their instances
        for name in only:
            signal_descriptor = getattr(owner_class, name, None)
            if isinstance(signal_descriptor, Signal):
                # Retrieve the SignalInstance using the descriptor protocol
                signal_instance = getattr(owner, name)
                self._signals[class_name][name] = signal_instance

    def register_callbacks(self, callback: CallbackType) -> None:
        """Register a document callback in the virtual bus.

        Allows other components of the system access to
        specific document routers through the `callbacks` property.

        Parameters
        ----------
        callback : ``CallbackType``
            The document callback to register.

        Raises
        ------
        TypeError
            If the provided callback is not callable or does not
            accept the correct parameters.
        """
        if isinstance(callback, DocumentRouter):
            self._callbacks[callback.__class__.__name__] = callback
        else:
            if not callable(callback):
                raise TypeError(f"{callback} is not callable.")
            # validate that the callback accepts only two parameters
            try:
                inspect.signature(callback).bind(None, None)
            except TypeError as e:
                raise TypeError(
                    "The callback function must accept exactly two parameters: "
                    "'name' (str) and 'document' (Document)."
                ) from e

            # determine the key based on the type of callback
            if inspect.ismethod(callback):
                # bound method: if it's __call__, use the class name; otherwise use the method name
                if callback.__name__ == "__call__":
                    key = callback.__self__.__class__.__name__
                else:
                    key = callback.__name__
            elif inspect.isfunction(callback):
                # regular function: use the function name
                key = callback.__name__
            elif hasattr(callback, "__call__"):
                # callable object (instance with __call__ method): use the class name
                key = callback.__class__.__name__
            else:
                # fallback (should not reach here due to earlier callable check)
                key = callback.__name__

            self._callbacks[key] = callback

    @property
    def callbacks(self) -> dict[str, CallbackType]:
        """The currently registered document callbacks in the virtual bus."""
        return self._callbacks

    @property
    def signals(self) -> dict[str, SignalCache]:
        """The currently registered signals in the virtual bus."""
        return self._signals

callbacks property 

callbacks: dict[str, CallbackType]

The currently registered document callbacks in the virtual bus.

signals property 

signals: dict[str, SignalCache]

The currently registered signals in the virtual bus.

register_signals 

register_signals(
    owner: object, only: Iterable[str] | None = None
) -> None

Register the signals of an object in the virtual bus.

Parameters:

Name Type Description Default
owner object

The instance whose class's signals are to be cached.

 required
only Iterable[str]

A list of signal names to cache. If not provided, all signals in the class will be cached automatically by inspecting the class attributes.

 None
Notes

This method inspects the attributes of the owner's class to find psygnal.Signal descriptors. For each such descriptor, it retrieves the psygnal.SignalInstance from the owner using the descriptor protocol and stores it in the registry.

Source code in src/sunflare/virtual/_bus.py 
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
def register_signals(
    self, owner: object, only: Iterable[str] | None = None
) -> None:
    """
    Register the signals of an object in the virtual bus.

    Parameters
    ----------
    owner : object
        The instance whose class's signals are to be cached.
    only : Iterable[str], optional
        A list of signal names to cache. If not provided, all
        signals in the class will be cached automatically by inspecting
        the class attributes.

    Notes
    -----
    This method inspects the attributes of the owner's class to find
    [`psygnal.Signal`][] descriptors. For each such descriptor, it retrieves
    the [`psygnal.SignalInstance`][] from the owner using the descriptor protocol and
    stores it in the registry.
    """
    owner_class = type(owner)  # Get the class of the object
    class_name = owner_class.__name__  # Name of the class

    if only is None:
        # Automatically detect all attributes of the class that are psygnal.Signal descriptors
        only = [
            name
            for name in dir(owner_class)
            if isinstance(getattr(owner_class, name, None), Signal)
        ]

    # Initialize the registry for this class if not already present
    if class_name not in self._signals:
        self._signals[class_name] = SignalCache()

    # Iterate over the specified signal names and cache their instances
    for name in only:
        signal_descriptor = getattr(owner_class, name, None)
        if isinstance(signal_descriptor, Signal):
            # Retrieve the SignalInstance using the descriptor protocol
            signal_instance = getattr(owner, name)
            self._signals[class_name][name] = signal_instance

register_callbacks 

register_callbacks(callback: CallbackType) -> None

Register a document callback in the virtual bus.

Allows other components of the system access to specific document routers through the callbacks property.

Parameters:

Name Type Description Default
callback ``CallbackType``

The document callback to register.

 required

Raises:

Type Description
TypeError

If the provided callback is not callable or does not accept the correct parameters.
Source code in src/sunflare/virtual/_bus.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
def register_callbacks(self, callback: CallbackType) -> None:
    """Register a document callback in the virtual bus.

    Allows other components of the system access to
    specific document routers through the `callbacks` property.

    Parameters
    ----------
    callback : ``CallbackType``
        The document callback to register.

    Raises
    ------
    TypeError
        If the provided callback is not callable or does not
        accept the correct parameters.
    """
    if isinstance(callback, DocumentRouter):
        self._callbacks[callback.__class__.__name__] = callback
    else:
        if not callable(callback):
            raise TypeError(f"{callback} is not callable.")
        # validate that the callback accepts only two parameters
        try:
            inspect.signature(callback).bind(None, None)
        except TypeError as e:
            raise TypeError(
                "The callback function must accept exactly two parameters: "
                "'name' (str) and 'document' (Document)."
            ) from e

        # determine the key based on the type of callback
        if inspect.ismethod(callback):
            # bound method: if it's __call__, use the class name; otherwise use the method name
            if callback.__name__ == "__call__":
                key = callback.__self__.__class__.__name__
            else:
                key = callback.__name__
        elif inspect.isfunction(callback):
            # regular function: use the function name
            key = callback.__name__
        elif hasattr(callback, "__call__"):
            # callable object (instance with __call__ method): use the class name
            key = callback.__class__.__name__
        else:
            # fallback (should not reach here due to earlier callable check)
            key = callback.__name__

        self._callbacks[key] = callback

HasConnection

Bases: Protocol

Protocol marking your class as requesting connection to other signals.

Tip

This protocol is optional and only usable with Presenters and Views. Models will not be affected by this protocol.

Source code in src/sunflare/virtual/_protocols.py 
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
93
@runtime_checkable
class HasConnection(Protocol):  # pragma: no cover
    """Protocol marking your class as requesting connection to other signals.

    !!! tip
        This protocol is optional and only usable with
        ``Presenters`` and ``Views``. ``Models``
        will not be affected by this protocol.
    """

    @abstractmethod
    def connection_phase(self) -> None:
        """Connect to other objects via the virtual bus.

        At application start-up, objects within Redsun can't know what signals are available from other parts of the session.
        This method is invoked after the object's construction and after `registration_phase` as well, allowing to
        connect to all available registered signals in the virtual bus.
        Objects may be able to connect to other signals even after this phase (provided those signals
        were registered before).

        An implementation example:

        ```python
        def connection_phase(self) -> None:
            # you can connect signals from another controller to your local slots...
            self.virtual_bus["OtherController"]["signal"].connect(self._my_slot)

            # ... or to other signals ...
            self.virtual_bus["OtherController"]["signal"].connect(self.sigMySignal)

            # ... or connect to a view component
            self.virtual_bus["OtherWidget"]["sigWidget"].connect(self._my_slot)
        ```
        """
        ...

connection_phase abstractmethod 

connection_phase() -> None

Connect to other objects via the virtual bus.

At application start-up, objects within Redsun can't know what signals are available from other parts of the session. This method is invoked after the object's construction and after registration_phase as well, allowing to connect to all available registered signals in the virtual bus. Objects may be able to connect to other signals even after this phase (provided those signals were registered before).

An implementation example:

def connection_phase(self) -> None:
    # you can connect signals from another controller to your local slots...
    self.virtual_bus["OtherController"]["signal"].connect(self._my_slot)

    # ... or to other signals ...
    self.virtual_bus["OtherController"]["signal"].connect(self.sigMySignal)

    # ... or connect to a view component
    self.virtual_bus["OtherWidget"]["sigWidget"].connect(self._my_slot)
Source code in src/sunflare/virtual/_protocols.py 
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
@abstractmethod
def connection_phase(self) -> None:
    """Connect to other objects via the virtual bus.

    At application start-up, objects within Redsun can't know what signals are available from other parts of the session.
    This method is invoked after the object's construction and after `registration_phase` as well, allowing to
    connect to all available registered signals in the virtual bus.
    Objects may be able to connect to other signals even after this phase (provided those signals
    were registered before).

    An implementation example:

    ```python
    def connection_phase(self) -> None:
        # you can connect signals from another controller to your local slots...
        self.virtual_bus["OtherController"]["signal"].connect(self._my_slot)

        # ... or to other signals ...
        self.virtual_bus["OtherController"]["signal"].connect(self.sigMySignal)

        # ... or connect to a view component
        self.virtual_bus["OtherWidget"]["sigWidget"].connect(self._my_slot)
    ```
    """
    ...

HasRegistration

Bases: Protocol

Protocol marking your class as capable of emitting signals.

Tip

This protocol is optional and only available for Presenters and Widgets. Models will not be affected by this protocol.

Source code in src/sunflare/virtual/_protocols.py 
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
@runtime_checkable
class HasRegistration(Protocol):  # pragma: no cover
    """Protocol marking your class as capable of emitting signals.

    !!! tip
        This protocol is optional and only available for
        ``Presenters`` and ``Widgets``. ``Models``
        will not be affected by this protocol.
    """

    @abstractmethod
    def registration_phase(self) -> None:
        r"""Register the signals listed in this method to expose them to the virtual bus.

        At application start-up, controllers can't know what signals are available from other controllers. \
        This method is called after all controllers are initialized to allow them to register their signals. \
        Presenters may be able to register further signals even after this phase (but not before the `connection_phase` ended). \

        Only signals defined in your object can be registered.

        An implementation example:

        ```python
        def registration_phase(self) -> None:
            # you can register all signals...
            self.virtual_bus.register_signals(self)

            # ... or only a selection of them
            self.virtual_bus.register_signals(self, only=["signal"])
        ```
        """
        ...

registration_phase abstractmethod 

registration_phase() -> None

Register the signals listed in this method to expose them to the virtual bus.

At application start-up, controllers can't know what signals are available from other controllers. \ This method is called after all controllers are initialized to allow them to register their signals. \ Presenters may be able to register further signals even after this phase (but not before the connection_phase ended). \

Only signals defined in your object can be registered.

An implementation example:

def registration_phase(self) -> None:
    # you can register all signals...
    self.virtual_bus.register_signals(self)

    # ... or only a selection of them
    self.virtual_bus.register_signals(self, only=["signal"])
Source code in src/sunflare/virtual/_protocols.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
@abstractmethod
def registration_phase(self) -> None:
    r"""Register the signals listed in this method to expose them to the virtual bus.

    At application start-up, controllers can't know what signals are available from other controllers. \
    This method is called after all controllers are initialized to allow them to register their signals. \
    Presenters may be able to register further signals even after this phase (but not before the `connection_phase` ended). \

    Only signals defined in your object can be registered.

    An implementation example:

    ```python
    def registration_phase(self) -> None:
        # you can register all signals...
        self.virtual_bus.register_signals(self)

        # ... or only a selection of them
        self.virtual_bus.register_signals(self, only=["signal"])
    ```
    """
    ...

HasShutdown

Bases: Protocol

Protocol marking your class as capable of shutting down.

Tip

This protocol is optional and only available for Presenters. Widgets and Models will not be affected by this protocol.

Source code in src/sunflare/virtual/_protocols.py 
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
class HasShutdown(Protocol):  # pragma: no cover
    """Protocol marking your class as capable of shutting down.

    !!! tip
        This protocol is optional and only available for
        ``Presenters``. ``Widgets`` and ``Models`` will not
        be affected by this protocol.
    """

    @abstractmethod
    def shutdown(self) -> None:
        """Shutdown an object. Performs cleanup operations.

        If the object holds any kind of resources,
        this method should invoke any equivalent shutdown method for each resource.
        """
        ...

shutdown abstractmethod 

shutdown() -> None

Shutdown an object. Performs cleanup operations.

If the object holds any kind of resources, this method should invoke any equivalent shutdown method for each resource.

Source code in src/sunflare/virtual/_protocols.py 
15
16
17
18
19
20
21
22
@abstractmethod
def shutdown(self) -> None:
    """Shutdown an object. Performs cleanup operations.

    If the object holds any kind of resources,
    this method should invoke any equivalent shutdown method for each resource.
    """
    ...