Using SignalInstance to bypass descriptor protocol

[jacopoabramo]

• psygnal version: 0.11.1
• Python version: 3.10
• Operating System: Windows

Description

A project I'm working on uses an internal framework that inherits the behavior of some Qt base classes, specifically the most critical one is the Qt signals and slot mechanism. I wanted to move away from having to natively depend from Qt from some time now so I reimplemented most of these classes - including the signal - in native Python. In case of Qt signals I - of course - switched to psygnal. After understanding a bit better the descriptor protocol, I came to the conclusion that for my specific implementation I don't really want to use it natively. The reason is the following: my application will work in the future with plugins that should be able to dynamically register new signals to a common exchange point, and we cannot know in advance how many signals an exchange point will have; with the descriptor protocol this did not seem a viable approach because all signals need to be defined as attributes, just like we did in the original Qt implementation.

What I Did

I re-implemented the SignalInstance class as follows:

class Signal(SignalInstance, abstract.Signal):
    """ `psygnal` implementation of the `base.Signal` abstract class.\\
    Qt signals can not be instantiated at runtime, while psygnal signals can. \\
    We use psygnal in synergy with the Qt framework to provide a way
    to create signals at runtime from different sources 
    (i.e. external plugins or communication channels).
    """

    def __init__(self, *argtypes: 'Any', info: str = "ImSwitch signal") -> None:
        SignalInstance.__init__(self, signature=argtypes)
        self._info = info

    def connect(self, func: 'Union[Callable, abstract.Signal]') -> None:
        if isinstance(func, abstract.Signal):
            if any([t1 != t2 for t1, t2 in zip(self.types, func.types)]):
                raise TypeError(f"Source and destination must have the same signature. Source signature: {self.types}, destination signature: {func.types}")
            func = func.emit
        super().connect(func)

    def disconnect(self, func: 'Union[Callable, abstract.Signal, None]' = None) -> None:
        if func is None:
            super().disconnect()
        super().disconnect(func)

    def emit(self, *args) -> None:
        super().emit(*args)

    @property
    @lru_cache
    def types(self) -> 'Tuple[type, ...]':
        return tuple([param.annotation for param in self._signature.parameters.values()])

    @property
    def info(self) -> str:
        return self._info

abstract.Signal is just an abstract base class that provides the blueprint of how the Signal class should look like, so it's just a bunch of @abstractmethods. The re-implementation of SignalInstance is to make sure that I would implement correctly all abstract methods and properties provided by the blueprint.

Together with this, I implemented a SignalInterface class which will be used to inherit from classes that want to emit signals (in origin, SignalInterface would inherit from QObject; as I got rid of it, now the class is just an empty class that doesn't do much but needs to be kept for consistency with legacy code).

class SignalInterface(abstract.SignalInterface):
    """ Base implementation of `abstract.SignalInterface`.
    """
    def __init__(self) -> None:
        ...

Now bringing everything together, I have a class that inherits from Signalinterface and provides methods to dynamically register new signals as attributes:

class CommunicationChannel(SignalInterface, ABC):
    """ Communication channel abstract base class.

    The communication channel is a mechanism for communication between objects. It is used to send signals from one object to another. \\
    This channel will allow to interconnect different controllers of the Presenter layer and provide a mechanism to exchange data between them. \\
    A channel can automatically register a new signal from a controller, exposing it to other controllers.

    Attributes:
        signals (`Mapping[str, Signal]`): A dictionary of signal names and their corresponding Signal objects.
    """

    def __init__(self) -> None:
        super().__init__()
        self.__logger = initLogger(self.__class__.__name__)

    def __setattr__(self, name: str, value: 'Any') -> None:
        """ Overloads `__setattr__` to allow registering signals as class attributes.
        If the attribute name starts with 'sig' and the value is a `Signal` object, it will be registered as a signal.
        Otherwise, it will be registered as a regular attribute.

        Args:
            name (`str`): attribute name.
            value (`Any`): attribute value.
        """
        if name.startswith('sig') and isinstance(value, Signal):
            if not hasattr(self, name):
                super().__setattr__(name, value)
            else:
                self.__logger.warning(f"Signal {name} already exists in {self.__class__.__name__}.")
        else:
            super().__setattr__(name, value)

    def registerSignal(self, name: str, *args, **kwargs) -> None:
        """ Creates a new `Signal` object with the given name and arguments,
        and stores it as class attribute.

        >>> channel.registerSignal('sigAcquisitionStarted', str)
        >>> # this will allow to access the signal as an attribute
        >>> channel.sigAcquisitionStarted.connect(mySlot)

        Signal names must start with 'sig' prefix.

        Args:
            name (`str`): signal name; this will be used as the attribute name.
            *args: data types carried by the signal.
            **kwargs: additional arguments to pass to the `Signal` constructor. Only 'info' is supported.

        Raises:
            `ValueError`: If `name` does not start with 'sig' prefix.
        """
        if not name.startswith('sig'):
            raise ValueError("Signal name must start with 'sig' prefix.")
        if 'info' in kwargs:
            signal = Signal(*args, info=kwargs['info'])
        else:
            signal = Signal(*args)
        setattr(self, name, signal)

I made some not-so-deep testing of all this and it seems to be working according to the behavior I expected.

The questions I have are the following:

• is it safe to use SignalInstance instead of Signal, when going with this approach? Do I risk making possible errors in connections or whatsoever?
• from the documentation you provide a SignalGroup class which allows to bundle up together multiple signals and provide some nifty methods to extract them/connecting them in one go. Is there any chance that SignalInterface could inherit from SignalGroup to maintain these traits and still be able to register SignalInstances dynamically?

Let me know if you need some further clarification.

[tlambert03]

is it safe to use SignalInstance instead of Signal, when going with this approach? Do I risk making possible errors in connections or whatsoever?

yes, absolutely, if you don't want to use the descriptor, then you can create SignalInstances dynamically. The primary benefit of the descriptor is type checking, autocompletion, etc... (which I value as a developer). If you want to be fully dynamic, there's nothing stopping you from manually managing SignalInstance however you want.

Is there any chance that SignalInterface could inherit from SignalGroup to maintain these traits and still be able to register SignalInstances dynamically?

yep, those objects are very similar.

on a quick glance, I'm not entirely following the motivation for the additional level of abstraction you're adding here; my quick (unstudied) feeling is that you're overcomplicating it a bit. It mostly sounds like you just need to manage a Mapping[str, SignalInstance]. Where the key is the name of the signal and the value is a SignalInstance. This is what SignalGroups do, albeit, without dynamically adding additional signals after instantiation.

Is the point of all your dynamism that you just don't know what signals your plugins are going to offer? Or do you actually want them to be able to be mutable, and register additional signals even after they have been initially registered? Here is an example of a dynamically generated SignalGroup:

from psygnal import Signal, SignalGroup

DynamicGroup = type(
    "DynamicGroup",
    (SignalGroup,),
    {"one_int": Signal(int), "int_str": Signal(int, str)},
)

my_signals = DynamicGroup()
my_signals.all.connect(lambda x: print("Emitting from all:", x))
my_signals.one_int.connect(lambda x: print("Emitting from one_int:", x))

my_signals.one_int.emit(1)
my_signals.int_str.emit(2, "two")

prints

Emitting from all: EmissionInfo(signal=<SignalInstance 'one_int' on <SignalGroup 'DynamicGroup' with 2 signals>>, args=(1,))
Emitting from one_int: 1
Emitting from all: EmissionInfo(signal=<SignalInstance 'int_str' on <SignalGroup 'DynamicGroup' with 2 signals>>, args=(2, 'two'))

[tlambert03]

or, in function form, similar to your registerSignals:

def create_signals(sigs: Mapping[str, tuple[Any, ...]]) -> SignalGroup:
    signals = {}
    for name, types in sigs.items():
        if not name.startswith("sig"):
            raise ValueError("Signal name must start with 'sig' prefix.")
        signals[name] = Signal(*types)
    return type("DynamicGroup", (SignalGroup,), signals)()

[jacopoabramo]

Is the point of all your dynamism that you just don't know what signals your plugins are going to offer? Or do you actually want them to be able to be mutable, and register additional signals even after they have been initially registered? Here is an example of a dynamically generated SignalGroup:

The first, yes. The idea is that CommChannel should be a singleton object that can register different signals depending on the plugin. This happens only at startup at any rate. Once the signals are created, they're immutable.

Come to think of it I guess that using SignalGroup is not necessary in my application, I can just leave the situation as it is. Doesn't seem advantageous and the thing that matters the most to me is that using SignalInstance doesn't give problems.

I'll close this issue as well, thanks for the quick feedback.