Source code for pylablib.core.thread.threadprop

from ..utils import general

from ..gui import QtCore

import threading

local_data=threading.local()

thread_uids=general.NamedUIDGenerator(thread_safe=True)

### Errors ###
[docs] class ThreadError(RuntimeError): """Generic thread error""" def __init__(self, msg=None): msg=msg or "thread error" super().__init__(msg)
[docs] class NoControllerThreadError(ThreadError): """Thread error for a case of thread having no controllers""" def __init__(self, msg=None): msg=msg or "thread has no controller" super().__init__(msg)
[docs] class DuplicateControllerThreadError(ThreadError): """Thread error for a case of a duplicate thread controller""" def __init__(self, msg=None): msg=msg or "trying to create a duplicate thread controller" super().__init__(msg)
[docs] class TimeoutThreadError(ThreadError,TimeoutError): """Thread error for a case of a wait timeout""" def __init__(self, msg=None): msg=msg or "waiting has timed out" super().__init__(msg)
[docs] class NoMessageThreadError(ThreadError): """Thread error for a case of trying to get a non-existing message""" def __init__(self, msg=None): msg=msg or "no message available" super().__init__(msg)
[docs] class SkippedCallError(ThreadError): """Thread error for a case of external call getting skipped (unscheduled)""" def __init__(self, msg=None): msg=msg or "call has been skipped" super().__init__(msg)
### Interrupts ###
[docs] class InterruptException(Exception): """Generic interrupt exception (raised by some function to signal interrupts from other threads)""" def __init__(self, msg=None): msg=msg or "thread interrupt" super().__init__(msg)
[docs] class InterruptExceptionStop(InterruptException): """Interrupt exception denoting thread stop request""" def __init__(self, msg=None): msg=msg or "thread interrupt: stop" super().__init__(msg)
[docs] def get_app(): """Get current application instance""" return QtCore.QCoreApplication.instance()
[docs] def get_gui_thread(): """Get main (GUI) thread, or ``None`` if application is not running""" app=get_app() return app and app.thread()
[docs] def is_gui_running(): """Check if GUI is running""" return get_app() is not None
[docs] def is_gui_thread(): """Check if the current thread is the one running the GUI loop""" if getattr(local_data,"is_gui",False): return True app=get_app() return (app is not None) and (QtCore.QThread.currentThread() is app.thread())
[docs] def current_controller(require_controller=True): """ Get controller of the current thread. If the current thread has not controller `and `require_controller==True``, raise an error; otherwise, return ``None``. """ controller=getattr(local_data,"controller",None) if require_controller and (controller is None): raise NoControllerThreadError("current thread has no controller") return controller