Source code for magicgui.events

"""deprecation strategy"""

import warnings
import weakref
from collections import namedtuple
from typing import Callable, Dict

import psygnal


def _new_style_slot(slot: Callable) -> bool:
    sig = psygnal._signal.signature(slot)
    if len(sig.parameters) != 1:
        return True
    p0 = list(sig.parameters.values())[0]
    return p0.annotation is not p0.empty and (
        (getattr(p0.annotation, "__name__", "") != "Event")
        or (str(p0.annotation) == "Event")
    )


Event = namedtuple("Event", ("value", "type", "source"))


class SignalInstance(psygnal.SignalInstance):
    _new_callback: Dict[psygnal._signal.NormedCallback, bool] = {}

    def connect(
        self,
        slot=None,
        *,
        check_nargs=None,
        check_types=None,
        unique=False,
    ):
        is_new_style = _new_style_slot(slot)
        if not is_new_style:
            name = getattr(self._instance, "name", "") or "widget"
            signame = self.name
            warnings.warn(
                "\n\nmagicgui 0.4.0 will change the way that callbacks are called.\n"
                "Instead of a single `Event` instance, with an `event.value` attribute,"
                "\ncallbacks will receive the value(s) directly:\n\n"
                f"@{name}.{signame}.connect\n"
                "def my_callback(*args):\n"
                "    # *args are the value(s) themselves!"
                "\n\nTo silence this warning you may either provide a callback that "
                "has more\nor less than 1 parameter.  Or annotate the single parameter "
                "as anything\n*other* than `Event`, e.g. `def callback(x: int): ...`"
                "\nFor details, see: https://github.com/napari/magicgui/issues/255",
                FutureWarning,
                stacklevel=2,
            )
        result = super().connect(
            slot, check_nargs=check_nargs, check_types=check_types, unique=unique
        )
        self._new_callback[self._normalize_slot(slot)] = is_new_style
        return result

    def _run_emit_loop(self, args) -> None:

        rem = []
        # allow receiver to query sender with Signal.current_emitter()
        with self._lock:
            with Signal._emitting(self), psygnal.Signal._emitting(self):
                for _slt in self._slots:
                    (slot, max_args) = _slt
                    if isinstance(slot, tuple):
                        _ref, method_name = slot
                        obj = _ref()
                        if obj is None:
                            rem.append(slot)  # add dead weakref
                            continue
                        cb = getattr(obj, method_name, None)
                        if cb is None:  # pragma: no cover
                            rem.append(slot)  # object has changed?
                            continue
                    else:
                        cb = slot

                    # TODO: add better exception handling
                    if self._new_callback.get(_slt[0]):
                        cb(*args[:max_args])
                    else:
                        cb(Event(args[0], self.name, self.instance))

            for slot in rem:
                self.disconnect(slot)

        return None

    def __call__(self, *args, **kwds):
        if kwds:
            name = getattr(self._instance, "name", "") or "widget"
            signame = self.name
            args = args + tuple(kwds.values())
            argrepr = ", ".join(repr(s) for s in args)
            kwargrepr = ",".join(f"{k}={v!r}" for k, v in kwds.items())
            warnings.warn(
                "\n\nmagicgui 0.4.0 is using psygnal for event emitters.\n"
                f"Keyword arguments ({set(kwds)!r}) are no longer accepted by the event"
                f" emitter.\nUse '{name}.{signame}({argrepr})' instead of "
                f"{name}.{signame}({kwargrepr}).\nIn the future this will be an error."
                "\nFor details, see: https://github.com/napari/magicgui/issues/255",
                FutureWarning,
                stacklevel=2,
            )
        return self._run_emit_loop(args)


class Signal(psygnal.Signal):
    def __get__(self, instance, owner=None):
        if instance is None:
            return self
        d = self._signal_instances.setdefault(self, weakref.WeakKeyDictionary())
        return d.setdefault(
            instance,
            SignalInstance(
                self.signature,
                instance=instance,
                name=self._name,
                check_nargs_on_connect=self._check_nargs_on_connect,
                check_types_on_connect=self._check_types_on_connect,
            ),
        )