from contextlib import contextmanager
import warnings
from typing import List, Union, TYPE_CHECKING, Optional, Sequence
import numpy as np
from typing import List, Union, TYPE_CHECKING, Optional, Sequence, Callable, Any
from pymodaq.control_modules.enums import MoveType
from pymodaq.control_modules.viewer_utility_classes import HW_SETTINGS_KEY as DETECTOR_SETTINGS_KEY
from qtpy.QtCore import QObject, Signal, Slot, QThread, QTimer
from qtpy import QtWidgets
import time
from pymodaq.utils.managers.modules.utils import ModuleType
from pymodaq_utils.enums import enum_checker
from pymodaq_utils.logger import set_logger, get_module_name
from pymodaq_utils import utils
from pymodaq_utils.config import GlobalConfig as Config
from pymodaq_data.data import DataToExport, DataSource, DataDim
from pymodaq_gui.managers.parameter_manager import ParameterManager
from pymodaq_gui.utils import Dock
from pymodaq_gui.messenger import messagebox
from pymodaq.utils.data import DataActuator, DataToActuators
from pymodaq.control_modules.thread_commands import ControlToHardwareMove, ControlToHardwareViewer
if TYPE_CHECKING:
from pymodaq.control_modules.daq_viewer import DAQ_Viewer
from pymodaq.control_modules.daq_move import DAQ_Move
logger = set_logger(get_module_name(__file__))
config = Config()
[docs]
class ModulesManager(QObject, ParameterManager):
"""Class to manage DAQ_Viewers and DAQ_Moves with UI to select some
Easier to connect control modules signals to slots, test, ...
Parameters
----------
detectors: list of DAQ_Viewer
actuators: list of DAQ_Move
selected_detectors: list of DAQ_Viewer
sublist of detectors
selected_actuators: list of DAQ_Move
sublist of actuators
"""
settings_name = 'ModulesManagerSettings'
modules_added_signal = Signal()
detectors_changed = Signal(list)
actuators_changed = Signal(list)
det_done_signal = Signal(DataToExport) # dte here contains DataWithAxes
move_done_signal = Signal(DataToExport) # dte here contains DataActuators
timeout_signal = Signal(list) # list of titles of the modules whose origin wasn't found among
# the received responses (see grab_data/move_actuators)
params = [
{'title': "Probe detectors", 'name': 'probe_detectors', 'type': 'action_led', 'value': False},
{'title': "Probe actuators", 'name': 'probe_actuators', 'type': 'action_led', 'value': False},
{'title': 'Detectors', 'name': 'detectors', 'type': 'itemselect', 'checkbox': True},
{'title': 'Actuators', 'name': 'actuators', 'type': 'itemselect', 'checkbox': True},
# Probe results live in their own group, not nested under the probe buttons above,
# so they render below the selection lists instead of pushing them out of view.
{'title': 'Probed detector data', 'name': 'probe_detectors_results', 'type': 'group', 'children': []},
{'title': 'Probed actuator positions', 'name': 'probe_actuators_results', 'type': 'group', 'children': []},
]
def __init__(self,
detectors: Optional[Sequence['DAQ_Viewer']] = None,
actuators: Optional[Sequence['DAQ_Move']] = None,
selected_detectors: Optional[Sequence['DAQ_Viewer']] = None,
selected_actuators: Optional[Sequence['DAQ_Move']] = None,
parent_name='',
**kwargs):
self.parent_name = parent_name
QObject.__init__(self)
ParameterManager.__init__(self)
if detectors is None:
detectors = []
if actuators is None:
actuators = []
if selected_detectors is None:
selected_detectors = []
if selected_actuators is None:
selected_actuators = []
for mod in selected_actuators:
assert mod in actuators
for mod in selected_detectors:
assert mod in detectors
self.det_done_datas: DataToExport = None
self.det_done_flag = False
self.move_done_positions: DataToExport = None
self.move_done_flag = False
# sigActivated emits the param itself (ActionLedParameter.activate()), but neither
# handler takes any argument, so Qt drops it -- same as any other zero-arg slot.
self.settings.child('probe_detectors').sigActivated.connect(self.probe_detectors)
self.settings.child('probe_actuators').sigActivated.connect(self.probe_actuators)
self._detectors = []
self._actuators = []
self.actuators_connected = False
self.detectors_connected = False
self.set_actuators(actuators, selected_actuators)
self.set_detectors(detectors, selected_detectors)
self.detectors_timeout_timer = QTimer()
self.detectors_timeout_timer.setSingleShot(True)
self.detectors_timeout_timer.timeout.connect(self._on_detectors_timeout)
self.actuators_timeout_timer = QTimer()
self.actuators_timeout_timer.setSingleShot(True)
self.actuators_timeout_timer.timeout.connect(self._on_actuators_timeout)
[docs]
def on_hardware_initialization(self, do_init: bool, module: Union['DAQ_Move', 'DAQ_Viewer']):
""" Bypass method during initialization to assert whether a Master of some slave module already exists"""
if not module.master:
for other_module in self.modules_all:
if other_module.initialized_state and other_module.id == module.id and other_module.master:
module.controller_and_thread.controller = other_module.controller_and_thread.controller
module.controller_and_thread.thread = other_module.controller_and_thread.thread
break
module.init_hardware(do_init)
[docs]
def get_random_id(self) -> int:
""" get a random *id* not already used by a module"""
ids = [mod.id for mod in self.modules_all]
id = np.random.randint(1, 1000)
while id in ids:
id = np.random.randint(1, 1000)
return id
@property
def actuator_timeout(self):
return config('pymodaq', 'actuator', 'polling_timeout_s')
@property
def detector_timeout(self):
return config('pymodaq', 'viewer', 'timeout')
def enable_modules(self, enable=True):
for module in self.modules_all:
module.ui.toolbar.setEnabled(enable)
def __repr__(self):
try:
return f'ModulesManager of "{self.parent_name}" with control modules: {self.get_names(self.modules_all)}'
except AttributeError:
return f'ModulesManager of "{self.parent_name}"'
def show_only_control_modules(self, show: True):
self.settings.child('probe_detectors').show(not show)
self.settings.child('probe_actuators').show(not show)
self.settings.child('probe_detectors_results').show(not show)
self.settings.child('probe_actuators_results').show(not show)
[docs]
@classmethod
def get_names(cls, modules: list[Union['DAQ_Move', 'DAQ_Viewer']]):
"""Get the titles of a list of Control Modules
Parameters
----------
modules: list of DAQ_Move and/or DAQ_Viewer
"""
if not hasattr(modules, '__iter__'):
modules = [modules]
return [mod.title for mod in modules]
[docs]
def get_mods_from_names(self, names, mod=ModuleType.Detector) -> List[Union['DAQ_Move', 'DAQ_Viewer']]:
"""Getter of a list of given modules from their name (title)
Parameters
----------
names: list of str
mod: str
either ModuleType.Detector for DAQ_Viewer modules or ModuleType.Actuator for DAQ_Move modules
"""
mods = []
for name in names:
d = self.get_mod_from_name(name, mod)
if d is not None:
mods.append(d)
return mods
[docs]
def get_mod_from_name(self, name, mod=ModuleType.Detector) -> Union['DAQ_Move', 'DAQ_Viewer', None]:
"""Getter of a given module from its name (title)
Returns None is no control module with this name exists
Parameters
----------
name: str
mod: str
either ModuleType.Detector for DAQ_Viewer modules or ModuleType.Actuator for DAQ_Move modules
"""
if mod == ModuleType.Detector or mod == 'det': #backcompat when comparing to 'det'
modules = self._detectors
elif mod == ModuleType.Actuator or mod == 'act':
modules = self._actuators
elif mod == ModuleType.Control:
modules = self._actuators + self._detectors
else:
return None
if name in self.get_names(modules):
return modules[self.get_names(modules).index(name)]
else:
logger.warning(f'No detector with this name: {name}')
return None
[docs]
def add_modules(self, modules: list[Union['DAQ_Move', 'DAQ_Viewer']]):
""" Add new modules to the manager keeping the initial set of selected actuators or detectors"""
actuators = [act for act in modules if act.__class__.__name__ == 'DAQ_Move']
detectors = [det for det in modules if det.__class__.__name__ == 'DAQ_Viewer']
self.set_actuators(self.actuators_all + actuators, self.actuators)
self.set_detectors(self.detectors_all + detectors, self.detectors)
logger.debug(f"New modules added to the ModulesManager: {modules}"
f"Total list is: {self.modules_name}")
self.modules_added_signal.emit()
[docs]
def set_actuators(self, actuators: list['DAQ_Move'], selected_actuators: list['DAQ_Move']):
"""Populates actuators and the subset to be selected in the UI"""
self._actuators = actuators
self.settings.child('actuators').setValue(dict(all_items=self.get_names(actuators),
selected=self.get_names(selected_actuators)))
[docs]
def set_actuators_from_names(self, actuators: list[str], selected_actuators: list[str]):
"""Populates actuators and the subset to be selected in the UI from their names"""
for act in actuators:
assert act in self.actuators_name
for act in selected_actuators:
assert act in self.actuators_name
self.settings.child('actuators').setValue(dict(all_items=actuators,
selected=selected_actuators))
[docs]
def set_detectors(self, detectors: list['DAQ_Viewer'], selected_detectors: list['DAQ_Viewer']):
"""Populates detectors and the subset to be selected in the UI"""
self._detectors = detectors
self.settings.child('detectors').setValue(dict(all_items=self.get_names(detectors),
selected=self.get_names(selected_detectors)))
[docs]
def set_detectors_from_names(self, detectors: list[str], selected_detectors: list[str]):
"""Populates detectors and the subset to be selected in the UI from their names"""
for det in detectors:
assert det in self.detectors_name
for det in selected_detectors:
assert det in self.detectors_name
self.settings.child('detectors').setValue(dict(all_items=detectors,
selected=selected_detectors))
@property
def detectors(self) -> List['DAQ_Viewer']:
"""Get the list of selected detectors"""
return self.get_mods_from_names(self.selected_detectors_name)
@property
def detectors_all(self) -> List['DAQ_Viewer']:
"""Get/Set the list of all detectors"""
return self._detectors
@detectors_all.setter
def detectors_all(self, detectors: List['DAQ_Viewer']):
self.set_detectors(detectors, [])
@property
def actuators(self) -> List['DAQ_Move']:
"""Get the list of selected actuators"""
return self.get_mods_from_names(self.selected_actuators_name, mod=ModuleType.Actuator)
@property
def actuators_all(self) -> list['DAQ_Move']:
"""Get the list of all actuators"""
return self._actuators
@actuators_all.setter
def actuators_all(self, actuators: List['DAQ_Move']):
self.set_actuators(actuators, [])
@property
def modules(self):
"""Get the list of detectors and actuators"""
return self.detectors + self.actuators
@property
def modules_all(self):
"""Get the list of all detectors and actuators"""
return self.detectors_all + self.actuators_all
@property
def modules_name(self) -> list[str]:
return [mod.title for mod in self.modules_all]
@property
def Ndetectors(self):
"""Get the number of selected detectors"""
return len(self.detectors)
@property
def Nactuators(self):
"""Get the number of selected actuators"""
return len(self.actuators)
@property
def detectors_name(self):
"""Get all the names of the detectors"""
return self.settings.child('detectors').value()['all_items']
@property
def selected_detectors_name(self):
"""Get/Set the names of the selected detectors"""
return self.settings.child('detectors').value()['selected']
@selected_detectors_name.setter
def selected_detectors_name(self, detectors):
if set(detectors).issubset(self.detectors_name):
self.settings.child('detectors').setValue(dict(all_items=self.detectors_name,
selected=detectors))
@property
def actuators_name(self):
"""Get all the names of the actuators"""
return self.settings.child('actuators').value()['all_items']
@property
def selected_actuators_name(self) -> List[str]:
"""Get/Set the names of the selected actuators"""
return self.settings.child('actuators').value()['selected']
@selected_actuators_name.setter
def selected_actuators_name(self, actuators):
if set(actuators).issubset(self.actuators_name):
self.settings.child('actuators').setValue(dict(all_items=self.actuators_name,
selected=actuators))
def value_changed(self, param):
if param.name() == 'detectors':
# Previously probed channels no longer reflect the current selection
self.settings.child('probe_detectors_results').clearChildren()
self.detectors_changed.emit(param.value()['selected'])
elif param.name() == 'actuators':
self.settings.child('probe_actuators_results').clearChildren()
self.actuators_changed.emit(param.value()['selected'])
@contextmanager
def _probing(self, connect, timeout_handler):
"""Shared connect/timeout-watch/disconnect scaffold around a probe action.
Both get_det_data_list (grab_data) and probe_actuators (move_actuators) bracket
their actual hardware call the same way: connect modules, watch timeout_signal for
a reply, always disconnect both afterwards. Only what happens *inside* differs.
Parameters
----------
connect: Callable[[bool], Any]
connect_detectors or connect_actuators
timeout_handler: Callable[[List[str]], Any]
_on_detector_probe_timeout or _on_actuator_probe_timeout
"""
connect()
self.timeout_signal.connect(timeout_handler)
try:
yield
finally:
self.timeout_signal.disconnect(timeout_handler)
connect(False)
[docs]
def probe_detectors(self) -> DataToExport:
"""Do a snap of selected detectors, to populate the Detectors panel's result tree
(probe_detectors_results) and return the data"""
if len(self.detectors) == 0:
return DataToExport(name=__class__.__name__, control_module='DAQ_Viewer')
with self._probing(self.connect_detectors, self._on_detector_probe_timeout):
datas: DataToExport = self.grab_data(Naverage=1)
logger.debug(f'Acquired: {datas.get_full_names()}')
results = self.settings.child('probe_detectors_results')
results.clearChildren()
data_children = []
for data_dim in DataDim.names():
data_from_dim = datas.get_data_from_dim(data_dim)
if len(data_from_dim) != 0:
data_children.append(
{'title': data_dim, 'name': data_dim, 'type': 'group', 'children':[
{'title': dwa.origin, 'name': dwa.get_full_name(), 'type': 'str',
'value': dwa.name, 'readonly': True} for dwa in data_from_dim
]})
results.addChildren(data_children)
return datas
[docs]
def get_det_data_list(self) -> DataToExport:
"""Deprecated alias for probe_detectors(), kept for backward compatibility."""
warnings.warn(
"get_det_data_list() is deprecated, use probe_detectors() instead",
DeprecationWarning, stacklevel=2,
)
return self.probe_detectors()
def _report_probe_timeout(self, probe_param_name: str, kind: str, missing_modules: List[str]):
""" Reset the probe button's LED and report which modules failed to answer, by name """
self.settings.child(probe_param_name).setValue(False)
messagebox(text=f"The following {kind}(s) did not respond in time:\n"
+ "\n".join(missing_modules))
def _on_detector_probe_timeout(self, missing_modules: List[str]):
self._report_probe_timeout('probe_detectors', 'detector', missing_modules)
def _on_actuator_probe_timeout(self, missing_modules: List[str]):
self._report_probe_timeout('probe_actuators', 'actuator', missing_modules)
[docs]
def get_probed_data_full_names(self, dim: DataDim | str = None) -> List[str]:
"""Return full names (origin/name) of probed data, optionally filtered by dim.
Parameters
----------
dim: str, optional
One of 'Data0D', 'Data1D', 'Data2D', 'DataND'. If None, all dims are returned.
Returns
-------
list of str
"""
names = []
if dim is not None:
dim = enum_checker(DataDim, dim)
for det_param in self.settings.child('probe_detectors_results').children():
if dim is None or det_param.name() == dim.name:
names.extend([child.name() for child in det_param.children()])
return names
def _grab_data(self, check_do_override=True, Naverage: Optional[int] = None, **kwargs):
self.det_done_datas = DataToExport(name=__class__.__name__, control_module='DAQ_Viewer')
self._received_data = 0
self.det_done_flag = False
self.settings.child('probe_detectors').setValue(self.det_done_flag)
if check_do_override and 'DataMixer' in self.selected_detectors_name:
overridden_detectors = self.get_mod_from_name(
'DataMixer', ModuleType.Detector).settings.child(
DETECTOR_SETTINGS_KEY, 'overridden_detectors').opts['limits']
else:
overridden_detectors = []
for mod in self.detectors:
if mod.title not in overridden_detectors:
kwargs.update(dict(Naverage=Naverage if Naverage is not None else mod.Naverage))
mod.command_hardware.emit(utils.ThreadCommand(ControlToHardwareViewer.SINGLE, kwargs))
def grab_data_with_callback(self,
check_do_override=True,
Naverage: Optional[int] = None,
callback: Callable[[DataToExport], Any] = None,
do_connect_modules=True,
**kwargs):
if callback is not None:
self.detectors_timeout_timer.setInterval(int(self.detector_timeout))
self.detectors_timeout_timer.start()
self.det_done_signal.connect(callback)
if do_connect_modules:
self.connect_detectors(True)
self._grab_data(check_do_override, Naverage, **kwargs)
def _on_detectors_timeout(self):
missing_detectors = self.get_missing_detectors()
self.timeout_signal.emit(missing_detectors)
logger.error('Timeout Fired during waiting for data to be acquired from: '
f'{", ".join(missing_detectors)}')
def _on_actuators_timeout(self):
missing_actuators = self.get_missing_actuators()
self.timeout_signal.emit(missing_actuators)
logger.error('Timeout Fired during waiting for data to be acquired from: '
f'{", ".join(missing_actuators)}')
[docs]
def grab_data(self, check_do_override=True, Naverage: Optional[int] = None, **kwargs) -> DataToExport:
"""Do a single grab of connected and selected detectors
Parameter
---------
check_do_override: bool
If this is True the signal emission to the DAQ_Viewers will be conditionned to the status of their internal
override_grab_from_extension attribute
Naverage: int, optional
If provided, overrides each detector's own Naverage setting. Useful for probing data shape without averaging.
"""
self._grab_data(check_do_override=check_do_override, Naverage=Naverage, **kwargs)
tzero = time.perf_counter()
while not self.det_done_flag:
# wait for grab done signals to end
QtWidgets.QApplication.processEvents() # mandatory for the det_done_flag boolean to be modified in the corresponding method
if time.perf_counter() - tzero > self.detector_timeout / 1000:
missing_detectors = self.get_missing_detectors()
self.timeout_signal.emit(self.get_missing_detectors())
logger.error('Timeout Fired during waiting for data to be acquired from: '
f'{", ".join(missing_detectors)}')
break
QThread.msleep(10)
return self.det_done_datas
[docs]
def get_missing_detectors(self) -> list[str]:
"""match on origin (stable per-module id) rather than name, since a single
detector emits one DataWithAxes per channel, each possibly named differently
"""
received_origins = {dwa.origin for dwa in self.det_done_datas}
return [det_title for det_title in self.selected_detectors_name
if det_title not in received_origins]
[docs]
def get_missing_actuators(self) -> list[str]:
""" match on origin (stable per-module id) rather than name: dte_act's
`.name` holds the requested actuator's title, which is what gets
stamped as `.origin` on the DataActuator the actuator reports back
"""
received_origins = {dwa.origin for dwa in self.move_done_positions}
return [act_title for act_title in self.selected_actuators_name
if act_title not in received_origins]
[docs]
def grab_datas(self, **kwargs):
""" For back compatibility but use self.grab_data"""
return self.grab_data(**kwargs)
[docs]
def connect_actuators(self, connect=True, slot=None, signal='move_done'):
"""Connect the selected actuators signal to a given or default slot
Parameters
----------
connect: bool
slot: builtin_function_or_method
method or function the chosen signal will be connected to
if None, then the default move_done slot is used
signal: str
What kind of signal is to be used:
* 'move_done' will connect the `move_done_signal` to the slot
* 'current_value' will connect the 'current_value_signal' to the slot
See Also
--------
:meth:`move_done`
"""
if slot is None:
slot = self.move_done
if connect:
for sig in [mod.move_done_signal if signal == 'move_done' else mod.current_value_signal
for mod in self.actuators]:
sig.connect(slot)
else:
for sig in [mod.move_done_signal if signal == 'move_done' else mod.current_value_signal
for mod in self.actuators]:
try:
sig.disconnect(slot)
except TypeError as e:
logger.error(str(e))
self.actuators_connected = connect
[docs]
def connect_detectors(self, connect=True, slot=None):
"""
Connect selected DAQ_Viewers's grab_done_signal to the given slot
Parameters
----------
connect: bool
if True, connect to the given slot (or default slot)
if False, disconnect all detectors (not only the currently selected ones.
This is made because when selected detectors changed if you only disconnect those one,
the previously connected ones will stay connected)
slot: method
A method that should be connected, if None self.det_done is connected by default
"""
if slot is None:
slot = self.det_done
if connect:
for sig in [mod.grab_done_signal for mod in self.detectors]:
sig.connect(slot)
else:
for sig in [mod.grab_done_signal for mod in self.detectors]:
try:
sig.disconnect(slot)
except TypeError as e:
# means the slot was not previously connected
logger.info(f'Could not disconnect grab signal from the {slot} slot', stacklevel=2)
self.detectors_connected = connect
[docs]
def probe_actuators(self):
"""Open a single dialog to set target positions for all selected actuators, move them,
then populate the Actuators panel's result tree (probe_actuators_results)"""
actuators = self.actuators
if not actuators:
return
dialog = QtWidgets.QDialog()
dialog.setWindowTitle('Move actuators to target position')
layout = QtWidgets.QVBoxLayout()
form = QtWidgets.QFormLayout()
spinboxes: dict[str, QtWidgets.QDoubleSpinBox] = {}
for mod in actuators:
spinbox = QtWidgets.QDoubleSpinBox()
spinbox.setRange(-1e9, 1e9)
spinbox.setDecimals(4)
spinbox.setValue(mod._current_value.value())
form.addRow(mod.title, spinbox)
spinboxes[mod.title] = spinbox
layout.addLayout(form)
buttons = QtWidgets.QDialogButtonBox(
QtWidgets.QDialogButtonBox.StandardButton.Ok |
QtWidgets.QDialogButtonBox.StandardButton.Cancel,
)
buttons.accepted.connect(dialog.accept)
buttons.rejected.connect(dialog.reject)
layout.addWidget(buttons)
dialog.setLayout(layout)
if dialog.exec() != QtWidgets.QDialog.DialogCode.Accepted:
return
dte_act = DataToExport('Actuators', control_module='DAQ_MOVE')
for mod in actuators:
dte_act.append(DataActuator(mod.title, data=spinboxes[mod.title].value()))
with self._probing(self.connect_actuators, self._on_actuator_probe_timeout):
self.move_actuators(dte_act)
results = self.settings.child('probe_actuators_results')
results.clearChildren()
for dact in self.move_done_positions:
results.addChild(
{'title': dact.name, 'name': dact.name.replace(' ', '_'),
'type': 'float', 'value': dact.value(), 'readonly': True},
)
[docs]
def test_move_actuators(self):
"""Deprecated alias for probe_actuators(), kept for backward compatibility."""
warnings.warn(
"test_move_actuators() is deprecated, use probe_actuators() instead",
DeprecationWarning, stacklevel=2,
)
return self.probe_actuators()
[docs]
def connect_and_move_actuators(self, dte_act: DataToExport,
mode=MoveType.ABS,
polling=True,
slot=None,
signal='move_done') -> DataToExport:
""" Connect Actuators specified in the dte object and move them either absolute or relative to the
given value
"""
self.selected_actuators_name = [dwa.name for dwa in dte_act]
self.connect_actuators(True, slot=slot, signal=signal)
dte = self.move_actuators(dte_act, mode=mode, polling=polling)
self.connect_actuators(False)
return dte
[docs]
def move_actuators_with_callback(self, dte_act: DataToExport | DataToActuators,
mode: MoveType = MoveType.REL,
callback: Callable[[DataToExport], Any] = None,
do_connect_modules=True):
""" Move actuators defined within a DataToExport to the value included in the DatActuators within
This method will emit a signal to a given callback with a DataToExport containing
DataActuators when the moves are done"""
if isinstance(dte_act, DataToActuators):
mode = dte_act.mode
self.selected_actuators_name = [dwa.name for dwa in dte_act]
if callback is not None:
self.actuators_timeout_timer.setInterval(int(self.actuator_timeout * 1000))
self.actuators_timeout_timer.start()
self.move_done_signal.connect(callback)
if do_connect_modules:
self.connect_actuators(True)
self._move_actuators(dte_act, mode=mode,)
[docs]
def forget_callback(self,
callback : Callable,
module_type=ModuleType.Detector,
disconnect_modules=True):
""" to be called by the caller of self.move_actuators_with_callback or self.grab_data_with_callback
in order to disconnect properly the callback
Optionaly also disconnect each selected actuaor and/or detector to the inner method checking when each move/grab
is done
"""
if module_type == ModuleType.Detector or module_type==ModuleType.Control:
if disconnect_modules:
self.connect_detectors(False)
try:
self.det_done_signal.disconnect(callback)
except TypeError:
pass
if module_type == ModuleType.Actuator or module_type == ModuleType.Control:
if disconnect_modules:
self.connect_actuators(False)
try:
self.move_done_signal.disconnect(callback)
except TypeError:
pass
def _move_actuators(self, dte_act: DataToExport | DataToActuators,
mode: MoveType = MoveType.ABS,):
self.move_done_positions = DataToExport(name=__class__.__name__, control_module='DAQ_Move')
self.move_done_flag = False
self.settings.child('probe_actuators').setValue(self.move_done_flag)
if mode == MoveType.ABS:
command = ControlToHardwareMove.MOVE_ABS
elif mode == MoveType.REL:
command = ControlToHardwareMove.MOVE_REL
else:
logger.error(f'Invalid positioning mode: {mode}')
return self.move_done_positions
if len(dte_act) == self.Nactuators:
for dact in dte_act:
act = self.get_mod_from_name(dact.name, ModuleType.Actuator)
if act is not None:
act.command_hardware.emit(
utils.ThreadCommand(command=command, attribute=[dact, True]))
else:
logger.error('Invalid number of positions compared to selected actuators')
return self.move_done_positions
[docs]
def move_actuators(self, dte_act: DataToExport,
mode=MoveType.ABS,
polling=True,
) -> DataToExport:
"""will apply positions to each currently selected actuators. By Default the mode is absolute but can be
Deprecated, you should use move_actuators_with_callback to avoid using a polling mechanism which uses Qt event loop
processevents in a while loop
Parameters
----------
dte_act: DataToExport
the DataToExport of position to apply. Its length must be equal to the number of selected actuators
mode: str
either MoveType.ABS ('abs') for absolute positioning or MoveType.REL ('rel') for relative
polling: bool (should not be used, prefer the callback version)
if True will wait for the selected actuators to reach their target positions (they have to be
connected to a method checking for the position and letting the programm know the move is done (default
connection is this object `move_done` method)
Returns
-------
DataToExport with the selected actuators's name as key and current actuators's value as value
"""
self._move_actuators(dte_act, mode)
if polling:
tzero = time.perf_counter()
while not self.move_done_flag: # polling move done
QtWidgets.QApplication.processEvents() # mandatory for the det_done_flag boolean to be modified in the corresponding method
if time.perf_counter() - tzero > self.actuator_timeout: # timeout in seconds
missing_actuators = self.get_missing_actuators()
self.timeout_signal.emit(missing_actuators)
logger.error('Timeout Fired during waiting for actuators to be moved: '
f'{", ".join(missing_actuators)}')
break
QThread.msleep(10)
self.move_done_signal.emit(self.move_done_positions)
return self.move_done_positions
def raise_actuator_timeout(self, module_name: str):
pass
def reset_signals(self):
self.move_done_flag = True
self.det_done_flag = True
[docs]
def order_positions(self, positions: DataToExport):
""" Reorder the content of the DataToExport given the order of the selected actuators"""
actuators = self.selected_actuators_name
pos = DataToExport('actuators')
for act in actuators:
pos.append(positions.get_data_from_name(act))
return pos
@Slot(DataActuator)
def move_done(self, data_act: DataActuator):
try:
# match on origin, not name, for the same reason as in move_actuators above
if data_act.origin not in {dact.origin for dact in self.move_done_positions}:
self.move_done_positions.append(data_act)
if len(self.move_done_positions) == len(self.actuators):
self.actuators_timeout_timer.stop()
self.move_done_flag = True
self.settings.child('probe_actuators').setValue(self.move_done_flag)
self.move_done_signal.emit(self.move_done_positions)
except Exception as e:
logger.exception(str(e))
def det_done(self, data: DataToExport):
if self.det_done_datas is not None: # means that somehow data are not initialized so no further processing
self._received_data += 1
if len(data) != 0:
self.det_done_datas.append(data)
if self._received_data == len(self.detectors):
self.detectors_timeout_timer.stop()
self.det_done_flag = True
self.settings.child('probe_detectors').setValue(self.det_done_flag)
self.det_done_signal.emit(self.det_done_datas)
if __name__ == '__main__':
import sys
from pymodaq_gui.qt_utils import mkQApp
from pymodaq.dashboard import create_load_dashboard
app = mkQApp('ModulesManager')
win, dashboard = create_load_dashboard()
win.show()
dashboard.experiment_manager.execute_entry()
dashboard.modules_manager.settings_tree.show()
dashboard.modules_manager.get_det_data_list()
print(dashboard.modules_manager.get_probed_data_full_names())
sys.exit(app.exec())