Skip to content

Utilities#

These utilities may help when using signals and evented objects.

decompile() #

Mangle names of mypyc-compiled files so that they aren't used.

This function requires write permissions to the psygnal source directory.

Source code in psygnal/utils.py
120
121
122
123
124
125
126
127
def decompile() -> None:
    """Mangle names of mypyc-compiled files so that they aren't used.

    This function requires write permissions to the psygnal source directory.
    """
    for suffix in _COMPILED_EXTS:  # pragma: no cover
        for path in Path(__file__).parent.rglob(f"**/*{suffix}"):
            path.rename(path.with_suffix(f"{suffix}{_BAK}"))

iter_signal_instances(obj, include_private_attrs=False) #

Yield all SignalInstance attributes found on obj.

Parameters:

  • obj (object) –

    Any object that has an attribute that has a SignalInstance (or SignalGroup).

  • include_private_attrs (bool) –

    Whether private signals (starting with an underscore) should also be logged, by default False

Yields:

  • SignalInstance

    SignalInstances (and SignalGroups) found as attributes on obj.

Source code in psygnal/utils.py
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
def iter_signal_instances(
    obj: Any, include_private_attrs: bool = False
) -> Generator[SignalInstance, None, None]:
    """Yield all `SignalInstance` attributes found on `obj`.

    Parameters
    ----------
    obj : object
        Any object that has an attribute that has a SignalInstance (or SignalGroup).
    include_private_attrs : bool
        Whether private signals (starting with an underscore) should also be logged,
        by default False

    Yields
    ------
    SignalInstance
        SignalInstances (and SignalGroups) found as attributes on `obj`.
    """
    for n in dir(obj):
        if include_private_attrs or not n.startswith("_"):
            with suppress(AttributeError, FutureWarning):
                attr = getattr(obj, n)
                if isinstance(attr, SignalInstance):
                    yield attr
                if isinstance(attr, SignalGroup):
                    yield attr._psygnal_relay

monitor_events(obj=None, logger=_default_event_monitor, include_private_attrs=False) #

Context manager to print or collect events emitted by SignalInstances on obj.

Parameters:

  • obj (object, optional) –

    Any object that has an attribute that has a SignalInstance (or SignalGroup). If None, all SignalInstances will be monitored.

  • logger (Callable[[EmissionInfo], None], optional) –

    A optional function to handle the logging of the event emission. This function must take two positional args: a signal name string, and a tuple that contains the emitted arguments. The default logger simply prints the signal name and emitted args.

  • include_private_attrs (bool) –

    Whether private signals (starting with an underscore) should also be logged, by default False

Source code in psygnal/utils.py
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
@contextmanager
def monitor_events(
    obj: Any | None = None,
    logger: Callable[[EmissionInfo], Any] = _default_event_monitor,
    include_private_attrs: bool = False,
) -> Iterator[None]:
    """Context manager to print or collect events emitted by SignalInstances on `obj`.

    Parameters
    ----------
    obj : object, optional
        Any object that has an attribute that has a SignalInstance (or SignalGroup).
        If None, all SignalInstances will be monitored.
    logger : Callable[[EmissionInfo], None], optional
        A optional function to handle the logging of the event emission.  This function
        must take two positional args: a signal name string, and a tuple that contains
        the emitted arguments. The default logger simply prints the signal name and
        emitted args.
    include_private_attrs : bool
        Whether private signals (starting with an underscore) should also be logged,
        by default False
    """
    code = getattr(logger, "__code__", None)
    _old_api = bool(code and code.co_argcount > 1)

    if obj is None:
        # install the hook globally
        if _old_api:
            raise ValueError(
                "logger function must take a single argument (an EmissionInfo instance)"
            )
        before, SignalInstance._debug_hook = SignalInstance._debug_hook, logger
    else:
        if _old_api:
            warn(
                "logger functions must now take a single argument (an instance of "
                "psygnal.EmissionInfo). Please update your logger function.",
                stacklevel=2,
            )
        disconnectors = set()
        for siginst in iter_signal_instances(obj, include_private_attrs):
            if isinstance(siginst, SignalRelay):
                # TODO: ... but why?
                continue

            if _old_api:

                def _report(*args: Any, signal: SignalInstance = siginst) -> None:
                    logger(signal.name, args)  # type: ignore

            else:

                def _report(*args: Any, signal: SignalInstance = siginst) -> None:
                    logger(EmissionInfo(signal, args))

            disconnectors.add(partial(siginst.disconnect, siginst.connect(_report)))

    try:
        yield
    finally:
        if obj is None:
            SignalInstance._debug_hook = before
        else:
            for disconnector in disconnectors:
                disconnector()

recompile() #

Fix all name-mangled mypyc-compiled files so that they ARE used.

This function requires write permissions to the psygnal source directory.

Source code in psygnal/utils.py
130
131
132
133
134
135
136
137
def recompile() -> None:
    """Fix all name-mangled mypyc-compiled files so that they ARE used.

    This function requires write permissions to the psygnal source directory.
    """
    for suffix in _COMPILED_EXTS:  # pragma: no cover
        for path in Path(__file__).parent.rglob(f"**/*{suffix}{_BAK}"):
            path.rename(path.with_suffix(suffix))