Coverage for graphica/gui/task_runner.py: 100.0%

1"""時間のかかる関数を別スレッドで動かす。""" 

2import logging 

3from PySide6.QtCore import QThread, Signal 

4 

5logger = logging.getLogger(__name__) 

6 

7 

8class TaskRunner(QThread): 

9 """fn(*args, report_progress, is_cancelled, **kwargs) を別スレッドで動かす。 

10 

11 report_progress(done, total, message="") と is_cancelled() は必ず渡る(使うかは fn の自由)。fn は Qt に依存させない。 

12 fn が普通に戻れば、キャンセル中でも succeeded を出す(キャンセル時に途中までの結果を返す使い方のため。 

13 抑えると succeeded も failed も出ず、呼び出し側が待ち続ける)。 

14 """ 

15 progress = Signal(int, int, str) # (done, total, message) 

16 succeeded = Signal(object) # fn の戻り値 

17 failed = Signal(str) # str(例外) 

18 

19 def __init__(self, fn, *args, parent=None, **kwargs): 

20 super().__init__(parent) 

21 self._fn = fn 

22 self._args = args 

23 self._kwargs = kwargs 

24 

25 def run(self): 

26 def _report_progress(done, total, message=""): 

27 self.progress.emit(done, total, message) 

28 

29 try: 

30 result = self._fn( 

31 *self._args, 

32 report_progress=_report_progress, 

33 is_cancelled=self.isInterruptionRequested, 

34 **self._kwargs, 

35 ) 

36 except Exception as e: 

37 logger.exception("バックグラウンドタスクが失敗しました") 

38 self.failed.emit(str(e)) 

39 return 

40 

41 self.succeeded.emit(result)