adam/ui/background.py

"""Coalesced background reads; widget updates always return to the Qt thread."""
from __future__ import annotations

import time
from typing import Callable

from PySide6.QtCore import QObject, QRunnable, QThreadPool, Signal, Slot


class _Result(QObject):
    finished = Signal(object, object)


class _Read(QRunnable):
    def __init__(self, read: Callable, result: _Result) -> None:
        super().__init__()
        self.read = read
        self.result = result

    def run(self) -> None:
        try:
            value = self.read()
        except Exception as exc:
            self._emit(None, str(exc))
        else:
            self._emit(value, None)

    def _emit(self, value: object, error: object) -> None:
        """Ignore a late completion after Qt has already torn down the window."""
        try:
            self.result.finished.emit(value, error)
        except RuntimeError:
            # A screenshot/normal application shutdown may delete the receiver
            # while this harmless read is still completing.
            pass


class BackgroundRead(QObject):
    ready = Signal(object)
    failed = Signal(str)

    def __init__(self, parent: QObject, *, max_age: float = 30) -> None:
        super().__init__(parent)
        self.max_age = max_age
        self._completed_at: float | None = None
        self._running = False
        self._pending: Callable | None = None
        # The runnable retains this object if its page closes during a read.
        self._result = _Result()
        self._result.finished.connect(self._finished)

    def request(self, read: Callable, *, force: bool = False) -> None:
        if self._running:
            if force:
                self._pending = read
            return
        if not force and self._completed_at is not None and time.monotonic() - self._completed_at < self.max_age:
            return
        self._running = True
        QThreadPool.globalInstance().start(_Read(read, self._result))

    @Slot(object, object)
    def _finished(self, value: object, error: object) -> None:
        self._running = False
        if self._pending is not None:
            read, self._pending = self._pending, None
            self.request(read, force=True)
            return
        if error is not None:
            self.failed.emit(str(error))
        else:
            self._completed_at = time.monotonic()
            self.ready.emit(value)