Source code for qtrio._qt

"""This module provides general Qt related utilities that are not Trio specific."""

import contextlib
import sys
import typing

import typing_extensions

if typing.TYPE_CHECKING or "sphinx_autodoc_typehints" in sys.modules:
    from qts import QtCore

import qtrio._python

class SignalProtocol(typing_extensions.Protocol):
    signal: typing.ClassVar["QtCore.Signal"]

[docs]class Signal: """This is a (nearly) drop-in replacement for :class:`QtCore.Signal`. The useful difference is that it does not require inheriting from :class:`QtCore.QObject`. The not-quite part is that it will be a bit more complicated to change thread affinity of the relevant :class:`QtCore.QObject`. If you need this, maybe just inherit. This signal gets around the normally required inheritance by creating :class:`QtCore.QObject` instances behind the scenes to host the real signals. Just as :class:`QtCore.Signal` uses the Python descriptor protocol to intercept the attribute access, so does this so it can 'redirect' to the signal on the other object. """ _attribute_name: typing.ClassVar[str] = "" def __init__(self, *types: type, name: typing.Optional[str] = None) -> None: from qts import QtCore class _SignalQObject(QtCore.QObject): if name is None: signal = QtCore.Signal(*types) else: signal = QtCore.Signal(*types, name=name) self.object_cls: typing.Type[SignalProtocol] = _SignalQObject @typing.overload def __get__(self, instance: None, owner: object) -> "Signal": ... @typing.overload def __get__(self, instance: object, owner: object) -> "QtCore.SignalInstance": ... def __get__( self, instance: object, owner: object ) -> typing.Union["Signal", "QtCore.SignalInstance"]: if instance is None: return self o = self.object(instance=instance) return o.signal
[docs] def object(self, instance: object) -> "QtCore.QObject": """Get the :class:`QtCore.QObject` that hosts the real signal. This can be called such as ``type(instance).signal_name.object(instance)``. Yes this is non-obvious but you have to do something special to get around the :ref:`descriptor protocol <python:descriptors>` so you can get at this method instead of just having the underlying :class:`QtCore.SignalInstance`. Arguments: instance: The object on which this descriptor instance is hosted. Returns: The signal-hosting :class:`QtCore.QObject`. """ d: typing.Optional[ typing.Dict[typing.Type[SignalProtocol], QtCore.QObject] ] = getattr(instance, self._attribute_name, None) if d is None: d = {} setattr(instance, self._attribute_name, d) o: typing.Optional[QtCore.QObject] = d.get(self.object_cls) if o is None: o = self.object_cls() d[self.object_cls] = o return o
Signal._attribute_name = qtrio._python.identifier_path(Signal) @contextlib.contextmanager def connection( signal: "QtCore.SignalInstance", slot: typing.Callable[..., object] ) -> typing.Generator[ typing.Union[ "QtCore.QMetaObject.Connection", typing.Callable[..., object], "QtCore.SignalInstance", # TODO: ], None, None, ]: """Connect a signal and slot for the duration of the context manager. Args: signal: The signal to connect. slot: The callable to connect the signal to. """ # if you get segfault or sigsegv here, especially from pyside2<5.15.2, make # sure the slot isn't on a non-hashable (frozen will make it hashable) attrs # class. this_connection = signal.connect(slot) import qts if qts.is_pyside_5_wrapper or qts.is_pyside_6_wrapper: # PySide2 presently returns a bool rather than a QMetaObject.Connection # this_connection = slot try: yield this_connection finally: expected_exception: typing.Type[Exception] if qts.is_pyside_5_wrapper or qts.is_pyside_6_wrapper: expected_exception = RuntimeError else: expected_exception = TypeError try: # can we precheck and avoid the exception? signal.disconnect(this_connection) except expected_exception: pass