[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.