from dataclasses import dataclass
from typing import Any
from qtpy import QtCore
from pymodaq_data.h5modules.data_saving import DataBundle
from pymodaq_data import DataToExport
[docs]
def get_thread_params(worker_setting_name: str) -> list[dict]:
return [
{'title': worker_setting_name.replace('_', ' ').title(), 'name': worker_setting_name, 'type': 'group',
'children': [
{'title': 'Worker Running:', 'name': 'worker_running', 'type': 'led', 'value': False, 'readonly': True},
{'title': 'Pending save tasks:', 'name': 'worker_tasks', 'type': 'int', 'value': 0, 'readonly': True,
'tooltip': 'Scan points queued to be written but not yet saved to disk. Usually near 0 since '
'saving is normally faster than acquisition; a sustained non-zero value means the '
'saver is falling behind the scan.'},
]},
]
[docs]
@dataclass
class DataForProcessor:
""" To be subclassed if necessary!"""
node_path: str
[docs]
class ThreadWorker(QtCore.QObject):
worker_setting_name: str = 'worker'
params = get_thread_params(worker_setting_name)
n_jobs_done_signal = QtCore.Signal(str, int)
data_to_save_signal = QtCore.Signal(DataBundle)
data_to_process_signal = QtCore.Signal(DataForProcessor)
data_processed_signal = QtCore.Signal(DataToExport)
name = 'ThreadWorker'
def __init__(self, parent = None):
""" Base class for any worker to be moved in a thread"""
super().__init__(parent)
self._n_jobs_done = 0
[docs]
class WorkerThreadManager(QtCore.QObject):
def __init__(self, parent=None):
super().__init__(parent)
self.worker_threads: dict[str, QtCore.QThread] = {}
self.workers: dict[str, ThreadWorker] = {}
self.n_jobs: dict[str, int] = {}
self.current_name: str = None
@property
def worker_thread(self) -> QtCore.QThread:
return self.worker_threads.get(self.current_name, None)
@property
def worker(self) -> ThreadWorker:
return self.workers.get(self.current_name, None)
[docs]
def get_worker(self, name: str) -> ThreadWorker:
return self.workers.get(name, None)
[docs]
def get_thread(self, name: str) -> QtCore.QThread:
return self.worker_threads.get(name, None)
[docs]
def create_thread_for_worker(self, name: str,
worker: ThreadWorker,
delete_if_exists=True,
start_thread=False,
thread: QtCore.QThread = None) -> QtCore.QThread:
""" Create a new thread (or return an existing one) for a worker, and move the worker to it
I
t is up to you to connect the worker methods with your main app using Signal/Slot connections
Do not use direct method call, otherwise the method will be executed in the calling thread
"""
self.current_name = name
if delete_if_exists and name in self.worker_threads:
if self.worker_threads[name].isRunning():
self.exit_worker_thread(name)
if name not in self.worker_threads:
if thread is None:
thread = QtCore.QThread()
self.worker_threads[name] = thread
self.workers[name] = worker
self.workers[name].moveToThread(self.worker_threads[name])
self.n_jobs[name] = 0
if start_thread:
self.worker_threads[name].start()
return self.worker_threads.get(name)
[docs]
def exit_worker_threads(self, delete_worker=False):
while len(self.worker_threads) > 0:
self.exit_worker_thread(self.get_last_name(),
delete_worker=delete_worker)
[docs]
def exit_runner_thread(self, duration: int = 5000):
""" for back compatibility """
self.exit_worker_thread(self.current_name, duration)
@property
def runner_thread(self) -> QtCore.QThread:
""" for back compatibility """
return self.get_thread(self.get_last_name())
@runner_thread.setter
def runner_thread(self, runner: QtCore.QThread):
self.create_thread_for_worker('default', None,
delete_if_exists=False,
start_thread=False,
thread=runner)
[docs]
def exit_worker_thread(self,
runner_name: str = None,
duration : int = 5000,
delete_worker=False):
if runner_name is None:
runner_name = self.current_name
runner_thread = self.worker_threads.pop(runner_name, None)
worker = self.workers.pop(runner_name, None)
if runner_thread is not None:
runner_thread.quit()
terminated = runner_thread.wait(duration)
if not terminated:
runner_thread.terminate()
runner_thread.wait()
runner_thread.deleteLater()
if delete_worker:
worker.deleteLater()
self.current_name = self.get_last_name()
[docs]
def get_last_name(self) -> str | None:
names = list(self.worker_threads.keys())
if len(names) > 0:
return names[-1]
else:
return None
[docs]
def start_thread(self, name: str = None):
if name is None:
name = self.current_name
self.get_thread(name).start()