[Feature Request] Signal System

[gentlegiantJGC]

There are aspects of the library where it would be useful to get a push notification when something happens.
For example when using the library with a 3D renderer it would be useful to know when a chunk changed.
Currently we a system storing modification time and a thread that regularly queries the chunks but that is inefficient.

I would like to be able to directly use the PySide6 signal system and have a fallback when it is not installed.

Here is my current prototype which allows registering a signal back-end to enable hooking into other compatible signal systems.

from __future__ import annotations

import logging
from abc import ABC, abstractmethod
from typing import Optional, Union, Callable, Any, overload, Protocol, TYPE_CHECKING
from weakref import WeakMethod
from inspect import ismethod


if TYPE_CHECKING:
    import PySide6.QtCore  # noqa

    class SignalInstance(Protocol):
        def connect(self, slot: Union[Callable, SignalInstance], type: Union[None, PySide6.QtCore.Qt.ConnectionType] = ...): ...

        def disconnect(self, slot: Optional[Union[Callable, SignalInstance]] = None): ...

        def emit(self, *args: Any): ...


_signal_instance_constructor: Optional[SignalInstanceConstructor] = None


SignalInstanceCacheName = "_SignalCache"


class Signal:
    def __init__(self, *types: type, name: Optional[str] = "", arguments: Optional[str] = ()):
        self._types = types
        self._name = name
        self._arguments = arguments

    @overload
    def __get__(self, instance: None, owner: Optional[Any]) -> Signal:
        ...

    @overload
    def __get__(self, instance: Any, owner: Optional[Any]) -> SignalInstance:
        ...

    def __get__(self, instance, owner):
        if instance is None:
            return self
        try:
            signal_instances = getattr(instance, SignalInstanceCacheName)
        except:
            signal_instances = {}
            setattr(instance, SignalInstanceCacheName, signal_instances)
        if self not in signal_instances:
            if _signal_instance_constructor is None:
                set_signal_instance_constructor(get_fallback_signal_instance_constructor())
            signal_instances[self] = _signal_instance_constructor(
                types=self._types,
                name=self._name,
                arguments=self._arguments,
                signal=self,
                instance=instance,
                owner=owner
            )
        return signal_instances[self]


class SignalInstanceConstructor(Protocol):
    def __call__(
        self,
        *,
        types: tuple[type, ...],
        name: Optional[str],
        arguments: Optional[str],
        signal: Signal,
        instance: Any,
        owner: Any
    ) -> SignalInstance:
        ...


def set_signal_instance_constructor(constructor: SignalInstanceConstructor):
    global _signal_instance_constructor
    if _signal_instance_constructor is not None:
        raise RuntimeError("Signal instance constructor has already been set.")
    _signal_instance_constructor = constructor


def get_fallback_signal_instance_constructor() -> SignalInstanceConstructor:
    class FallbackSignalInstance:
        def __init__(
            self,
            *types: type
        ):
            self._types = types
            self._callbacks: set[Union[
                Callable,
                WeakMethod,
                FallbackSignalInstance
            ]] = set()

        @staticmethod
        def _wrap_slot(slot: Union[Callable, FallbackSignalInstance]):
            if ismethod(slot):
                return WeakMethod(slot)
            elif isinstance(slot, FallbackSignalInstance) or callable(slot):
                return slot
            else:
                raise RuntimeError(f"{slot} is not a supported slot type.")

        def connect(self, slot: Union[Callable, FallbackSignalInstance], type=None):
            if type is not None:
                logging.warning(
                    "FallbackSignalInstance does not support custom connection types. Using DirectConnection"
                )
            self._callbacks.add(self._wrap_slot(slot))

        def disconnect(self, slot: Union[Callable, FallbackSignalInstance] = None):
            self._callbacks.remove(self._wrap_slot(slot))

        def emit(self, *args: Any):
            if len(args) != len(self._types):
                raise TypeError(f"SignalInstance{self._types}.emit expected {len(self._types)} argument(s), {len(args)} given.")
            for slot in self._callbacks:
                try:
                    if isinstance(slot, FallbackSignalInstance):
                        slot.emit(*args)
                    elif isinstance(slot, WeakMethod):
                        slot = slot()
                        if slot is not None:
                            slot(*args)
                    else:
                        slot(*args)
                except Exception as e:
                    logging.error(e)

    def fallback_signal_instance_constructor(
        *,
        types: tuple[type, ...],
        name: Optional[str],
        arguments: Optional[str],
        signal: Signal,
        instance: Any,
        owner: Any
    ) -> FallbackSignalInstance:
        return FallbackSignalInstance(*types)

    return fallback_signal_instance_constructor


def get_pyside6_signal_instance_constructor() -> SignalInstanceConstructor:
    try:
        from PySide6.QtCore import QObject, Signal as PySide6_Signal, SignalInstance as PySide6_SignalInstance
    except ImportError as e:
        raise ImportError("Could not import PySide6") from e

    QObjectCacheName = "_QObjectCache"

    def pyside6_signal_instance_constructor(
        *,
        types: tuple[type, ...],
        name: Optional[str],
        arguments: Optional[str],
        signal: Signal,
        instance: Any,
        owner: Any
    ) -> PySide6_SignalInstance:
        if isinstance(instance, QObject):
            return PySide6_Signal(*types, name=name, arguments=arguments).__get__(instance, QObject)
        else:
            try:
                obj = getattr(instance, QObjectCacheName)
            except AttributeError:
                obj = QObject()
                setattr(instance, QObjectCacheName, obj)
            if not isinstance(obj, QObject):
                raise RuntimeError
            return PySide6_Signal(*types, name=name, arguments=arguments).__get__(obj, QObject)

    return pyside6_signal_instance_constructor


class MyAbstractObject(ABC):
    def __init__(self):
        super().__init__()

    test_signal = Signal(str)

    @abstractmethod
    def test(self):
        raise NotImplementedError


class MyObject(MyAbstractObject):
    def test(self):
        print("test")


def main():
    # set_signal_instance_constructor(get_fallback_signal_instance_constructor())
    # set_signal_instance_constructor(get_pyside6_signal_instance_constructor())

    assert MyObject.test_signal is MyObject.test_signal
    assert MyObject().test_signal is not MyObject().test_signal is not MyObject.test_signal

    obj = MyObject()

    def func(a):
        print(a)

    obj.test_signal.connect(func)
    # obj.test_signal.disconnect()
    obj.test_signal.emit("hi")


if __name__ == '__main__':
    main()

Edit: cache the SignalInstance on the instance using the Signal as a lookup. This was previously cached on the Signal but there is one Signal instance for all instances of the class which meant they all shared the same SignalInstance.
Switched storage variables to more obscure names.

[gentlegiantJGC]

@jevexendo I would like your thoughts on this when you get a chance.

[jevexendo]

So I'm not sure why we would need a fallback system. If PySide6 isn't installed, wouldn't there be bigger problems than a broken event notification system? As in, Amulet shouldn't even be able to load right?

As soon as I wrote that I realized this was for Amulet Core, not Amulet Editor.

I'll take a look at this when I get home from work later today.

[gentlegiantJGC]

The core library can be used independently of the GUI.

[Podshot]

I think an event system is a good approach, but I'm not sure if I'm on board using PySide as an event backend since it seems a little bloated to install a whole UI package in what's supposed to be a non-UI library just for an event system. Instead it looks like this package accomplishes the same goal and we can still keep our fallback implementation just in case: https://pypi.org/project/Events/

[gentlegiantJGC]

I agree that if you aren't already using PySide6 it is unnecessary bloat but the system I wrote isn't dependent on PySide6.
It is compatible with the PySide6 signal system and if enabled will directly use it.
If you don't enable PySide6 support it just uses a simple callback system which looks similar to what you linked.
It theoretically also allows hooking into other GUI signal systems.

[Podshot]

Oh yep I stand corrected, I see now that the fallback is just mimicking the PySide event functions, I initially didn't catch that in my first look through and was seeing the imports in the try/catch and thought that was the default behavior. I think this is a really good way to approach it

[jevexendo]

One concern I have is that PySide signals and slots only work in QObjects when the Qt main application thread is running. With that being the case, I think it would be better if we required the user to explicitly define which version they would like to use. It would probably help avoid confusion since a custom signal/slot implementation isn't going to be a drop-in replacement for Qt's system.

[gentlegiantJGC]

It would be the job of the application using the library to request the Signal provider that is used or provide their own.
If this isn't set then it defaults to the simple callback system.
The first line in the main function is where that is set.
If undefined in Signal.__get__ it initialises the default

[jevexendo]

Another concern I have relates to object inheritance. If the user says they want to use PySide6 Signals and Slots, all Amulet classes that emit Signals would need to inherit from QObject. I'm not particularly sure how one dynamically modifies class inheritance, but it doesn't sound like a particularly good idea.

Also, since QObject is a metaclass, any classes in Amulet that inherit from an abstract base class cannot also directly inherit from QObject or you'll get an error like:

TypeError: metaclass conflict: the metaclass of a derived class must be a (non-strict) subclass of the metaclasses of all its bases

I did manage to find a workaround for this issue on Stack Overflow, but I think that allowing the dynamically swapping between simple callbacks and signals/slots may be more annoying to implement than it's worth.

from abc import ABC, ABCMeta

from PySide6.QtCore import QObject
from PySide6.QtWidgets import QWidget

QObjectMeta = type(QObject)
QWidgetMeta = type(QWidget)

class _ABCQObjectMeta(QObjectMeta, ABCMeta):...
class _ABCQWidgetMeta(QObjectMeta, ABCMeta):...

class ABCQObject(QObject, ABC, metaclass=_ABCQObjectMeta):...
class ABCQWidget(QWidget, ABC, metaclass=_ABCQWidgetMeta):...

Source: https://stackoverflow.com/questions/66591752/

[jevexendo]

Now admittedly, I must just be misunderstanding something about your sample code since the PySide6 signal instance appears to be working fine despite being used within an abstract base class that doesn't inherit from QObject. Especially since it clearly thinks there's a QObject since this code exists:

if not isinstance(obj, QObject):
    raise RuntimeError

What am I missing here?

[gentlegiantJGC]

MyAbstractObject isn't an instance of QObject and it doesn't have to be.
In PySide6 Signal isn't actually a signal it is a getter for a signal. SignalInstance is the actual signal.
In the code above my Signal is a custom class that doesn't touch PySide6 it just calls out to the implementation.

The PySide6 SignalInstance constructor needs a QObject to exist but the class we end up setting it on doesn't need to be an instance of QObject.

Here is the important part

def get_pyside6_signal_instance_constructor() -> SignalInstanceConstructor:
    try:
        from PySide6.QtCore import QObject, Signal as PySide6_Signal, SignalInstance as PySide6_SignalInstance
    except ImportError as e:
        raise ImportError("Could not import PySide6") from e

    def pyside6_signal_instance_constructor(
        *,
        types: tuple[type, ...],
        name: Optional[str],
        arguments: Optional[str],
        signal: Signal,
        instance: Any,
        owner: Any
    ) -> PySide6_SignalInstance:
        if isinstance(instance, QObject):
            return PySide6_Signal(*types, name=name, arguments=arguments).__get__(instance, QObject)
        else:
            try:
                obj = instance._qobject
            except AttributeError:
                obj = instance._qobject = QObject()
            if not isinstance(obj, QObject):
                raise RuntimeError
            return PySide6_Signal(*types, name=name, arguments=arguments).__get__(obj, QObject)

    return pyside6_signal_instance_constructor

I would suggest looking into the descriptor protocol if you don't know what it is. It is the mechanic how properties work.
instance is the instance of MyAbstractObject
If instance is an instance of QObject (you inherited directly from QObject) we just create a PySide6 SignalInstance on the instance.
If it isn't an instance we need a QObject so we create a new one and set it as an attribute on instance and create a SignalInstance on that instead.
The RuntimeError you commented on is just in case you already defined _qobject on the instance as something other than a QObject.

I have just spotted a bug in the code that caches the SignalInstance that means that all instance of MyAbstractObject share the same SignalInstance. I will need to fix that.
Edit: Just fixed the above in the original post.