psygnal #

def insert_path(self, *path: PathStep) -> EmissionInfo:
    """Return a new EmissionInfo with the given path steps inserted at the front."""
    return EmissionInfo(self.signal, self.args, (*path, *self.path))

def reset(self) -> None:
    """Reset the state of the model to default values."""
    model_config = self.model_config
    model_fields = _get_fields(type(self))
    for name, value in self._defaults.items():
        if isinstance(value, EventedModel):
            cast("EventedModel", getattr(self, name)).reset()
        elif not model_config.get("frozen") and not model_fields[name].frozen:
            setattr(self, name, value)

def update(self, values: Union["EventedModel", dict], recurse: bool = True) -> None:
    """Update a model in place.

    Parameters
    ----------
    values : Union[dict, EventedModel]
        Values to update the model with. If an EventedModel is passed it is
        first converted to a dictionary. The keys of this dictionary must
        be found as attributes on the current model.
    recurse : bool
        If True, recursively update fields that are EventedModels.
        Otherwise, just update the immediate fields of this EventedModel,
        which is useful when the declared field type (e.g. ``Union``) can have
        different realized types with different fields.
    """
    if isinstance(values, pydantic.BaseModel):
        values = values.model_dump()

    if not isinstance(values, dict):  # pragma: no cover
        raise TypeError(f"values must be a dict or BaseModel. got {type(values)}")

    with self.events._psygnal_relay.paused():  # TODO: reduce?
        for key, value in values.items():
            field = getattr(self, key)
            if isinstance(field, EventedModel) and recurse:
                field.update(value, recurse=recurse)
            else:
                setattr(self, key, value)

@classmethod
def current_emitter(cls) -> SignalInstance | None:
    """Return currently emitting `SignalInstance`, if any.

    This will typically be used in a callback.

    Examples
    --------
    ```python
    from psygnal import Signal


    def my_callback():
        source = Signal.current_emitter()
    ```
    """
    return cls._current_emitter

@classmethod
def sender(cls) -> Any:
    """Return currently emitting object, if any.

    This will typically be used in a callback.
    """
    return getattr(cls._current_emitter, "instance", None)

@classmethod
def is_uniform(cls) -> bool:
    """Return true if all signals in the group have the same signature."""
    warnings.warn(
        "The `is_uniform` method on SignalGroup is deprecated. Use "
        "`psygnals_uniform` instead. This will be an error in v0.11.",
        FutureWarning,
        stacklevel=2,
    )
    return cls._psygnal_uniform

@classmethod
def psygnals_uniform(cls) -> bool:
    """Return true if all signals in the group have the same signature."""
    return cls._psygnal_uniform

def block(self, exclude: Container[str | SignalInstance] = ()) -> None:
    """Block this signal from emitting.

    NOTE: the `exclude` argument is only for SignalGroup subclass, but we
    have to include it here to make mypyc happy.
    """
    self._is_blocked = True

def blocked(self) -> AbstractContextManager[None]:
    """Context manager to temporarily block this signal.

    Useful if you need to temporarily block all emission of a given signal,
    (for example, to avoid a recursive signal loop)

    Examples
    --------
    ```python
    class MyEmitter:
        changed = Signal()

        def make_a_change(self):
            self.changed.emit()

    obj = MyEmitter()

    with obj.changed.blocked()
        obj.make_a_change()  # will NOT emit a changed signal.
    ```
    """
    return _SignalBlocker(self)

def connect(
    self,
    slot: F | None = None,
    *,
    thread: threading.Thread | Literal["main", "current"] | None = None,
    check_nargs: bool | None = None,
    check_types: bool | None = None,
    unique: bool | Literal["raise"] = False,
    max_args: int | None = None,
    on_ref_error: RefErrorChoice = "warn",
    priority: int = 0,
    emit_on_evented_child_events: bool = False,
) -> Callable[[F], F] | F:
    """Connect a callback (`slot`) to this signal.

    `slot` is compatible if:

    * it requires no more than the number of positional arguments emitted by this
      `SignalInstance`.  (It *may* require less)
    * it has no *required* keyword arguments (keyword only arguments that have
      no default).
    * if `check_types` is `True`, the parameter types in the callback signature must
      match the signature of this `SignalInstance`.

    This method may be used as a decorator.

    ```python
    @signal.connect
    def my_function(): ...
    ```

    !!!important
        If a signal is connected with `thread != None`, then it is up to the user
        to ensure that `psygnal.emit_queued` is called, or that one of the backend
        convenience functions is used (e.g. `psygnal.qt.start_emitting_from_queue`).
        Otherwise, callbacks that are connected to signals that are emitted from
        another thread will never be called.

    Parameters
    ----------
    slot : Callable
        A callable to connect to this signal.  If the callable accepts less
        arguments than the signature of this slot, then they will be discarded when
        calling the slot.
    check_nargs : Optional[bool]
        If `True` and the provided `slot` requires more positional arguments than
        the signature of this Signal, raise `TypeError`. by default `True`.
    thread: Thread | Literal["main", "current"] | None
        If `None` (the default), this slot will be invoked immediately when a signal
        is emitted, from whatever thread emitted the signal. If a thread object is
        provided, then the callback will only be immediately invoked if the signal
        is emitted from that thread.  Otherwise, the callback will be added to a
        queue. **Note!**, when using the `thread` parameter, the user is responsible
        for calling `psygnal.emit_queued()` in the corresponding thread, otherwise
        the slot will never be invoked. (See note above). (The strings `"main"` and
        `"current"` are also accepted, and will be interpreted as the
        `threading.main_thread()` and `threading.current_thread()`, respectively).
    check_types : Optional[bool]
        If `True`, An additional check will be performed to make sure that types
        declared in the slot signature are compatible with the signature
        declared by this signal, by default `False`.
    unique : Union[bool, str, None]
        If `True`, returns without connecting if the slot has already been
        connected.  If the literal string "raise" is passed to `unique`, then a
        `ValueError` will be raised if the slot is already connected.
        By default `False`.
    max_args : Optional[int]
        If provided, `slot` will be called with no more more than `max_args` when
        this SignalInstance is emitted.  (regardless of how many arguments are
        emitted).
    on_ref_error : {'raise', 'warn', 'ignore'}, optional
        What to do if a weak reference cannot be created.  If 'raise', a
        ReferenceError will be raised.  If 'warn' (default), a warning will be
        issued and a strong-reference will be used. If 'ignore' a strong-reference
        will be used (silently).
    priority : int
        The priority of the callback. This is used to determine the order in which
        callbacks are called when multiple are connected to the same signal.
        Higher priority callbacks are called first. Negative values are allowed.
        The default is 0.
    emit_on_evented_child_events : bool
        If `True`, and if this is a SignalInstance associated with a specific field
        on an evented dataclass, and if that field itself is an evented dataclass,
        then the slot will be called both when the field is set directly, *and* when
        a child member of that field is set.
        For example, if `Team` is an evented-dataclass with a field `leader: Person`
        which is itself an evented-dataclass, then
        `team.events.leader.connect(callback, emit_on_evented_child_events=True)`
        will invoke callback even when `team.leader.age` is mutated (in addition to
        when `team.leader` is set directly).

    Raises
    ------
    TypeError
        If a non-callable object is provided.
    ValueError
        If the provided slot fails validation, either due to mismatched positional
        argument requirements, or failed type checking.
    ValueError
        If `unique` is `'raise'` and `slot` has already been connected.
    """
    if check_nargs is None:
        check_nargs = self._check_nargs_on_connect
    if check_types is None:
        check_types = self._check_types_on_connect

    def _wrapper(
        slot: F,
        max_args: int | None = max_args,
        _on_ref_err: RefErrorChoice = on_ref_error,
    ) -> F:
        if not callable(slot):
            raise TypeError(f"Cannot connect to non-callable object: {slot}")

        with self._lock:
            if unique and slot in self:
                if unique == "raise":
                    raise ValueError(
                        "Slot already connect. Use `connect(..., unique=False)` "
                        "to allow duplicate connections"
                    )
                return slot

            slot_sig: Signature | None = None
            if check_nargs and (max_args is None):
                slot_sig, max_args, isqt = self._check_nargs(slot, self.signature)
                if isqt:
                    _on_ref_err = "ignore"
            if check_types:
                slot_sig = slot_sig or signature(slot)
                if not _parameter_types_match(slot, self.signature, slot_sig):
                    extra = f"- Slot types {slot_sig} do not match types in signal."
                    self._raise_connection_error(slot, extra)

            cb = weak_callback(
                slot,
                max_args=max_args,
                finalize=self._try_discard,
                on_ref_error=_on_ref_err,
                priority=priority,
            )
            if thread is not None:
                cb = QueuedCallback(cb, thread=thread)
            self._append_slot(cb)

        if emit_on_evented_child_events:
            self._connect_child_event_listener(slot)
        return slot

    return _wrapper if slot is None else _wrapper(slot)

def connect_setattr(
    self,
    obj: object,
    attr: str,
    maxargs: int | None | object = _NULL,
    *,
    on_ref_error: RefErrorChoice = "warn",
    priority: int = 0,
) -> WeakCallback[None]:
    """Bind an object attribute to the emitted value of this signal.

    Equivalent to calling `self.connect(functools.partial(setattr, obj, attr))`,
    but with additional weakref safety (i.e. a strong reference to `obj` will not
    be retained). The return object can be used to
    [`disconnect()`][psygnal.SignalInstance.disconnect], (or you can use
    [`disconnect_setattr()`][psygnal.SignalInstance.disconnect_setattr]).

    Parameters
    ----------
    obj : object
        An object.
    attr : str
        The name of an attribute on `obj` that should be set to the value of this
        signal when emitted.
    maxargs : Optional[int]
        max number of positional args to accept
    on_ref_error: {'raise', 'warn', 'ignore'}, optional
        What to do if a weak reference cannot be created.  If 'raise', a
        ReferenceError will be raised.  If 'warn' (default), a warning will be
        issued and a strong-reference will be used. If 'ignore' a strong-reference
        will be used (silently).
    priority : int
        The priority of the callback. This is used to determine the order in which
        callbacks are called when multiple are connected to the same signal.
        Higher priority callbacks are called first. Negative values are allowed.
        The default is 0.

    Returns
    -------
    Tuple
        (weakref.ref, name, callable).  Reference to the object, name of the
        attribute, and setattr closure.  Can be used to disconnect the slot.

    Raises
    ------
    ValueError
        If this is not a single-value signal
    AttributeError
        If `obj` has no attribute `attr`.

    Examples
    --------
    >>> class T:
    ...     sig = Signal(int)
    >>> class SomeObj:
    ...     x = 1
    >>> t = T()
    >>> my_obj = SomeObj()
    >>> t.sig.connect_setattr(my_obj, "x")
    >>> t.sig.emit(5)
    >>> assert my_obj.x == 5
    """
    if maxargs is _NULL:
        warnings.warn(
            "The default value of maxargs will change from `None` to `1` in "
            "version 0.11. To silence this warning, provide an explicit value for "
            "maxargs (`None` for current behavior, `1` for future behavior).",
            FutureWarning,
            stacklevel=2,
        )
        maxargs = None

    if not hasattr(obj, attr):
        raise AttributeError(f"Object {obj} has no attribute {attr!r}")

    with self._lock:
        caller = WeakSetattr(
            obj,
            attr,
            max_args=cast("int | None", maxargs),
            finalize=self._try_discard,
            on_ref_error=on_ref_error,
            priority=priority,
        )
        self._append_slot(caller)
    return caller

def connect_setitem(
    self,
    obj: object,
    key: str,
    maxargs: int | None | object = _NULL,
    *,
    on_ref_error: RefErrorChoice = "warn",
    priority: int = 0,
) -> WeakCallback[None]:
    """Bind a container item (such as a dict key) to emitted value of this signal.

    Equivalent to calling `self.connect(functools.partial(obj.__setitem__, attr))`,
    but with additional weakref safety (i.e. a strong reference to `obj` will not
    be retained). The return object can be used to
    [`disconnect()`][psygnal.SignalInstance.disconnect], (or you can use
    [`disconnect_setitem()`][psygnal.SignalInstance.disconnect_setitem]).

    Parameters
    ----------
    obj : object
        An object.
    key : str
        Name of the key in `obj` that should be set to the value of this
        signal when emitted
    maxargs : Optional[int]
        max number of positional args to accept
    on_ref_error: {'raise', 'warn', 'ignore'}, optional
        What to do if a weak reference cannot be created.  If 'raise', a
        ReferenceError will be raised.  If 'warn' (default), a warning will be
        issued and a strong-reference will be used. If 'ignore' a strong-reference
        will be used (silently).
    priority : int
        The priority of the callback. This is used to determine the order in which
        callbacks are called when multiple are connected to the same signal.
        Higher priority callbacks are called first. Negative values are allowed.
        The default is 0.

    Returns
    -------
    Tuple
        (weakref.ref, name, callable).  Reference to the object, name of the
        attribute, and setitem closure.  Can be used to disconnect the slot.

    Raises
    ------
    ValueError
        If this is not a single-value signal
    TypeError
        If `obj` does not support __setitem__.

    Examples
    --------
    >>> class T:
    ...     sig = Signal(int)
    >>> t = T()
    >>> my_obj = dict()
    >>> t.sig.connect_setitem(my_obj, "x")
    >>> t.sig.emit(5)
    >>> assert my_obj == {"x": 5}
    """
    if maxargs is _NULL:
        warnings.warn(
            "The default value of maxargs will change from `None` to `1` in"
            "version 0.11. To silence this warning, provide an explicit value for "
            "maxargs (`None` for current behavior, `1` for future behavior).",
            FutureWarning,
            stacklevel=2,
        )
        maxargs = None

    if not hasattr(obj, "__setitem__"):
        raise TypeError(f"Object {obj} does not support __setitem__")

    with self._lock:
        caller = WeakSetitem(
            obj,
            key,
            max_args=cast("int | None", maxargs),
            finalize=self._try_discard,
            on_ref_error=on_ref_error,
            priority=priority,
        )
        self._append_slot(caller)

    return caller

def disconnect(self, slot: Callable | None = None, missing_ok: bool = True) -> None:
    """Disconnect slot from signal.

    Parameters
    ----------
    slot : callable, optional
        The specific slot to disconnect.  If `None`, all slots will be disconnected,
        by default `None`
    missing_ok : Optional[bool]
        If `False` and the provided `slot` is not connected, raises `ValueError.
        by default `True`

    Raises
    ------
    ValueError
        If `slot` is not connected and `missing_ok` is False.
    """
    with self._lock:
        if slot is None:
            # NOTE: clearing an empty list is actually a RuntimeError in Qt
            self._remove_slot("all")
            return

        idx = self._slot_index(slot)
        if idx != -1:
            self._remove_slot(idx)
        elif not missing_ok:
            raise ValueError(f"slot is not connected: {slot}")

def disconnect_setattr(
    self, obj: object, attr: str, missing_ok: bool = True
) -> None:
    """Disconnect a previously connected attribute setter.

    Parameters
    ----------
    obj : object
        An object.
    attr : str
        The name of an attribute on `obj` that was previously used for
        `connect_setattr`.
    missing_ok : bool
        If `False` and the provided `slot` is not connected, raises `ValueError`.
        by default `True`

    Raises
    ------
    ValueError
        If `missing_ok` is `True` and no attribute setter is connected.
    """
    with self._lock:
        cb = WeakSetattr(obj, attr, on_ref_error="ignore")
        self._try_discard(cb, missing_ok)

def disconnect_setitem(
    self, obj: object, key: str, missing_ok: bool = True
) -> None:
    """Disconnect a previously connected item setter.

    Parameters
    ----------
    obj : object
        An object.
    key : str
        The name of a key in `obj` that was previously used for
        `connect_setitem`.
    missing_ok : bool
        If `False` and the provided `slot` is not connected, raises `ValueError`.
        by default `True`

    Raises
    ------
    ValueError
        If `missing_ok` is `True` and no item setter is connected.
    """
    if not hasattr(obj, "__setitem__"):
        raise TypeError(f"Object {obj} does not support __setitem__")

    with self._lock:
        caller = WeakSetitem(obj, key, on_ref_error="ignore")
        self._try_discard(caller, missing_ok)

def emit(
    self, *args: Any, check_nargs: bool = False, check_types: bool = False
) -> None:
    """Emit this signal with arguments `args`.

    !!! note

        `check_args` and `check_types` both add overhead when calling emit.

    Parameters
    ----------
    *args : Any
        These arguments will be passed when calling each slot (unless the slot
        accepts fewer arguments, in which case extra args will be discarded.)
    check_nargs : bool
        If `False` and the provided arguments cannot be successfully bound to the
        signature of this Signal, raise `TypeError`.  Incurs some overhead.
        by default False.
    check_types : bool
        If `False` and the provided arguments do not match the types declared by
        the signature of this Signal, raise `TypeError`.  Incurs some overhead.
        by default False.

    Raises
    ------
    TypeError
        If `check_nargs` and/or `check_types` are `True`, and the corresponding
        checks fail.
    """
    if self._is_blocked:
        return

    if check_nargs:
        try:
            self.signature.bind(*args)
        except TypeError as e:
            raise TypeError(
                f"Cannot emit args {args} from signal {self!r} with "
                f"signature {self.signature}:\n{e}"
            ) from e

    if check_types and not _parameter_types_match(
        lambda: None, self.signature, _build_signature(*[type(a) for a in args])
    ):
        raise TypeError(
            f"Types provided to '{self.name}.emit' "
            f"{tuple(type(a).__name__ for a in args)} do not match signal "
            f"signature: {self.signature}"
        )

    if self._is_paused:
        self._args_queue.append(args)
        return

    if SignalInstance._debug_hook is not None:
        from ._group import EmissionInfo

        SignalInstance._debug_hook(EmissionInfo(self, args))

    self._run_emit_loop(args)

def emit_fast(self, *args: Any) -> None:
    """Fast emit without any checks.

    This method can be up to 10x faster than `emit()`, but it lacks most of the
    features and safety checks of `emit()`. Use with caution.  Specifically:

    - It does not support `check_nargs` or `check_types`
    - It does not use any thread safety locks.
    - It is not possible to query the emitter with `Signal.current_emitter()`
    - It is not possible to query the sender with `Signal.sender()`
    - It does not support "queued" or "latest-only" reemission modes for nested
      emits. It will always use "immediate" mode, wherein signals emitted by
      callbacks are immediately processed in a deeper emission loop.

    It DOES, however, support `paused()` and `blocked()`

    Parameters
    ----------
    *args : Any
        These arguments will be passed when calling each slot (unless the slot
        accepts fewer arguments, in which case extra args will be discarded.)
    """
    if self._is_blocked:
        return

    if self._is_paused:
        self._args_queue.append(args)
        return

    if self._recursion_depth >= RECURSION_LIMIT:
        raise RecursionError(
            f"Psygnal recursion limit ({RECURSION_LIMIT}) reached when emitting "
            f"signal {self.name!r} with args {args}"
        )

    self._recursion_depth += 1
    try:
        for caller in self._slots:
            caller.cb(args)
    except RecursionError as e:
        raise RecursionError(
            f"RecursionError when emitting signal {self.name!r} with args {args}"
        ) from e
    except EmitLoopError as e:  # pragma: no cover
        raise e
    except Exception as cb_err:
        loop_err = EmitLoopError(exc=cb_err, signal=self).with_traceback(
            cb_err.__traceback__
        )
        # this comment will show up in the traceback
        raise loop_err from cb_err  # emit() call ABOVE || callback error BELOW
    finally:
        if self._recursion_depth > 0:
            self._recursion_depth -= 1

def pause(self) -> None:
    """Pause all emission and collect *args tuples from emit().

    args passed to `emit` will be collected and re-emitted when `resume()` is
    called. For a context manager version, see `paused()`.
    """
    self._is_paused = True

def paused(
    self, reducer: ReducerFunc | None = None, initial: Any = _NULL
) -> AbstractContextManager[None]:
    """Context manager to temporarily pause this signal.

    Parameters
    ----------
    reducer : Callable | None
        A optional function to reduce the args collected while paused into a single
        emitted group of args.  If not provided, all emissions will be re-emitted
        as they were collected when the signal is resumed. May be:

        - a function that takes two args tuples and returns a single args tuple.
          This will be passed to `functools.reduce` and is expected to reduce all
          collected/emitted args into a single tuple.
          For example, three `emit(1)` events would be reduced and re-emitted as
          follows: `self.emit(*functools.reduce(reducer, [(1,), (1,), (1,)]))`
        - a function that takes a single argument (an iterable of args tuples) and
          returns a tuple (the reduced args). This will be *not* be passed to
          `functools.reduce`. If `reducer` is a function that takes a single
          argument, `initial` will be ignored.
    initial: any, optional
        initial value to pass to `functools.reduce`

    Examples
    --------
    >>> with obj.signal.paused(lambda a, b: (a[0].union(set(b)),), (set(),)):
    ...     t.sig.emit(1)
    ...     t.sig.emit(2)
    ...     t.sig.emit(3)
    >>> # results in obj.signal.emit({1, 2, 3})
    """
    return _SignalPauser(self, reducer, initial)

def resume(self, reducer: ReducerFunc | None = None, initial: Any = _NULL) -> None:
    """Resume (unpause) this signal, emitting everything in the queue.

    Parameters
    ----------
    reducer : Callable | None
        A optional function to reduce the args collected while paused into a single
        emitted group of args.  If not provided, all emissions will be re-emitted
        as they were collected when the signal is resumed. May be:

        - a function that takes two args tuples and returns a single args tuple.
          This will be passed to `functools.reduce` and is expected to reduce all
          collected/emitted args into a single tuple.
          For example, three `emit(1)` events would be reduced and re-emitted as
          follows: `self.emit(*functools.reduce(reducer, [(1,), (1,), (1,)]))`
        - a function that takes a single argument (an iterable of args tuples) and
          returns a tuple (the reduced args). This will be *not* be passed to
          `functools.reduce`. If `reducer` is a function that takes a single
          argument, `initial` will be ignored.
    initial: any, optional
        initial value to pass to `functools.reduce`

    Examples
    --------
    >>> class T:
    ...     sig = Signal(int)
    >>> t = T()
    >>> t.sig.pause()
    >>> t.sig.emit(1)
    >>> t.sig.emit(2)
    >>> t.sig.emit(3)
    >>> t.sig.resume(lambda a, b: (a[0].union(set(b)),), (set(),))
    >>> # results in t.sig.emit({1, 2, 3})
    """
    self._is_paused = False
    # not sure why this attribute wouldn't be set, but when resuming in
    # EventedModel.update, it may be undefined (as seen in tests)
    if not getattr(self, "_args_queue", None):
        return
    if len(self._slots) == 0:
        self._args_queue.clear()
        return

    if reducer is not None:
        if len(inspect.signature(reducer).parameters) == 1:
            args = cast("ReducerOneArg", reducer)(self._args_queue)
        else:
            reducer = cast("ReducerTwoArgs", reducer)
            if initial is _NULL:
                args = reduce(reducer, self._args_queue)
            else:
                args = reduce(reducer, self._args_queue, initial)
        self._run_emit_loop(args)
    else:
        for args in self._args_queue:
            self._run_emit_loop(args)
    self._args_queue.clear()