Source code for napari.utils.events.event_utils
from __future__ import annotations
import logging
import weakref
from typing import TYPE_CHECKING, Any
if TYPE_CHECKING:
from collections.abc import Callable
from typing import Protocol
class Emitter(Protocol):
def connect(self, callback: Callable): ...
def disconnect(self, callback: Callable): ...
_logger = logging.getLogger(__name__)
[docs]
def disconnect_events(emitter, listener):
"""Disconnect all events between an emitter group and a listener.
Parameters
----------
emitter : napari.utils.events.event.EmitterGroup
Emitter group.
listener : Object
Any object that has been connected to.
"""
for em in emitter.emitters.values():
em.disconnect(listener)
def connect_setattr(
emitter: Emitter,
obj,
attr: str,
convert_fun: Callable[[Any], Any] | None = None,
) -> None:
ref = weakref.ref(obj)
if convert_fun:
# Handle passed `convert_func` function to map emitted values to valid
# values accepted for the receiver object attribute.
# A `convert_func` is needed to, for example, map `Qt.CheckState`
# values to boolean ones when a `QCheckBox` value change is connected
# to a layer attribute.
# See napari/napari#8154
def _cb(*value):
if (ob := ref()) is None:
emitter.disconnect(_cb)
return
value = tuple(convert_fun(x) for x in value)
setattr(ob, attr, value[0] if len(value) == 1 else value)
else:
def _cb(*value):
if (ob := ref()) is None:
emitter.disconnect(_cb)
return
setattr(ob, attr, value[0] if len(value) == 1 else value)
emitter.connect(_cb)
# There are scenarios where emitter is deleted before obj.
# Also there is no option to create weakref to QT Signal
# but even if keep reference to base object and signal name it is possible to meet
# problem with C++ "wrapped C/C++ object has been deleted"
# In all of these 3 functions, this should be uncommented instead of using
# the if clause in _cb but that causes a segmentation fault in tests
# weakref.finalize(obj, emitter.disconnect, _cb)
def connect_no_arg(emitter: Emitter, obj, attr: str):
ref = weakref.ref(obj)
def _cb(*_value):
if (ob := ref()) is None:
emitter.disconnect(_cb)
return
getattr(ob, attr)()
emitter.connect(_cb)
# as in connect_setattr
# weakref.finalize(obj, emitter.disconnect, _cb)
def connect_setattr_value(emitter: Emitter, obj, attr: str):
"""To get value from Event"""
ref = weakref.ref(obj)
def _cb(value):
if (ob := ref()) is None:
emitter.disconnect(_cb)
return
setattr(ob, attr, value.value)
emitter.connect(_cb)
# weakref.finalize(obj, emitter.disconnect, _cb)