# Copyright 2019 miruka # This file is part of harmonyqml, licensed under GPLv3. import functools import logging as log import sys import time import traceback from concurrent.futures import Executor, Future from typing import Any, Callable, Deque, Optional, Tuple, Union from PyQt5.QtCore import QObject, pyqtProperty, pyqtSignal, pyqtSlot class PyQtFuture(QObject): gotResult = pyqtSignal("QVariant") def __init__(self, future: Future, running_value: Any = None, parent: Optional[QObject] = None) -> None: super().__init__(parent) self.future = future self.running_value = running_value self._result = None self.future.add_done_callback( lambda future: self.gotResult.emit(future.result()) ) def __repr__(self) -> str: state = ("canceled" if self.cancelled else "running" if self.running else "finished") return "%s(state=%s, value=%r)" % ( type(self).__name__, state, self.value ) @pyqtSlot() def cancel(self): self.future.cancel() @pyqtProperty(bool) def cancelled(self): return self.future.cancelled() @pyqtProperty(bool) def running(self): return self.future.running() @pyqtProperty(bool) def done(self): return self.future.done() @pyqtSlot(result="QVariant") @pyqtSlot(int, result="QVariant") @pyqtSlot(float, result="QVariant") def result(self, timeout: Optional[Union[int, float]] = None): return self.future.result(timeout) @pyqtProperty("QVariant", notify=gotResult) def value(self): return self.future.result() if self.done else self.running_value def add_done_callback(self, fn: Callable[[Future], None]) -> None: self.future.add_done_callback(fn) _Task = Tuple[Executor, Callable, Optional[tuple], Optional[dict]] _RUNNING: Deque[_Task] = Deque() _PENDING: Deque[_Task] = Deque() def futurize(max_running: Optional[int] = None, consider_args: bool = False, discard_if_max_running: bool = False, pyqt: bool = True, running_value: Any = None) -> Callable: def decorator(func: Callable) -> Callable: @functools.wraps(func) def wrapper(self, *args, **kws) -> Optional[PyQtFuture]: task: _Task = ( self.pool, func, args if consider_args else None, kws if consider_args else None, ) def can_run_now() -> bool: if max_running is not None and \ _RUNNING.count(task) >= max_running: log.debug("!! Max %d tasks of this kind running: %r", max_running, task[1:]) return False if not consider_args or not _PENDING: return True log.debug(".. Pending: %r\n Queue: %r", task[1:], _PENDING) candidate_task = next(( pending for pending in _PENDING if pending[0] == self.pool and pending[1] == func ), None) if candidate_task is None: log.debug(">> No other candidate, starting: %r", task[1:]) return True if candidate_task[2] == args and candidate_task[3] == kws: log.debug(">> Candidate is us: %r", candidate_task[1:]) return True log.debug("XX Other candidate: %r", candidate_task[1:]) return False if not can_run_now() and discard_if_max_running: log.debug("\\/ Discarding task: %r", task[1:]) return None def run_and_catch_errs(): if not can_run_now(): log.debug("~~ Can't start now: %r", task[1:]) _PENDING.append(task) while not can_run_now(): time.sleep(0.05) _RUNNING.append(task) log.debug("Starting: %r", task[1:]) # Without this, exceptions are silently ignored try: return func(self, *args, **kws) except Exception: traceback.print_exc() log.error("Exiting thread/process due to exception.") sys.exit(1) finally: del _RUNNING[_RUNNING.index(task)] future = self.pool.submit(run_and_catch_errs) return PyQtFuture( future=future, running_value=running_value, parent=self ) if pyqt else future return wrapper return decorator