import numbers
from pymodaq.control_modules.daq_move_ui.utils import UiType
from pymodaq.control_modules.enums import MoveType
HW_KIND = 'actuator'
HW_SETTINGS_KEY = f'{HW_KIND}_settings'
from abc import abstractmethod
from time import perf_counter
from typing import Any, Union, List, Dict, TYPE_CHECKING, Optional, TypeVar, cast
from easydict import EasyDict as edict
import numpy as np
from qtpy.QtCore import Slot, Signal, QTimer
from pint.errors import OffsetUnitCalculusError
from pymodaq_utils.utils import ThreadCommand, find_keys_from_val
from pymodaq_utils.config import GlobalConfig as Config
from pymodaq_utils.logger import set_logger, get_module_name
from pymodaq_utils.enums import BaseEnum
from pymodaq_data.data import DataUnitError, Q_, Unit
from pymodaq_gui.parameter import Parameter
from pymodaq_gui.qt_utils import mkQApp
from pymodaq_utils.warnings import deprecation_msg
from pymodaq.utils.data import DataActuator
from pymodaq.control_modules.thread_commands import ThreadStatus, ThreadStatusMove
from pymodaq.control_modules.daq_move_ui.factory import ActuatorUIFactory
from pymodaq.control_modules.utils import (create_controller_param, create_remote_connection_params,
ControllerStatus)
from pymodaq.control_modules.plugin_base import PluginBase
if TYPE_CHECKING:
from pymodaq.control_modules.daq_move import ActuatorWorker
logger = set_logger(get_module_name(__file__))
config = Config()
HardwareController = TypeVar("HardwareController")
def check_units(dwa: DataActuator, units: str):
""" Check if dwa units is compatible with the units argument
If it is incompatible and has dimensionless units, brute force change the dwa units to units,
otherwise raise a DataUnitError
Parameters
----------
dwa: DataActuator
units: str
Returns
-------
DataActuator
"""
if Unit(dwa.units).is_compatible_with(units):
return dwa
elif Unit(dwa.units).dimensionless: # dimensionless
dwa.force_units(units)
return dwa
else:
raise DataUnitError(f'Units incompatibility between {dwa} and "{units}" units')
class DataActuatorType(BaseEnum):
"""Enum for new or old style holding the value of the actuator"""
float = 0
DataActuator = 1
def comon_parameters(epsilon: Union[float, List[float], Dict[str, float]] = config('pymodaq', 'actuator',
'epsilon_default'),
epsilons: Optional[Union[float, List[float], Dict[str, float]]] = None) -> list[dict]:
"""Common and mandatory parameters of the actuator plugins: units, epsilon, timeout, bounds and scaling
Parameters
----------
epsilon: float
Value (or list/dict of values, in which case the first one is used) below which the target
position is considered reached.
epsilons: float or list or dict, optional
Has priority over ``epsilon`` if given.
Returns
-------
list of dict
The definition of the parameters, to be used to build a ``Parameter``.
"""
if epsilons is not None:
epsilon = epsilons
if isinstance(epsilon, list):
epsilon = epsilon[0]
elif isinstance(epsilon, dict):
epsilon = epsilon[list(epsilon.keys())[0]]
return [{'title': 'Units:', 'name': 'units', 'type': 'str', 'value': '', 'readonly': True},
{'title': 'Epsilon:', 'name': 'epsilon', 'type': 'float',
'value': epsilon,
'tip': 'Differential Value at which the controller considers it reached the target position'},
{'title': 'Timeout (s):', 'name': 'timeout', 'type': 'int',
'value': config('pymodaq', 'actuator', 'polling_timeout_s')},
{'title': 'Bounds:', 'name': 'bounds', 'type': 'group', 'children': [
{'title': 'Set Bounds:', 'name': 'is_bounds', 'type': 'bool', 'value': False},
{'title': 'Min:', 'name': 'min_bound', 'type': 'float', 'value': 0, 'default': 0},
{'title': 'Max:', 'name': 'max_bound', 'type': 'float', 'value': 1, 'default': 1}]},
{'title': 'Scaling:', 'name': 'scaling', 'type': 'group', 'children': [
{'title': 'Use scaling:', 'name': 'use_scaling', 'type': 'bool', 'value': False,
'default': False},
{'title': 'Scaling factor:', 'name': 'scaling', 'type': 'float', 'value': 1., 'default': 1.},
{'title': 'Offset factor:', 'name': 'offset', 'type': 'float', 'value': 0., 'default': 0.}]}]
class MoveCommand:
"""Utility class to contain a given move type and value
Attributes
----------
move_type: pymodaq.control_modules.enums.MoveType | str
either:
* 'abs' or MoveType.ABS: performs an absolute action
* 'rel' or MoveType.REL: performs a relative action
* 'home or MoveType.HOME': find the actuator's home
value: float
the value the move should reach
"""
def __init__(self, move_type: MoveType | str, value: float = 0):
"""Raises a ValueError if ``move_type`` is not one of the :class:`MoveType` names"""
if move_type not in MoveType.names():
raise ValueError(f'The allowed move types for an actuator are {MoveType.names()}')
self.move_type = move_type
self.value = value
def comon_parameters_fun(is_multiaxes=False, axes_names=None,
axis_names: Union[List, Dict] = [],
master=True,
epsilon: Union[float, List[float], Dict[str, float]] = config('pymodaq', 'actuator',
'epsilon_default')):
"""Function returning the common and mandatory parameters that should be on the actuator plugin level
Parameters
----------
is_multiaxes: bool (deprecated, useless)
If True, display the particular settings to define which axis the controller is driving
axes_names: deprecated, use axis_names
axis_names: (deprecated, useless as the get_class_axis method is used now))
list of str or dictionnary of string as key and integer as value
The string identifier of every axis the controller can drive
master: bool (deprecated useless)
If True consider this plugin has to init the controller, otherwise use an already initialized instance
epsilon: float or list of float or dict of float
deprecated (< 5.0.0) no more used here
"""
axis_names = DAQ_Move_base.get_class_axis_names(axis_names, axes_names)
if isinstance(axis_names, list):
axis_name = axis_names[0]
elif isinstance(axis_names, dict):
axis_name = axis_names[list(axis_names.keys())[0]]
else:
raise ValueError('axis_names should be either a list of string or a dict with strings '
'as keys')
controller_status_param = create_controller_param(axis_name, axis_names.copy())
params = [controller_status_param] + comon_parameters(epsilon)
return params
params = [
{'title': 'Main Settings:', 'name': 'main_settings', 'type': 'group', 'children': [
{'title': 'Actuator type:', 'name': 'move_type', 'type': 'str', 'value': '', 'readonly': True},
{'title': 'Actuator name:', 'name': 'module_name', 'type': 'str', 'value': '', 'readonly': True},
{'title': 'UI type:', 'name': 'ui_type', 'type': 'list',
'value': config('pymodaq', 'actuator', 'ui') if config('pymodaq', 'actuator', 'ui') in ActuatorUIFactory.keys() else
ActuatorUIFactory.keys()[0],
'limits': ActuatorUIFactory.keys()},
{'title': 'Refresh value (ms):', 'name': 'refresh_timeout', 'type': 'int',
'value': config('pymodaq', 'actuator', 'refresh_timeout_ms')},
{'title': 'Value Green:', 'name': 'default_value_green', 'type': 'float',
'value': config('pymodaq', 'actuator', 'default_value_green')},
{'title': 'Value Red:', 'name': 'default_value_red', 'type': 'float',
'value': config('pymodaq', 'actuator', 'default_value_red')},
{'title': 'Value Relative:', 'name': 'default_value_relative', 'type': 'float',
'value': config('pymodaq', 'actuator', 'default_value_relative')},
] + create_remote_connection_params()},
{'title': 'Actuator Settings:', 'name': HW_SETTINGS_KEY, 'type': 'group'}
]
def main(plugin_file, init=True, title='test'):
"""
this method start a DAQ_Move object with this defined plugin as actuator
Returns
-------
"""
import sys
from pathlib import Path
from pymodaq.control_modules.instruments import find_actuator_class_from_name
from pymodaq.utils.gui_utils.loader_utils import create_load_daq_move
act = Path(plugin_file).stem.split('daq_move_')[1]
class_ = find_actuator_class_from_name(act)
if hasattr(class_, 'ui_type'):
ui_identifier = class_.ui_type
else:
ui_identifier = config('pymodaq', 'actuator', 'ui')
app = mkQApp("PyMoDAQ Move")
shared_ui, daq_move = create_load_daq_move(ui_identifier)
daq_move.actuator = act
shared_ui.show()
sys.exit(app.exec())
[docs]
class DAQ_Move_base(PluginBase):
""" The base class to be inherited by all actuator modules
This base class implements all necessary parameters and methods for the plugin to communicate with its parent (the
DAQ_Move module)
Parameters
----------
parent : ActuatorWorker
params_state : Parameter
pyqtgraph Parameter instance from which the module will get the initial settings (as defined in the experiment)
Attributes
----------
move_done_signal: Signal
signal represented by a float. Is emitted each time the hardware reached the target position within the epsilon
precision (see comon_parameters variable)
controller: object
the object representing the hardware in the plugin. Used to access hardware functionality
settings: Parameter
instance representing the hardware settings defined from the params attribute. Modifications on the GUI settings
will be transferred to this attribute. It stores at all times the current state of the hardware/plugin settings
params: List of dict used to create a Parameter object.
Its definition on the class level enable the automatic update of the GUI settings when changing plugins
(even in managers mode creation). To be populated on the plugin level as the base class does't represents a
real hardware
is_multiaxes: bool
class level attribute. Defines if the plugin controller controls multiple axes. If True, one has to define
a Master instance of this plugin and slave instances of this plugin (all sharing the same controller_ID
parameter)
current_value: DataActuator
stores the current position after each call to the get_actuator_value in the plugin
target_value: DataActuator
stores the target position the controller should reach within epsilon
"""
move_done_signal = Signal(DataActuator)
is_multiaxes = False
stage_names: list[str] = [] # deprecated
_axis_names: Optional[Union[List[str], Dict[str, int]]] = None
_controller_units: Union[str, List[str], Dict[str, str]] = ''
_epsilons: Optional[Union[float, List[float], Dict[str, float]]] = None
_epsilon = 1.0 # deprecated
params: list[dict] = []
data_actuator_type = DataActuatorType.float # for backcompatibility, but new plugins should have DataActuatorType.DataActuator
ui_type = UiType.NONE # should precise (force if possible) what should be the ui type to be used with this
# actuator. If NONE, PyMoDAQ will use the default ui type (see preferences).
supported_platforms: Optional[tuple] = None # systems the plugin works on, values of sys.platform ('win32', 'linux',
# 'darwin'). None: all of them. PyMoDAQ refuses to initialize the plugin on another system.
supported_architectures: Optional[tuple] = None # processor architectures the plugin works on, values of
# platform.machine() (for instance ('aarch64', 'armv7l') for a Raspberry Pi). None: all of them.
has_encoder = True # tell PyMoDAQ if this actuator is able to set an absolute position and read the controller
# value. If False, you should consider having ui_type = UiType.RELATIVE
data_shape = (1,) # expected shape of the underlying actuator's value (in general a float so shape = (1, ))
[docs]
@classmethod
def get_class_axis_names(cls, axis_names: Optional[Union[List[str], Dict[str, int]]] = None,
deprecated_names: Optional[List[str]] = None) -> Union[List[str], Dict[str, int]]:
""" Convenience method to access the declared axis in a given plugin
Handles some old declaration style and eventual attribute as None or empty string
Parameters
----------
axis_names: list of str or dict of str to int, optional
The names to use instead of the class attribute ``_axis_names`` (default).
deprecated_names: list of str, optional
The names to use instead of the deprecated class attribute ``stage_names`` (default).
Returns
-------
list of str or dict of str to int
The axis names, ``['']`` if none is declared.
"""
if axis_names is None:
axis_names = cls._axis_names
if deprecated_names is None:
deprecated_names = cls.stage_names
_axis_names: Optional[Union[List[str], Dict[str, int]]] = None
if axis_names is not None and len(deprecated_names) != 0:
#check for old and deprecated plugins
_axis_names = deprecated_names
deprecation_msg("using 'stage_names' class attribute in plugins is deprecated, please use"
"'_axis_names' instead" )
if axis_names is not None and len(axis_names) != 0:
# this _axis_names attribute has priority hence eventual overwriting of stage_names
_axis_names = axis_names
else:
_axis_names = [''] if (_axis_names is None or _axis_names == []) else _axis_names
return _axis_names
def __init__(self, parent: Optional['ActuatorWorker'] = None,
params_state: Optional[dict] = None,
**kwargs):
super().__init__(parent, params_state)
self._title = self._title if parent is not None else "myactuator"
self.move_is_done = False
self.stage = None
self.status['stage'] = None
self._ispolling = True
self._axis_units: Union[Dict[str, str], List[str]] = None # type: ignore[assignment] # set just below
if isinstance(self._controller_units, str):
self.axis_units = self._controller_units
else:
self.axis_units = self._controller_units.copy()
if self._epsilons is None:
self._epsilons = self._epsilon
self.epsilons = self._epsilons
self.axis_name = self.axis_name # to trigger some actions on units and epsilons
self._current_value = DataActuator(self._title,
data=[np.zeros(self.data_shape, dtype=float)],
units=self.axis_unit)
self._target_value = DataActuator(self._title,
data=[np.zeros(self.data_shape, dtype=float)],
units=self.axis_unit)
self.poll_timer = QTimer(self)
self.poll_timer.setInterval(config('pymodaq', 'actuator', 'polling_interval_ms'))
self.poll_timer.timeout.connect(self.check_target_reached)
self.ini_attributes()
@property
def axis_unit(self) -> str:
""" Get/set the unit of the currently chosen axis
Will update the printed controller unit in the UI
New in 4.4.0
"""
return self.axis_units[self.axis_index_key] # type: ignore[index] # int for a list, str for a dict
@axis_unit.setter
def axis_unit(self, unit: str):
self.axis_units[self.axis_index_key] = unit # type: ignore[index] # int for a list, str for a dict
self.settings.child('units').setValue(unit)
self.emit_status(ThreadCommand(ThreadStatusMove.UNITS, unit))
@property
def axis_units(self) -> Union[List[str], Dict[str, str]]:
""" Get/Set the units for each axis of the controller
New in 4.4.0
"""
return self._axis_units
@axis_units.setter
def axis_units(self, units: Union[str, List[str], Dict[str, str]]):
if isinstance(units, str):
units_tmp: Union[List[str], Dict[str, str]]
if isinstance(self.axis_names, list):
units_tmp = [units for _ in range(len(self.axis_names))]
else:
units_tmp = {}
for key in self.axis_names:
units_tmp[key] = units
else:
if not isinstance(units, type(self.axis_names)):
raise TypeError('units should be defined just like axis_names: a str, list of string or'
'dict of string')
if len(units) != len(self.axis_names):
raise ValueError('Units should be defined either as a single str or a list/dict with'
'a str defined for each axis')
units_tmp = units
self._axis_units = units_tmp
@property
def epsilon(self) -> float:
""" Get/Set the epsilon of the currently chosen axis
New in 4.4.0
"""
return self.epsilons[self.axis_index_key] # type: ignore[index] # int for a list, str for a dict
@epsilon.setter
def epsilon(self, eps: float):
self.epsilons[self.axis_index_key] = eps # type: ignore[index] # int for a list, str for a dict
@property
def epsilons(self) -> Union[List[float], Dict[str, float]]:
""" Get/Set the epsilon for each axis of the controller
New in 4.4.0
"""
return cast(Union[List[float], Dict[str, float]], self._epsilons)
@epsilons.setter
def epsilons(self, epsilons: Union[float, List[float], Dict[str, float]]):
epsilons_tmp: Union[List[float], Dict[str, float]]
if isinstance(epsilons, numbers.Number):
eps = cast(float, epsilons)
if isinstance(self.axis_names, list):
epsilons_tmp = [eps for _ in range(len(self.axis_names))]
else:
epsilons_tmp = {}
for key in self.axis_names:
epsilons_tmp[key] = eps
else:
if not isinstance(epsilons, type(self.axis_names)):
raise TypeError('units should be defined just like axis_names: a float, list of '
'float or dict of float')
if len(epsilons) != len(self.axis_names):
raise ValueError('epsilons should be defined either as a single float or a '
'list/dict with'
'a float defined for each axis')
epsilons_tmp = epsilons
self._epsilons = epsilons_tmp
@property
def controller_units(self):
""" Get/Set the units of the currently chosen axis of the controller
Deprecated with pymodaq >= 4.4.0
The property controller_units is deprecated please use the axis_unit property
"""
deprecation_msg(f'The property controller_units is deprecated please use the'
f'axis_unit property.')
return self.axis_unit
@controller_units.setter
def controller_units(self, units: str = ''):
deprecation_msg('The property controller_units is deprecated please use the'
'axis_unit property.')
self._axis_units[self.axis_index_key] = units # type: ignore[index] # int for a list, str for a dict
@property
def axis_name(self) -> Union[str]:
"""Get/Set the current axis using its string identifier"""
limits = self.settings.child('controller', 'axis').opts['limits']
if isinstance(limits, list):
return self.settings['controller', 'axis']
elif isinstance(limits, dict):
return find_keys_from_val(limits, val=self.settings['controller', 'axis'])[0]
else:
return ''
@axis_name.setter
def axis_name(self, name: str):
limits = self.settings.child('controller', 'axis').opts['limits']
if name in limits:
if isinstance(limits, list):
self.settings.child('controller', 'axis').setValue(name)
elif isinstance(limits, dict):
self.settings.child('controller', 'axis').setValue(limits[name])
self.axis_unit = self.axis_unit
self.settings.child('epsilon').setValue(self.epsilon)
if self.controller is not None:
self._current_value = cast(DataActuator, self.get_actuator_value())
@property
def axis_names(self) -> Union[List, Dict]:
""" Get/Set the names of all axes controlled by this instrument plugin
Returns
-------
List of string or dictionary mapping names to integers
"""
return self.settings.child('controller', 'axis').opts['limits']
@axis_names.setter
def axis_names(self, names: Union[List, Dict]):
self.settings.child('controller', 'axis').setLimits(names)
@property
def axis_value(self) -> Any:
"""Get the current value selected from the current axis
In case axis_names is a list, return the element of the list: self.axis_name (a str)
In case axis_names is a dict, return the value of the dict self.axis_names[self.axis_name] (an int)
"""
if isinstance(self.axis_names, list):
return self.axis_name
else:
return self.axis_names[self.axis_name]
@property
def axis_index_key(self) -> Union[int, str]:
""" Get the current index or key correspondingto the current axis
In case axis_names is a list, return the index wihtin the list
In case axis_names is a dict, return the key of the dict self.axis_name
"""
if isinstance(self.axis_names, list):
return self.axis_names.index(self.axis_name)
else:
return self.axis_name
[docs]
def ini_attributes(self):
"""To be subclassed, in order to init specific attributes needed by the real implementation."""
pass
def ini_stage_init(
self,
old_controller: Optional[HardwareController] = None,
new_controller: Optional[HardwareController] = None,
slave_controller: Optional[HardwareController] = None,
) -> Optional[HardwareController]:
"""Manage the Master/Slave controller issue
First initialize the status dictionary
Then check whether this stage is controlled by a multiaxe controller (to be defined for each plugin)
if it is a multiaxes controller then:
* if it is Master: init the controller here
* if it is Slave: use an already initialized controller (defined in the experiment of the dashboard)
Parameters
----------
old_controller: object
The particular object that allow the communication with the hardware, in general a python wrapper around the
hardware library. In case of Slave this one comes from a previously initialized plugin
new_controller: object
The particular object that allow the communication with the hardware, in general a python wrapper around the
hardware library. In case of Master it is the new instance of your plugin controller
"""
if old_controller is None and slave_controller is not None:
old_controller = slave_controller
self.status.update(edict(info="", controller=None, initialized=False))
controller: Optional[HardwareController]
if not self.is_master:
if old_controller is None:
raise Exception('no controller has been defined externally while this axe '
'is a slave one')
else:
controller = old_controller
else: # Master stage
controller = new_controller
self.controller = controller
return controller
@property
def current_value(self) -> DataActuator:
""" Get/Set the last known value of the actuator
A DataActuator. For deprecated old style plugins (``data_actuator_type`` is ``DataActuatorType.float``) a
float is returned instead. When set with a number or an array, it is converted into a DataActuator
using the current axis unit; a DataActuator without units gets the axis unit.
"""
if self.data_actuator_type == self.data_actuator_type.float:
return self._current_value.value() # type: ignore[return-value] # float for the deprecated old style plugins
else:
return self._current_value
@current_value.setter
def current_value(self, value: Union[int, float, np.ndarray, DataActuator]):
if isinstance(value, (numbers.Number, np.ndarray)):
self._current_value = DataActuator(self._title, data=value,
units=self.axis_unit)
else:
value = cast(DataActuator, value) # numbers (floats included) are handled above
if (not Unit(self.axis_unit).is_compatible_with(
Unit(value.units)) and
value.units == ''):
value.force_units(self.axis_unit)
self._current_value = value
@property
def target_value(self) -> DataActuator:
""" Get/Set the value the actuator should reach (within epsilon)
A DataActuator. For deprecated old style plugins (``data_actuator_type`` is ``DataActuatorType.float``) a
float is returned instead. Same conversions as :attr:`current_value` when set.
"""
if self.data_actuator_type.name == self.data_actuator_type.float:
return self._target_value.value() # type: ignore[return-value] # float for the deprecated old style plugins
else:
return self._target_value
@target_value.setter
def target_value(self, value: Union[int, float, DataActuator]):
if isinstance(value, numbers.Number):
self._target_value = DataActuator(self._title, data=value,
units=self.axis_unit)
else:
value = cast(DataActuator, value) # numbers (floats included) are handled above
if (not Unit(self.axis_unit).is_compatible_with(
Unit(value.units)) and
value.units == ''):
value.force_units(self.axis_unit)
self._target_value = value
@property
def current_position(self) -> DataActuator:
"""Deprecated, use :attr:`current_value`"""
deprecation_msg('current_position attribute should not be used, use current_value')
return self.current_value
@current_position.setter
def current_position(self, value):
self.current_value = value
@property
def target_position(self) -> DataActuator:
"""Deprecated, use :attr:`target_value`"""
deprecation_msg('target_position attribute should not be used, use target_value')
return self.target_value
@target_position.setter
def target_position(self, value):
self.target_value = value
@property
def is_master(self) -> bool:
""" Get the controller master/slave status
new in version 4.3.0
"""
return self.settings['controller', 'controller_status'] == ControllerStatus.MASTER
@property
def ispolling(self) -> bool:
""" Get/Set the polling status
If True (default), after a move the actuator value is polled until the target is reached (see
:meth:`check_target_reached`), otherwise the plugin has to call :meth:`move_done` itself.
"""
return self._ispolling
@ispolling.setter
def ispolling(self, polling=True):
self._ispolling = polling
[docs]
def check_bound(self, position: DataActuator) -> DataActuator:
""" Check if the current position is within the software bounds
Return the new position eventually coerced within the bounds
"""
if self.settings['bounds', 'is_bounds']:
if self.data_actuator_type == DataActuatorType.DataActuator:
for data_array in position:
if np.any(data_array > self.settings['bounds', 'max_bound']) or \
np.any(data_array < self.settings['bounds', 'min_bound']):
self.emit_status(ThreadCommand('outofbounds'))
data_array[data_array > self.settings['bounds', 'max_bound']] = self.settings['bounds', 'max_bound']
data_array[data_array < self.settings['bounds', 'min_bound']] = self.settings['bounds', 'min_bound']
else:
if position > self.settings['bounds', 'max_bound']:
self.emit_status(ThreadCommand('outofbounds'))
position = self.settings['bounds', 'max_bound']
elif position < self.settings['bounds', 'min_bound']:
self.emit_status(ThreadCommand('outofbounds'))
position = self.settings['bounds', 'min_bound']
return position
[docs]
@abstractmethod
def get_actuator_value(self) -> DataActuator:
"""Read the current value of the actuator from the hardware (to be implemented in the plugin)
Returns
-------
DataActuator
the value in the axis unit (a float is still accepted for deprecated old style plugins, see
``data_actuator_type``).
"""
if hasattr(self, 'check_position'):
deprecation_msg('check_position method in plugins is deprecated, use get_actuator_value',3)
return self.check_position() # type: ignore
else:
raise NotImplementedError
[docs]
@abstractmethod
def close(self) -> None:
"""Terminate the communication with the hardware (to be implemented in the plugin)"""
raise NotImplementedError
[docs]
def move_abs(self, value: Union[float, DataActuator]) -> None:
"""Move the actuator to the absolute position ``value`` (to be implemented in the plugin)
Call :meth:`move_done` when the move is finished if polling is not used (see :attr:`ispolling`).
"""
if hasattr(self, 'move_Abs'):
deprecation_msg('move_Abs method in plugins is deprecated, use move_abs', 3)
self.move_Abs(value) # type: ignore
else:
raise NotImplementedError
[docs]
def move_rel(self, value: Union[float, DataActuator]) -> None:
"""Move the actuator by ``value`` relative to its position (to be implemented in the plugin)
Call :meth:`move_done` when the move is finished if polling is not used (see :attr:`ispolling`).
"""
if hasattr(self, 'move_Rel'):
deprecation_msg('move_Rel method in plugins is deprecated, use move_rel', 3)
self.move_Rel(value) # type: ignore
else:
raise NotImplementedError
[docs]
def move_home(self, *args, **kwargs):
"""Let the actuator look for its reference, so that this position becomes the home (zero) position
To be implemented in the plugin. PyMoDAQ does not poll the position after this call: the plugin must
call :meth:`move_done` (that reads :meth:`get_actuator_value` and emits the ``move_done_signal``) when the homing
is finished, right after a blocking homing or from the callback of an asynchronous one. See
https://pymodaq.cnrs.fr/en/latest/developer_folder/instrument_plugins.html#moving-to-the-home-position
"""
if hasattr(self, 'move_Home'):
deprecation_msg('move_Home method in plugins is deprecated, use move_home', 3)
self.move_Home() # type: ignore
else:
raise NotImplementedError
[docs]
def emit_value(self, pos: DataActuator):
"""Convenience method to emit the current actuator value back to the UI"""
self.emit_status(ThreadCommand(ThreadStatusMove.GET_ACTUATOR_VALUE, pos))
[docs]
def set_default_values(self, red: Optional[float] = None, green: Optional[float] = None,
relative: Optional[float] = None):
"""Set from the plugin the values of the red, green and relative spinboxes of the UI
Those are the *Value Red*, *Value Green* and *Value Relative* entries of the main settings of the module, and
the values are expressed in the units of the actuator. Only the values that are not None are updated.
Parameters
----------
red: float
green: float
relative: float
"""
for name, value in (('default_value_red', red), ('default_value_green', green),
('default_value_relative', relative)):
if value is not None:
self.emit_status(ThreadCommand(ThreadStatus.UPDATE_MAIN_SETTINGS, [[name], value, 'value']))
[docs]
def commit_settings(self, param: Parameter):
"""
to subclass to transfer parameters to hardware
"""
[docs]
def move_done(self, position: Optional[DataActuator] = None): # the position argument is just there to match some signature of child classes
""" Emit a move done signal transmitting the actuator's value to the GUI
The position argument is just there to match some signature of child classes.
"""
if position is None:
if self.data_actuator_type == DataActuatorType.float:
position = DataActuator(self._title, data=self.get_actuator_value(),
units=self.axis_unit)
else:
position = cast(DataActuator, self.get_actuator_value())
if position.name != self._title: # make sure the emitted DataActuator has the name of the real implementation
#of the plugin
position.name = self._title
self.current_value = position # update the value in case move_done was called from a homing step
self.move_done_signal.emit(position)
self.move_is_done = True
[docs]
def poll_moving(self):
""" Poll the current moving. In case of timeout emit the raise timeout Thread command.
See Also
--------
DAQ_utils.ThreadCommand, move_done
"""
if not ('TCPServer' in self.__class__.__name__ or
'LECODirector' in self.__class__.__name__):
self.start_time = perf_counter()
if self.ispolling:
self.poll_timer.start()
else:
if not self.has_encoder:
self._current_value = self.target_value
else:
if self.data_actuator_type == DataActuatorType.float:
self._current_value = DataActuator(
data=self.get_actuator_value(),
units=self.axis_unit)
else:
self._current_value = self.get_actuator_value()
if (not Unit(self.axis_unit).is_compatible_with(
Unit(self._current_value.units)) and
self._current_value.units == ''):
# this happens if the units have not been specified in
# the plugin
self._current_value.force_units(self.axis_unit)
logger.debug(f'Current position: {self._current_value}')
self.move_done(self._current_value)
def _condition_to_reach_target(self, check_absolute_difference=True) -> bool:
"""Implement the condition for exiting the polling mechanism and specifying that the
target value has been reached
Returns
-------
bool: if True, PyMoDAQ considers the target value has been reached
See Also
--------
absolute_difference_condition_to_reach_target
user_condition_to_reach_target
"""
cond = True
if check_absolute_difference:
cond = self.absolute_difference_condition_to_reach_target()
return cond and self.user_condition_to_reach_target()
[docs]
def absolute_difference_condition_to_reach_target(self) -> bool:
""" Implement the condition for exiting the polling mechanism and specifying that the
target value has been reached
Returns
-------
bool: if True, PyMoDAQ considers the target value has been reached at epsilon
See Also
--------
user_condition_to_reach_target
"""
try:
epsilon_calculated = (
self._current_value - self._target_value).abs().value(self.axis_unit)
except DataUnitError as e:
epsilon_calculated = abs(self._current_value.value() - self._target_value.value())
logger.warning(f'Unit issue when calculating epsilon, units are not compatible between'
f'target and current values')
except OffsetUnitCalculusError as e:
if '°C' in self.axis_unit or 'celcius' in self.axis_unit.lower():
epsilon_calculated = (
self._current_value.to_base_units() -
self._target_value.to_base_units()).abs().value()
else:
raise e
return (epsilon_calculated < self.epsilon)
[docs]
def user_condition_to_reach_target(self) -> bool:
""" Implement a user defined condition for exiting the polling mechanism and specifying
that the target value has been reached (on top of the existing epsilon mechanism)
Should be reimplemented in plugins to implement other conditions
Returns
-------
bool: if True, PyMoDAQ considers the target value has been reached
"""
return True
[docs]
def check_target_reached(self) -> None:
"""Slot of the polling timer: read the actuator value and check if the target has been reached
If it has, stop the timer and call :meth:`move_done`. Otherwise emit the new value, and stop and
emit a timeout command if the ``timeout`` setting is exceeded.
"""
logger.debug(f"epsilon value is {self.epsilon}")
logger.debug(f"current_value value is {self._current_value}")
logger.debug(f"target_value value is {self._target_value}")
if not self._condition_to_reach_target():
logger.debug(f'Check move_is_done: {self.move_is_done}')
if self.move_is_done:
self.emit_status(ThreadCommand(ThreadStatus.UPDATE_STATUS, 'Move has been stopped'))
logger.info('Move has been stopped')
try:
self.current_value = self.get_actuator_value()
except Exception as e:
logger.exception(str(e))
self.poll_timer.stop()
self.emit_status(ThreadCommand(ThreadStatus.ERROR, f'check_target_reached: {e}'))
return
self.emit_value(self._current_value)
logger.debug(f'Current value: {self._current_value}')
if perf_counter() - self.start_time >= self.settings['timeout']:
self.poll_timer.stop()
self.emit_status(ThreadCommand(ThreadStatus.RAISE_TIMEOUT))
logger.info('Timeout activated')
else:
self.poll_timer.stop()
try:
self.current_value = self.get_actuator_value()
except Exception as e:
logger.exception(str(e))
self.emit_status(ThreadCommand(ThreadStatus.ERROR, f'check_target_reached: {e}'))
return
logger.debug(f'Current value: {self._current_value}')
self.move_done(self._current_value)
[docs]
def get_position_with_scaling(self, pos: DataActuator) -> DataActuator:
""" Get the current position from the hardware with scaling conversion.
"""
if self.settings['scaling', 'use_scaling']:
if self.data_actuator_type == DataActuatorType.float:
pos = (pos - self.settings['scaling', 'offset']) * self.settings['scaling', 'scaling']
else:
pos = (pos - Q_(self.settings['scaling', 'offset'],
self.axis_unit)) * self.settings['scaling', 'scaling']
return pos
[docs]
def set_position_with_scaling(self, pos: DataActuator) -> DataActuator:
""" Set the current position from the parameter and hardware with scaling conversion.
"""
if self.settings['scaling', 'use_scaling']:
if self.data_actuator_type == DataActuatorType.DataActuator:
pos = pos / self.settings['scaling', 'scaling'] + Q_(self.settings['scaling', 'offset'], self.axis_unit)
else:
pos = pos / self.settings['scaling', 'scaling'] + self.settings['scaling', 'offset']
return pos
[docs]
def set_position_relative_with_scaling(self, pos: DataActuator) -> DataActuator:
""" Set the scaled positions in case of relative moves
"""
if self.settings['scaling', 'use_scaling']:
pos = pos / self.settings['scaling', 'scaling']
return pos
[docs]
@Slot(edict)
def update_settings(self, settings_parameter_dict):
"""Apply settings-tree change and handle actuator-specific axis/epsilon side-effects."""
super().update_settings(settings_parameter_dict)
if settings_parameter_dict['change'] == 'value':
param = settings_parameter_dict['param']
if param.name() == 'axis':
# the value of the parameter is the axis name for a list of names but the value of the dict for named
# axes: the axis_name getter always returns the name, setting it updates units and epsilon
self.axis_name = self.axis_name
elif param.name() == 'epsilon':
self.epsilon = param.value()
[docs]
def ini_stage_init(self, old_controller=None, new_controller=None, slave_controller=None): # type: ignore[no-redef]
"""Deprecated — use ini_controller_init instead."""
import warnings
warnings.warn("'ini_stage_init' is deprecated, use 'ini_controller_init' instead.",
DeprecationWarning, stacklevel=2)
return self.ini_controller_init(old_controller, new_controller, slave_controller)
# abstract methods to be overwritten by the concrete implementations
[docs]
@abstractmethod
def ini_stage(self, controller: Optional[HardwareController] = None) -> tuple[str, bool]:
"""Actuator communication initialization
Parameters
----------
controller: (object)
custom object of a PyMoDAQ plugin (Slave case). None if only one actuator by controller (Master case)
Returns
-------
info: str
initialized: bool
False if initialization failed otherwise True
"""
pass
[docs]
@abstractmethod
def stop_motion(self, *args, **kwargs) -> None:
"""Stop the actuator and emit move_done signal."""
pass
if __name__ == '__main__':
test = DAQ_Move_base() # type: ignore[abstract]