forked from Ulm-IQO/qudi
-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathgeneric_logic.py
61 lines (47 loc) · 2.01 KB
/
generic_logic.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
# -*- coding: utf-8 -*-
"""
This file contains the Qudi logic module base class.
Qudi is free software: you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.
Qudi is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU General Public License for more details.
You should have received a copy of the GNU General Public License
along with Qudi. If not, see <http://www.gnu.org/licenses/>.
Copyright (c) the Qudi Developers. See the COPYRIGHT.txt file at the
top-level directory of this distribution and at <https://github.com/Ulm-IQO/qudi/>
"""
from qtpy import QtCore
from core.module import Base
from core.util.mutex import Mutex
class GenericLogic(Base):
"""A generic logic interface class.
"""
_threaded = True
def __init__(self, **kwargs):
""" Initialzize a logic module.
@param dict kwargs: dict of additional arguments
"""
super().__init__(**kwargs)
self.taskLock = Mutex()
@QtCore.Slot(QtCore.QThread)
def moveToThread(self, thread):
super().moveToThread(thread)
def getModuleThread(self):
""" Get the thread associated to this module.
@return QThread: thread with qt event loop associated with this module
"""
return self._manager.tm._threads['mod-logic-' + self._name].thread
def getTaskRunner(self):
""" Get a reference to the task runner module registered in the manager.
@return object: reference to task runner
If there isno registered task runner, an exception is raised.
"""
with self._manager.lock:
if self._manager.tr is not None:
return self._manager.tr
else:
raise Exception('Tried to access task runner without loading one!')