signal.py

# Original file from https://github.com/dgovil/PySignal


# The MIT License (MIT)

# Copyright (c) 2016 Dhruv Govil

# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:

# The above copyright notice and this permission notice shall be included in all
# copies or substantial portions of the Software.

# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.


from __future__ import annotations

from collections.abc import Callable
import inspect
import sys
from typing import Any, Never, Protocol, Self, overload
from typing_extensions import TypeIs
import weakref
from functools import partial


type Slot[**P, T] = Callable[P, T] | weakref.ref[Callable[P, T]] | weakref.WeakMethod[Callable[P, T]]


class RdvSignalInstance[**P, T]:
    """
    The RdvSignalInstance is the core object that handles connection and emission.
    """

    def __init__(self) -> None:
        self._block: bool = False
        self._sender: weakref.WeakMethod[Callable] | None = None
        self._slots: list[Slot[P, T]] = []
        self._results: weakref.WeakKeyDictionary[Slot[P, T], T] = weakref.WeakKeyDictionary()

    def __call__(self, *args: P.args, **kwargs: P.kwargs) -> None:
        self.emit(*args, **kwargs)

    def emit(self, *args: P.args, **kwargs: P.kwargs) -> None:
        """
        Calls all the connected slots with the provided args and kwargs unless block is activated
        """
        if self._block:
            return

        self._results = weakref.WeakKeyDictionary()

        self._determine_sender()

        def is_weak_ref(slot: Slot[P, T]) -> TypeIs[weakref.ref[Callable[P, T]]]:
            return isinstance(slot, weakref.ref)

        for slot in self._slots:
            if is_weak_ref(slot):
                # If it's a weakref, call the ref to get the instance and then call the func
                # Don't wrap in try/except so we don't risk masking exceptions from the actual func call
                tested_slot = slot()
                if tested_slot is not None:
                    self._results[slot] = tested_slot(*args, **kwargs)
            else:
                # Else call it in a standard way. Should be just lambdas at this point
                self._results[slot] = slot(*args, **kwargs)

        return

    def _get_slot(self, slot: Callable[P, T]) -> Slot:
        if not callable(slot):
            raise TypeError(f"{slot} is not callable")

        # If it's a partial, a Signal or a lambda
        if isinstance(slot, (partial, RdvSignalInstance)) or slot.__name__ == "<lambda>":
            return slot

        # If it's an instance method
        if inspect.ismethod(slot):
            return weakref.WeakMethod[Callable[P, T]](slot)

        # If it's just a function
        return weakref.ref(slot)

    def connect(self, slot: Callable[P, T]) -> None:
        """
        Connects the signal to any callable object
        """
        slot_ref = self._get_slot(slot)
        if slot_ref not in self._slots:
            self._slots.append(slot_ref)

    def disconnect(self, slot: Callable[P, T]) -> None:
        """
        Disconnects the slot from the signal
        """
        slot_ref = self._get_slot(slot)
        self._slots.remove(slot_ref)

    def result(self, slot: Callable[P, T]) -> T:
        """
        Returns this slot's return value from the last time the signal was emitted
        """
        slot_ref = self._get_slot(slot)
        if slot_ref not in self._results:
            raise ValueError(f"{slot} was not called last time {self} was emitted!")
        return self._results[slot_ref]

    def clear(self) -> None:
        """Clears the signal of all connected slots"""
        self._slots = []

    def block(self, isBlocked: bool) -> None:
        """Sets blocking of the signal"""
        self._block = bool(isBlocked)

    def _determine_sender(self) -> None:
        def _get_sender() -> Callable:
            """Try to get the bound, class or module method calling the emit."""
            prev_frame = sys._getframe(3)
            func_name = prev_frame.f_code.co_name

            # Faster to try/catch than checking for 'self'
            try:
                return getattr(prev_frame.f_locals['self'], func_name)

            except KeyError:
                return getattr(inspect.getmodule(prev_frame), func_name)

        # Get the sender
        try:
            self._sender = weakref.WeakMethod(_get_sender())

        # Account for when func_name is at '<module>'
        except AttributeError:
            self._sender = None

        # Handle unsupported module level methods for WeakMethod.
        # TODO: Support module level methods.
        except TypeError:
            self._sender = None

    def sender(self) -> Callable | None:
        """Return the callable responsible for emitting the signal, if found."""
        sender = self._sender

        if isinstance(sender, weakref.WeakMethod):
            return sender()

        return None


class RdvSignal[**P, T]:
    """
    The RdvSignal allows a signal to be set on a class rather than an instance.
    This emulates the behavior of a PyQt signal
    """
    _map: dict[RdvSignal, weakref.WeakKeyDictionary[Any, RdvSignalInstance]] = {}

    @overload
    def __get__(self, instance: None, owner: Any) -> Self: ...
    @overload
    def __get__(self, instance: Any, owner: Any) -> RdvSignalInstance[P, T]: ...

    def __get__(self, instance: Any, owner: Any) -> Self | RdvSignalInstance[P, T]:
        if instance is None:
            # When we access RdvSignal element on the class object without any instance,
            # we return the RdvSignal itself
            return self
        tmp = self._map.setdefault(self, weakref.WeakKeyDictionary())
        return tmp.setdefault(instance, RdvSignalInstance[P, T]())

    def __set__(self, instance: Any, value: Any) -> Never:
        raise RuntimeError("Cannot reassign an RdvSignal")

class Test:
    sig = RdvSignal[[], None]()
    sig2 = RdvSignal[[str, int], bool]()
    sig3 = RdvSignal[[str], None]()
    sig4 = RdvSignal[[], str]()

    def __init__(self) -> None:
        self.sig.connect(self.a)
        self.sig2.connect(self.b)
        self.sig3.connect(self.c)
        self.sig4.connect(self.d)

    def test(self) -> None:
        self.sig.emit()
        assert (a := self.sig.result(self.a)) is None

        self.sig2.emit("foo", 1)
        assert (b := self.sig2.result(self.b)) is True

        self.sig3.emit("foo")
        assert (c := self.sig3.result(self.c)) is None

        self.sig4.emit()
        assert (d := self.sig4.result(self.d)) == "foo"

        print(a)
        print(b)
        print(c)
        print(d)

    def a(self) -> None:
        pass

    def b(self, s: str, i: int) -> bool:
        return True

    def c(self, s: str) -> None:
        pass

    def d(self) -> str:
        return "foo"

if __name__ == "__main__":
    test = Test()
    test.test()