Coverage for vibefpmgui/background.py: 100%

1"""Blocking calls (a program run to completion, a slow server) on a pool 

2thread, so the window never waits for them; the result comes back in the 

3GUI thread.""" 

4 

5from __future__ import annotations 

6 

7from collections.abc import Callable 

8 

9from qtpy.QtCore import QObject, QRunnable, QThreadPool, Signal 

10 

11 

12class _Receiver(QObject): 

13 """Brings a background call's result to the GUI thread (it lives there, 

14 so its signal is delivered there) and hands it to ``then``.""" 

15 

16 done = Signal(object) 

17 

18 def __init__(self, then, parent=None): 

19 super().__init__(parent) 

20 self._then = then 

21 self.done.connect(self._deliver) 

22 

23 def _deliver(self, result): 

24 self.deleteLater() 

25 self._then(result) 

26 

27 

28class _Call(QRunnable): 

29 def __init__(self, fn, receiver: _Receiver): 

30 super().__init__() 

31 self.fn, self.receiver = fn, receiver 

32 

33 def run(self): 

34 result = self.fn() 

35 try: 

36 self.receiver.done.emit(result) 

37 except RuntimeError: 

38 pass # (the window that asked has closed) 

39 

40 

41def call_in_background(fn: Callable[[], object], then: Callable[[object], None], parent: QObject): 

42 """Run ``fn()`` on a pool thread, then ``then(result)`` in the GUI thread 

43 (not at all if ``parent`` is gone by then).""" 

44 QThreadPool.globalInstance().start(_Call(fn, _Receiver(then, parent)))