EmissionsΒΆ

import typing

import attr
import qts
from qts import QtCore
from qts import QtGui
from qts import QtWidgets
import qtrio
import trio
import trio_typing


class QSignaledWidget(QtWidgets.QWidget):
    """A :class:`QtWidgets.QWidget` with extra signals for events of interest.

    Attributes:
        closed: A signal that will be emitted after a close event.
    """

    closed = QtCore.Signal()
    shown = QtCore.Signal()

    def closeEvent(self, event: QtGui.QCloseEvent) -> None:
        """Detect close events and emit the ``closed`` signal."""

        super().closeEvent(event)
        if event.isAccepted():
            # TODO: https://bugreports.qt.io/browse/PYSIDE-1318
            if qts.is_pyqt_5_wrapper:
                self.closed.emit()
            elif qts.is_pyside_5_wrapper:
                signal = typing.cast(QtCore.SignalInstance, self.closed)
                signal.emit()
            else:  # pragma: no cover
                raise qtrio.InternalError(
                    "You should not be here but you are running neither PyQt5 nor PySide2.",
                )
        else:  # pragma: no cover
            pass

    def showEvent(self, event: QtGui.QShowEvent) -> None:
        """Detect show events and emit the ``shown`` signal."""

        super().showEvent(event)
        if event.isAccepted():
            # TODO: https://bugreports.qt.io/browse/PYSIDE-1318
            if qts.is_pyqt_5_wrapper:
                self.shown.emit()
            elif qts.is_pyside_5_wrapper:
                signal = typing.cast(QtCore.SignalInstance, self.shown)
                signal.emit()
            else:  # pragma: no cover
                raise qtrio.InternalError(
                    "You should not be here but you are running neither PyQt5 nor PySide2.",
                )
        else:  # pragma: no cover
            pass


@attr.s(auto_attribs=True)
class Widget:
    """A manager for a simple window with increment and decrement buttons to change a
    counter which is displayed via a widget in the center.
    """

    widget: QSignaledWidget = attr.ib(factory=QSignaledWidget)
    increment: QtWidgets.QPushButton = attr.ib(factory=QtWidgets.QPushButton)
    decrement: QtWidgets.QPushButton = attr.ib(factory=QtWidgets.QPushButton)
    label: QtWidgets.QLabel = attr.ib(factory=QtWidgets.QLabel)
    layout: QtWidgets.QHBoxLayout = attr.ib(factory=QtWidgets.QHBoxLayout)
    count: int = 0
    serving_event: trio.Event = attr.ib(factory=trio.Event)

    def setup(self, title: str, parent: typing.Optional[QtWidgets.QWidget]) -> None:
        self.widget.setParent(parent)

        self.widget.setWindowTitle(title)
        self.widget.setLayout(self.layout)

        self.increment.setText("+")
        self.decrement.setText("-")

        self.label.setText(str(self.count))

        self.layout.addWidget(self.decrement)
        self.layout.addWidget(self.label)
        self.layout.addWidget(self.increment)

    def increment_count(self) -> None:
        """Increment the counter and update the label."""

        self.count += 1
        self.label.setText(str(self.count))

    def decrement_count(self) -> None:
        """Decrement the counter and update the label."""

        self.count -= 1
        self.label.setText(str(self.count))

    async def show(self) -> None:
        """Show the primary widget for this window."""

        self.widget.show()

    async def serve(
        self,
        *,
        task_status: trio_typing.TaskStatus[None] = trio.TASK_STATUS_IGNORED,
    ) -> None:
        signals = [
            self.decrement.clicked,
            self.increment.clicked,
            self.widget.closed,
        ]

        async with qtrio.enter_emissions_channel(signals=signals) as emissions:
            await self.show()
            task_status.started()
            self.serving_event.set()

            async for emission in emissions.channel:
                if emission.is_from(self.decrement.clicked):
                    self.decrement_count()
                elif emission.is_from(self.increment.clicked):
                    self.increment_count()
                elif emission.is_from(self.widget.closed):
                    break
                else:  # pragma: no cover
                    raise qtrio.QTrioException(f"Unexpected emission: {emission}")


async def start_widget(
    title: str = "QTrio Emissions Example",
    parent: typing.Optional[QtWidgets.QWidget] = None,
    hold_event: typing.Optional[trio.Event] = None,
    *,
    cls: typing.Type[Widget] = Widget,
    task_status: trio_typing.TaskStatus[Widget] = trio.TASK_STATUS_IGNORED,
) -> None:
    self = cls()
    self.setup(title=title, parent=parent)

    task_status.started(self)

    if hold_event is not None:
        await hold_event.wait()

    await self.serve()