from contextlib import contextmanager
from copy import deepcopy
from pathlib import Path
from queue import Queue
from threading import RLock, Thread
from time import perf_counter, sleep
from warnings import warn
import numpy as np
import pandas as pd
from zaber_motion import InvalidDataException, Library, Measurement, Units
from zaber_motion.ascii import SettingConstants
from zaber_motion.exceptions import BadDataException
from owl.instruments._connection_manager import FakeMotorConnectionManager, MotorConnectionManager
from owl.util.util import cache_dir
from ._utils import with_thread_lock
# Enable offline caching of device database
# Without this, the motor doesn't work offline
# https://gitlab.com/ZaberTech/zaber-motion-lib/-/issues/4
# Maybe we can find a way to package this in the conda package?
# https://www.zaber.com/software/docs/motion-library/ascii/howtos/device_db/
zaber_cache_dir = cache_dir / 'owl' / 'instruments' / 'zaber'
Library.enable_device_db_store(str(zaber_cache_dir))
"""
# On first connection
from pprint import pprint
from zaber_motion import Library
from zaber_motion.ascii import Connection
from zaber_motion.ascii import SettingConstants
from zaber_motion import Units, Library
from zaber_motion import Units, MovementFailedException
connection = Connection.open_serial_port("/dev/ttyUSB0")
connection.renumber_devices()
# To shorten the stages
from pprint import pprint
from zaber_motion import Library
from zaber_motion.ascii import Connection
from zaber_motion.ascii import SettingConstants
from zaber_motion import Units, Library
from zaber_motion import Units, MovementFailedException
connection = Connection.open_serial_port("/dev/ttyUSB0")
device_address = 1
device = connection.detect_devices()[device_address - 1]
controller_id = 1
stage = device.get_axis(controller_id)
original_maximum_position = stage.settings.get(SettingConstants.LIMIT_MAX, unit=Units.LENGTH_METRES)
print(original_maximum_position)
desired_maximum_position = 100E-3
stage.settings.set(SettingConstants.LIMIT_MAX, desired_maximum_position, unit=Units.LENGTH_METRES)
# The following script may help create an entry for a stage
# It should be used with caution, and the user should use it in conjunction
# With the terminal command
# mcam_list to retrieve the FTDI serial number as well as the stage serial number
# It will auto populate the stage maximum velocity and acceleration
# And make it ready to copy and paste into this dictionary with minimal
# formatting
from pprint import pprint
from zaber_motion import Library
from zaber_motion.ascii import Connection
from zaber_motion.ascii import SettingConstants
from zaber_motion import Units, Library
from zaber_motion import Units, MovementFailedException
connection = Connection.open_serial_port("/dev/ttyUSB0")
device_address = 1
device = connection.detect_devices()[device_address - 1]
controller_id = 1
stage = device.get_axis(controller_id)
default_speed = stage.settings.get(
SettingConstants.MAXSPEED,
unit=Units.VELOCITY_METRES_PER_SECOND)
default_acceleration = stage.settings.get(
SettingConstants.ACCEL,
unit=Units.ACCELERATION_METRES_PER_SECOND_SQUARED)
stage.home()
try:
stage.move_max()
except MovementFailedException:
pass
maximum_position = stage.get_position(Units.LENGTH_METRES)
# You can also get the maximum position by looking at the limit property
# but this may be off due to the fact that some stages have extra limit sensors
# We can also change the maximum travel
# maximum_position = stage.settings.get(SettingConstants.LIMIT_MAX, unit=Units.LENGTH_METRES)
stage_entry = {
'stage_serial_number': 'GET_FROM_MCAM_LIST_COMMAND',
'communication_adapter_serial_number': 'GET_FROM_MCAM_LIST_COMMAND',
'controller_type': 'axis',
'device_address': device_address,
'controller_id': controller_id,
'default_speed': default_speed,
'default_acceleration': default_acceleration,
'flip_axis': False,
'axis': 'CHOOSE_EITHER_X_Y_Z',
'travel_range': maximum_position,
}
pprint(stage_entry, indent=8, sort_dicts=False)
"""
"""
# How to swap the direction of the movement for the X-ASR stage:
1. Open the zaber launcher
2. Open the serial port.
3. Open the terminal
4. Click on the controller (X-ASR050B050B-SE03D12)
5. Elevate privileges
```
set system.access 2
```
6. Tell it that the hardware has been modified:
```
set device.hw.modified 1
```
7. Click on the axis of interest
8. Send it a command that we have swapped in the home and away limits.
```
set limit.swapinputs 1
```
This reverses the home and away sensor assignment.
The home sensor will now be considered away, and vice versa.
9. Swap the driver direction
```
set driver.dir 1
```
10.If the device has an encoder and you are seeing stalling, you may also need to send:
```
set encoder.dir 1
```
11. Now try to home the device. it should home well.
"""
"""Adding offsets to the zaber motors:
It may be desirable to have an offset to the zaber home positions.
For example, when configuring the x_sample_stage, we want to be able to
fix the home location, but to define "offsets" to it during the main operation.
There are 2 operating modes we are interested in:
1. Movement from our software
2. Movement from the knob.
Both should bring back the stage to the "imaging position" in this case.
The settings that need to be changed in the zaber launcher are;
limit.min -- 10.000 mm for example
limit.home.offset -- 10.000 for example
The limit.home.action must be set to something that will "move the stage to the offset"
after the home sensor is hit.
For example, the "2" action will do this.
"""
zaber_registry_path = (
Path(__file__).parent /
'device_registries' /
'zaber_registry.csv'
)
def _extra_settings_converter(value):
s = str(value)
settings_entries_string = s.split('), (')
settings_entries = []
for entry in settings_entries_string:
if len(entry) == 0:
continue
entry = entry.strip('(), ')
setting, value, unit = entry.split(',')
settings_entries.append((
str(setting).strip('\'" '),
float(value),
str(unit).strip('\'" '),
))
return tuple(settings_entries)
known_devices = pd.read_csv(
zaber_registry_path,
header=0,
dtype={
'stage_serial_number': str,
'ftdi_serial_number': str,
'communication_adapter_serial_number': str,
'controller_type': str,
'controller_id': int,
'axis': str,
'device_address': int,
'default_speed': float,
'default_acceleration': float,
'flip_axis': bool,
'travel_range': float,
'limit_min': float,
'limit_max': float,
'limit_home_offset': float,
'motion_accel_ramptime': float,
'limit_away_posupdate': float, # not actually a float, but useful since float has nan
'limit_approach_maxspeed': float, # not actually a float, but useful since float has nan
'driver_current_run': float,
'driver_current_hold': float,
'knob_mode': float,
'knob_distance': float,
},
converters={
'extra_settings': _extra_settings_converter,
},
)
# In 2025/08/03 we transitioned from ftdi_serial_number to communication_adapter_serial_number
# But we want to keep the registry backward compatible with old code
# that might still need to use ftdi_serial_number
# So in the registry, we kept both columns
# We should remove this in 2026/01/01
if 'communication_adapter_serial_number' in known_devices.columns:
known_devices = known_devices.drop(columns=['ftdi_serial_number'])
else:
print("""Warning, you have an old version of the registry download a new one
pip install -e . -vv --no-build-isolation
""")
known_devices = known_devices.rename(
columns={'ftdi_serial_number': 'communication_adapter_serial_number'}
)
# The yyyyyR / xxxxxR entries are fake lockstep stages used only for debugging
# and by the test suite. They are generated here rather than stored as the last
# rows of the CSV so that new real devices can always be appended to the
# registry without anyone having to insert rows above a fixed footer.
_debug_devices = pd.DataFrame([
{
'stage_serial_number': 'yyyyyR',
'communication_adapter_serial_number': 'y_ftdi',
'controller_type': 'lockstep',
'controller_id': 2,
'axis': 'y',
'device_address': 1,
'default_speed': 0.5,
'default_acceleration': 0.01073,
'flip_axis': False,
'use_park': False,
'fake_device': True,
'extra_settings': (),
},
{
'stage_serial_number': 'xxxxxR',
'communication_adapter_serial_number': 'x_ftdi',
'controller_type': 'lockstep',
'controller_id': 2,
'axis': 'y',
'device_address': 1,
'default_speed': 0.5,
'default_acceleration': 0.01073,
'flip_axis': False,
'use_park': False,
'fake_device': True,
'extra_settings': (),
},
])
# Skip any debug device already present so an older CSV that still ships them as
# rows does not produce duplicates.
_debug_devices = _debug_devices[
~_debug_devices.stage_serial_number.isin(known_devices.stage_serial_number)
]
if len(_debug_devices):
known_devices = pd.concat(
[known_devices, _debug_devices], ignore_index=True
)
fake_known_devices = known_devices.copy(deep=True)
fake_known_devices = fake_known_devices.loc[
[pd.notna(sn) and not sn.endswith('R') for sn in fake_known_devices.stage_serial_number]
]
for i in range(len(fake_known_devices)):
fake_known_devices.loc[i, 'stage_serial_number'] += 'R'
fake_known_devices.loc[i, 'fake_device'] = True
known_devices = pd.concat([known_devices, fake_known_devices], ignore_index=True)
def _do_threaded_movement(controller, position, queue):
try:
controller.wait_until_idle()
try:
controller.move_absolute(
position=position,
unit=Units.LENGTH_METRES,
wait_until_idle=False,
)
except BadDataException as e:
raise RuntimeError(
f"Failed to move to position {position:.6f} m. "
f"Error: {e}"
)
queue.put((True, None))
return
except Exception as e:
import traceback
traceback.print_exc()
queue.put((False, e))
return
[docs]
class X_LSM_E:
"""Zaber Miniature Motorized linear stage with encoder and controller.
Tested with the X-LSM-E series stages.
Parameters
----------
communication_adapter_serial_number: str
Serial number for the port on which to connect to for the motor.
port: str
USB port on which to connect to for the motor.
stage_serial_number: str
The serial number of the Zaber motor to connect to.
connection_manager: MotorConnectionManager
The MotorConnectionManager which manages the connection this motor should use.
The connection manager should currently be active and in use. This will
override a supplied serial_number or port.
ensure_homed: bool
If True, ensures that the stage is homed upon connection.
Motors that have not been homed do not report an accurate position
offset.
controller_type: str
One of "lockstep" or "axis". The type of controller to be used for this stage.
controller_id: int
The id of the axis or lockstep group being used for this stage.
device: zaber_motion.ascii.Device
An open device to be used with this stage.
Using a pre-opened device may speed up the opening process
device_address: int
Numerical order of the device on the connection.
If provided, this can speed up opening the device.
Examples
--------
>>> from owl.instruments import X_LSM_E
>>> z_stage = X_LSM_E()
>>> # Set the stage position to 3 mm
>>> z_stage.position = 3E-3
"""
_known_devices = known_devices
__slots__ = [
'is_open',
'_device',
'controller_type',
'controller_id',
'_flip_axis',
'travel_range',
'connection_manager',
'_axes',
'_controller',
'_firmware_version',
'_thread_lock',
'_fake_device',
'_fake_properties',
'_expected_stream_durations',
'_init_kwargs',
'_anti_backlash_thread',
'_anti_backlash_result_queue',
]
[docs]
@classmethod
def by_axis(cls, axis, *args, connection_managers=None, **kwargs):
"""Connect to an X_LSM_E stage by device axis.
Parameters
----------
axis: str
The axis direction of the stage. One of 'z', 'x', or 'y'.
connection_managers: List[MotorConnectionManager] or List[FakeMotorConnectionManager]
If the device may be on an open connection, the
``MotorConnectionManager`` object(s) for the possible connection(s)
must be provided.
"""
stage_serial_number = kwargs.get('stage_serial_number', '')
fake_device = kwargs.get(
'fake_device',
isinstance(stage_serial_number, str) and stage_serial_number.endswith('R')
)
if connection_managers is None:
connection_managers = []
if fake_device:
if 'stage_serial_number' not in kwargs:
kwargs['stage_serial_number'] = f"{axis}_dummy"
else:
device = X_LSM_E._check_if_stage_is_known(**kwargs)
if device is None:
raise ValueError(f"No stage found with axis = {axis}")
# use parameters from the database
kwargs['axis'] = axis
kwargs['communication_adapter_serial_number'] = list(
device.communication_adapter_serial_number
)[0]
kwargs['stage_serial_number'] = list(device.stage_serial_number)[0]
kwargs['controller_type'] = list(device.controller_type)[0]
kwargs['controller_id'] = list(device.controller_id)[0]
kwargs['device_address'] = list(device.device_address)[0]
kwargs['fake_device'] = True
else:
# raise an error if information is provided for a stage not in the database
kwargs['axis'] = axis
information = X_LSM_E._check_if_stage_is_known(**kwargs)
if information is None:
raise ValueError(f"No stage found with axis = {axis}")
kwargs.pop('axis')
# get ftdi serial numbers for the currently open connections
open_connection_numbers = [i.serial_number for i in connection_managers]
ids = cls._known_devices.axis == axis
# the devices from the database with the axis of interest
possible_devices = cls._known_devices[ids]
available_communication_adapter_serial_numbers = set(
MotorConnectionManager.list_all_serial_numbers()
)
available_ids = [i in available_communication_adapter_serial_numbers
for i in possible_devices.communication_adapter_serial_number]
# the devices from the database with the correct axis,
# associated with connected ftdi serial numbers
available_devices = possible_devices[available_ids]
# Iterate through each connected ftdi serial number to check for matching devices
for number in list(available_devices.communication_adapter_serial_number):
close_connection = True
device = None
current_ids = available_devices.communication_adapter_serial_number == number
# the devices on this serial port with the axis of interest
current_devices = available_devices[current_ids]
# check if this ftdi serial number is associated with an already open connection
if number in open_connection_numbers:
index = open_connection_numbers.index(number)
connection_manager = connection_managers[index]
close_connection = False
else:
connection_manager = MotorConnectionManager(serial_number=number)
connection = connection_manager._connection
possible_device_addresses = list(current_devices.device_address)
device = None
for address in possible_device_addresses:
try:
found_device = connection.get_device(int(address))
found_device.identify()
device_id = current_devices.device_address == int(address)
device = current_devices[device_id]
break
except Exception:
continue
if device is not None:
break
if close_connection:
connection.close()
else:
raise RuntimeError(f"Could not find any stages along {axis} axis")
# use parameters from the database
kwargs['communication_adapter_serial_number'] = list(
device.communication_adapter_serial_number
)[0]
kwargs['stage_serial_number'] = list(device.stage_serial_number)[0]
kwargs['controller_type'] = list(device.controller_type)[0]
kwargs['controller_id'] = list(device.controller_id)[0]
kwargs['connection_manager'] = connection_manager
kwargs['device'] = found_device
kwargs['device_address'] = list(device.device_address)[0]
if kwargs['device'].serial_number != int(kwargs['stage_serial_number']):
raise RuntimeError(
f"Unable to connect to stage axis '{axis}'. "
"Check the device_address for stage with "
f"serial number {kwargs['device'].serial_number}."
)
stage = cls(*args, **kwargs)
return stage
@staticmethod
def _check_if_stage_is_known(**kwargs):
possible_devices = X_LSM_E._known_devices
for parameter in ['axis', 'communication_adapter_serial_number', 'stage_serial_number',
'controller_type', 'controller_id']:
if parameter in kwargs:
ids = getattr(possible_devices, parameter) == kwargs[parameter]
possible_devices = possible_devices[ids]
if len(possible_devices) == 0:
return None
return possible_devices
def __enter__(self):
return self
def __exit__(self, type, value, traceback):
self.close()
def __init__(
self,
port=None, *,
ftdi_serial_number=None,
communication_adapter_serial_number=None,
ensure_homed=True,
stage_serial_number=None,
connection_manager=None,
controller_type="axis",
controller_id=1,
device=None,
device_address=None,
set_default_parameters=True,
flip_axis=None,
**kwargs,
):
# 2025/07
# This compatibility shim was added to ease the transition to the Maggeallan stages
# We can remove some time in 2026
if ftdi_serial_number is not None and communication_adapter_serial_number is not None:
raise ValueError(
"Cannot provide both ftdi_serial_number and communication_adapter_serial_number. "
"Please use communication_adapter_serial_number instead."
)
if ftdi_serial_number is not None:
warn("The ftdi_serial_number parameter is deprecated, please use "
"communication_adapter_serial_number instead.",
stacklevel=2)
communication_adapter_serial_number = ftdi_serial_number
for slot in self.__slots__:
setattr(self, slot, None)
self.is_open = False
self._thread_lock = RLock()
init_kwargs = deepcopy(kwargs)
init_kwargs['port'] = port
init_kwargs['communication_adapter_serial_number'] = communication_adapter_serial_number
init_kwargs['ensure_homed'] = ensure_homed
init_kwargs['stage_serial_number'] = stage_serial_number
init_kwargs['connection_manager'] = connection_manager
init_kwargs['controller_type'] = controller_type
init_kwargs['controller_id'] = controller_id
init_kwargs['device'] = device
init_kwargs['device_address'] = device_address
init_kwargs['set_default_parameters'] = set_default_parameters
init_kwargs['flip_axis'] = flip_axis
self._init_kwargs = init_kwargs
self.open(_stacklevel_increment=1)
[docs]
def open(self, *, _stacklevel_increment=0):
"""Open the device for communication.
This function is automatically called at the end of the object
creation. It is mostly useful for users that need to manually
close the device and reopen it later for communication.
"""
try:
self._open(_stacklevel_increment=_stacklevel_increment + 1)
except Exception:
self.close()
raise
@with_thread_lock
def _open(self, *, _stacklevel_increment):
connection_manager = self._init_kwargs.get('connection_manager')
port = self._init_kwargs.get('port')
communication_adapter_serial_number = self._init_kwargs.get(
'communication_adapter_serial_number')
stage_serial_number = self._init_kwargs.get('stage_serial_number')
controller_type = self._init_kwargs.get('controller_type')
controller_id = self._init_kwargs.get('controller_id')
device_address = self._init_kwargs.get('device_address')
set_default_parameters = self._init_kwargs.get('set_default_parameters')
ensure_homed = self._init_kwargs.get('ensure_homed')
flip_axis = self._init_kwargs.get('flip_axis')
if isinstance(stage_serial_number, str):
self._fake_device = bool(
self._init_kwargs.get('fake_device', stage_serial_number.endswith('R'))
)
else:
self._fake_device = self._init_kwargs.get('fake_device', False)
self.is_open = False
self._device = self._init_kwargs.get('device')
self.controller_type = self._init_kwargs.get('controller_type')
self.controller_id = self._init_kwargs.get('controller_id')
# Allow the user to override the flip_axis setting
if flip_axis is not None:
self._flip_axis = flip_axis
else:
self._flip_axis = False
self.travel_range = None
if not self._fake_device:
if connection_manager is None:
connection_manager = MotorConnectionManager(
port=port, serial_number=communication_adapter_serial_number)
if not connection_manager.is_open:
raise RuntimeError("Provided connection manager should already be open.")
self.connection_manager = connection_manager
self._device = self.connection_manager.add_motor(stage_serial_number,
controller_type, controller_id,
self._device, device_address)
firmware_version = self._device.firmware_version
self._firmware_version = (f"{firmware_version.major}."
f"{firmware_version.minor}."
f"{firmware_version.build}")
self.is_open = True
if controller_type == "axis":
self._controller = self._device.get_axis(controller_id)
# Try to read the axis type, to ensure that the controller_id is valid
self._controller.axis_type
self._axes = [self._controller]
else:
self._controller = self._device.get_lockstep(controller_id)
self._axes = []
for index in self._controller.get_axis_numbers():
self._axes.append(self._device.get_axis(index))
if set_default_parameters:
information = X_LSM_E._check_if_stage_is_known(
communication_adapter_serial_number=self.communication_adapter_serial_number,
stage_serial_number=self.stage_serial_number,
controller_type=self.controller_type,
controller_id=self.controller_id,
)
if information is not None:
information = information.iloc[0]
self.maximum_speed = information.default_speed
self.acceleration = information.default_acceleration
self.travel_range = information.travel_range
if flip_axis is None:
self.flip_axis = information.flip_axis
# Other things to setup
# home speed -- limit.approach.maxspeed 0.010 m/s
# limit.max -- maximum physical limit
axis = self._axes[0]
with self._elevate_access():
if not np.isnan(information.knob_mode):
axis.settings.set(
'knob.mode',
int(information.knob_mode),
Units.NATIVE
)
if not np.isnan(information.knob_distance):
axis.settings.set(
'knob.distance',
information.knob_distance,
Units.LENGTH_METRES
)
if not np.isnan(information.limit_approach_maxspeed):
axis.settings.set(
'limit.approach.maxspeed',
information.limit_approach_maxspeed,
Units.VELOCITY_METRES_PER_SECOND
)
if not np.isnan(information.driver_current_run):
axis.settings.set(
'driver.current.run',
information.driver_current_run,
Units.AC_ELECTRIC_CURRENT_AMPERES_PEAK
)
if not np.isnan(information.driver_current_hold):
axis.settings.set(
'driver.current.hold',
information.driver_current_hold,
Units.DC_ELECTRIC_CURRENT_AMPERES
)
if not np.isnan(information.limit_min):
axis.settings.set(
'limit.min',
information.limit_min,
Units.LENGTH_METRES
)
if not np.isnan(information.limit_max):
axis.settings.set(
'limit.max',
information.limit_max,
Units.LENGTH_METRES
)
if not np.isnan(information.limit_home_offset):
axis.settings.set(
'limit.home.offset',
information.limit_home_offset,
Units.LENGTH_METRES
)
if not np.isnan(information.motion_accel_ramptime):
axis.settings.set(
'motion.accel.ramptime',
int(information.motion_accel_ramptime),
Units.NATIVE
)
if not np.isnan(information.limit_away_posupdate):
axis.settings.set(
'limit.away.posupdate',
int(information.limit_away_posupdate),
Units.NATIVE
)
extra_settings = information.extra_settings
if extra_settings is not None:
for setting, value, unit in extra_settings:
self._axes[0].settings.set(
setting=setting,
value=value,
unit=getattr(Units, unit)
)
else: # fake_device
# We know they are unique
entry = self._known_devices.loc[
self._known_devices['stage_serial_number'] == stage_serial_number
]
if len(entry) == 0:
maximum_speed = 0.03
maximum_position = 0.0508
acceleration = 0.06
if not communication_adapter_serial_number:
communication_adapter_serial_number = "fake_device"
if not stage_serial_number:
stage_serial_number = "dummy"
else:
entry = entry.iloc[0]
maximum_speed = entry.default_speed
maximum_position = entry.travel_range
acceleration = entry.default_acceleration
communication_adapter_serial_number = entry.communication_adapter_serial_number
# valid serial, so just use it
self._fake_properties = {
"position": 0.001,
"homed": False,
"led_enable": True,
"maximum_speed": maximum_speed,
"approach_maximum_speed": 0.001,
"maximum_position": maximum_position,
"minimum_position": 0,
"acceleration": acceleration,
"communication_adapter_serial_number": communication_adapter_serial_number,
"stage_serial_number": stage_serial_number,
"motion_target_position": 0.,
"motion_starting_position": 0.,
"motion_start_time": 0.,
"moving": False,
}
if set_default_parameters:
information = X_LSM_E._check_if_stage_is_known(
communication_adapter_serial_number=self.communication_adapter_serial_number,
stage_serial_number=self.stage_serial_number,
controller_type=self.controller_type,
controller_id=self.controller_id,
)
if information is not None:
self.maximum_speed = list(information.default_speed)[0]
self.acceleration = list(information.default_acceleration)[0]
self.travel_range = list(information.travel_range)[0]
if flip_axis is None:
self.flip_axis = list(information.flip_axis)[0]
else:
self.travel_range = 0.0508
self._firmware_version = "0.0.1"
if connection_manager is None:
connection_manager = FakeMotorConnectionManager(
serial_number=self._fake_properties['communication_adapter_serial_number'])
if not connection_manager.is_open:
raise RuntimeError("Provided connection manager should already be open.")
self.connection_manager = connection_manager
self.connection_manager.add_motor(self._fake_properties['stage_serial_number'],
controller_type,
controller_id)
self.is_open = True
self._expected_stream_durations = {}
if self.led_enable:
self.led_enable = False
self._anti_backlash_thread = None
# TODO: empty queue when the user cancels the threaded movement
self._anti_backlash_result_queue = Queue()
if ensure_homed:
self.ensure_homed()
@property
@with_thread_lock
def homed(self):
"""A property describing whether or not the stage has been homed.
Stages require homing every time they are power cycled.
"""
if self._fake_device:
update_fake_position(self._fake_properties)
return self._fake_properties['homed']
homed = []
for axis in self._axes:
is_homed = axis.settings.get(SettingConstants.LIMIT_HOME_TRIGGERED)
is_homed = is_homed > 0.5
homed.append(is_homed)
is_homed = any(homed)
return is_homed
[docs]
@with_thread_lock
def ensure_homed(self, wait_until_idle=True):
"""Ensures that the stage has been homed.
Returns immediately if the stage is already homed.
Parameters
----------
wait_until_idle: bool
If True, the call will return immediately.
If False, the caller must monitor the busy signal themselves to
ensure the stage has stopped moving.
"""
if self.homed:
return
self.home(wait_until_idle=wait_until_idle)
[docs]
@with_thread_lock
def home(self, wait_until_idle=True):
"""Move stage to home position."""
# Clare, Mark, 2021/06/06
# Hack, when using a lockstep, we've found that we had a very
# hard time using the home command. It would cause an error like
#
# MovementFailedException:
# Movement may have failed because a limit error was observed (FE,WR)
#
# Setting the position seems to help home in this case
if self._fake_device:
self.set_position(0.0, wait=wait_until_idle, ignore_flip=True)
return
if self.controller_type == 'lockstep':
self.set_position(0, wait=wait_until_idle, ignore_flip=True)
else:
self._controller.home(wait_until_idle=wait_until_idle)
@with_thread_lock
def stop(self):
if self._fake_device:
update_fake_position(self._fake_properties)
self._fake_properties["moving"] = False
return
if self._anti_backlash_thread:
if self._anti_backlash_thread.is_alive():
self._controller.stop(wait_until_idle=False)
success, maybe_exception = self._anti_backlash_result_queue.get()
if not success:
raise maybe_exception
# It might need a second stop after stopping the primary motion of
# the anti backlash, so stop it again
self._controller.stop()
def __del__(self):
# During the python interpreter shutdown, sometimes it isn't possible to call
# certain methods cannot be called.
if self.is_open:
self.close()
[docs]
@with_thread_lock
def close(self):
"""Close the device for communication."""
if self.is_open:
try:
self.stop()
except InvalidDataException:
pass
self._anti_backlash_result_queue = None
# Cache the stage serial_number since we need to get it from the
# device
stage_serial_number = self.stage_serial_number
# 2022/04/05 Mark:
# Set references to low level properties to None
# early so that they are released according to python???
self._axes = None
self._controller = None
self._device = None
self.connection_manager.remove_motor(stage_serial_number,
controller_type=self.controller_type,
controller_id=self.controller_id)
# Set the rest of the slots to be None to validate that
self._axes = None
self._controller = None
self.connection_manager = None
# If the stage had an error during opening, for example
# when connected to the FTDI but the stages are powered down.
# in this case, we need to close the FTDI comm
if self.connection_manager is not None:
# I'm not too sure why it doesn't close itself during its own
# __del__ operation. But maybe it is because debugging code
# holds a reference to it.
# It just seems to make pytest hang without this.
self.connection_manager.close_if_empty()
self.controller_type = None
self.controller_id = None
self._flip_axis = None
self.travel_range = None
self._firmware_version = None
self._fake_device = None
self._fake_properties = None
self._expected_stream_durations = None
self.is_open = False
[docs]
@with_thread_lock
def wait_until_idle(self, *, sleep_time=0.01):
"""Wait until the device is idle.
Parameters
----------
sleep_time: float
Amount of time to wait between polling the device.
"""
if self._fake_device:
while self.busy:
sleep(sleep_time)
return
if self._anti_backlash_thread:
self._anti_backlash_thread.join()
success, maybe_exception = self._anti_backlash_result_queue.get()
if not success:
raise maybe_exception
self._anti_backlash_thread = None
self._controller.wait_until_idle()
@property
@with_thread_lock
def busy(self):
"""A property describing if the device is busy processing a command.
A device is typically busy when moving to a specified position or when
homing.
See also
--------
ensure_homed, position
"""
if self._fake_device:
update_fake_position(self._fake_properties)
return self._fake_properties['moving']
else:
return self._controller.is_busy()
# move to a position, but have the function return immediately
[docs]
@with_thread_lock
def set_position(self, value, *, wait=True, ignore_flip=False, anti_backlash=0.,
position_tolerance=1E-6):
"""Set the device to a specified position.
Parameters
----------
value: float
Position to provide to the stage in meters.
wait: bool
If True, the call will wait until the stage is idle before
returning.
anti_backlash: float
The distance and direction in which to travel to avoid backlash.
If the direction of the travel is already the same as that the stage
is travelling is, there is be no backlash movement.
If the direction of travel is opposite to that of the anti-backlash
a movement the size of the anti-backlash amount will be attempted.
This parameter will be ignored near the edges of the travel.
position_tolerance: float
The tolerance in meters for the position to be considered reached.
This is used when determining if anti_backlash is needed.
See also
--------
set_position_mm, position, position_mm
"""
if not ignore_flip and self.flip_axis:
value = self.travel_range - value
anti_backlash = - anti_backlash
if self._fake_device:
# Freeze the position here
maximum_position = self._fake_properties['maximum_position']
if value > maximum_position:
raise ValueError(
f"Invalid position value ({value:.6f} m. "
f"Maximum position is {maximum_position:.6f} m "
f"for stage {self.stage_serial_number}.")
if value < 0:
raise ValueError(
f"Invalid position value ({value:.6f} m. "
"Minimum position is 0 m "
f"for stage {self.stage_serial_number}.")
self.stop()
self._fake_properties["motion_target_position"] = value
self._fake_properties["motion_starting_position"] = self._fake_properties['position']
self._fake_properties["motion_start_time"] = perf_counter()
self._fake_properties["moving"] = True
self._fake_properties["anti_backlash"] = anti_backlash
if wait:
self.wait_until_idle()
return
if anti_backlash:
# Reading the position issues a command through serial, so we avoid if we
# don't need to do it
current_position = self._get_true_position()
if value > (current_position + position_tolerance) and anti_backlash < 0.:
position = max(
min(value - anti_backlash, self._get_true_maximum_position()),
self._get_true_minimum_position()
)
elif value < (current_position - position_tolerance) and anti_backlash > 0.:
position = max(
min(value - anti_backlash, self._get_true_maximum_position()),
self._get_true_minimum_position()
)
else:
# No need for anti backlash
position = value
anti_backlash = 0.
else:
position = value
if self._anti_backlash_thread:
# Zaber code is really slow, so avoid calling stop if the thread
# is already joined
if self._anti_backlash_thread.is_alive():
self._controller.stop(wait_until_idle=False)
success, maybe_exception = self._anti_backlash_result_queue.get()
if not success:
raise maybe_exception
self._anti_backlash_thread = None
try:
self._controller.move_absolute(
position=position,
unit=Units.LENGTH_METRES,
wait_until_idle=wait,
)
except BadDataException as e:
raise RuntimeError(
f"Failed to move to position {position:.6f} m. "
f"Error: {e}"
)
if anti_backlash:
self._anti_backlash_thread = Thread(
target=_do_threaded_movement,
kwargs=dict(
controller=self._controller,
position=value,
queue=self._anti_backlash_result_queue,
)
)
self._anti_backlash_thread.start()
if wait:
self._anti_backlash_thread.join()
success, maybe_exception = self._anti_backlash_result_queue.get()
if not success:
raise maybe_exception
self._anti_backlash_thread = None
[docs]
@with_thread_lock
def set_position_mm(self, value, *args, anti_backlash=0., **kwargs):
"""Sets the position of the stage in millimeters.
See also
--------
set_position, position, position_mm
"""
self.set_position(value / 1000., *args, anti_backlash / 1000., **kwargs)
return
@property
@with_thread_lock
def acceleration(self):
"""Stage accelreation in meters per second squared."""
if self._fake_device:
return self._fake_properties['acceleration']
axis = self._axes[0]
return axis.settings.get(SettingConstants.ACCEL,
unit=Units.ACCELERATION_METRES_PER_SECOND_SQUARED)
@acceleration.setter
@with_thread_lock
def acceleration(self, value):
if self._fake_device:
self._fake_properties['acceleration'] = value
return
for axis in self._axes:
axis.settings.set(SettingConstants.ACCEL,
unit=Units.ACCELERATION_METRES_PER_SECOND_SQUARED,
value=value)
@property
def motor_voltage(self):
if self._fake_device:
return 24.
return self._device.settings.get('system.voltage')
@property
def current_hold(self):
if self._fake_device:
return 0.1 # A
# Likely doesn't mean much for tandem axes
return self._axes[0].settings.get(
'driver.current.hold',
Units.DC_ELECTRIC_CURRENT_AMPERES,
)
def _set_current_hold(self, value):
if self._fake_device:
return
axis = self._axes[0]
with self._elevate_access():
axis.settings.set(
'driver.current.hold',
value,
Units.DC_ELECTRIC_CURRENT_AMPERES
)
@property
def current_run(self):
if self._fake_device:
return 0.85 # A
# Likely doesn't mean much for tandem axes
return self._axes[0].settings.get(
'driver.current.run',
Units.AC_ELECTRIC_CURRENT_AMPERES_PEAK,
)
def _set_current_run(self, value):
if self._fake_device:
return
axis = self._axes[0]
with self._elevate_access():
axis.settings.set(
'driver.current.run',
value,
Units.AC_ELECTRIC_CURRENT_AMPERES_PEAK
)
@contextmanager
def _elevate_access(self):
if self._fake_device:
yield
return
axis = self._axes[0]
device = axis.device
old_access = device.settings.get('system.access')
device.settings.set('system.access', 2)
try:
yield
finally:
device.settings.set('system.access', old_access)
def _get_true_position(self):
# You cannot directly get the position of a lockstep,
# since theoretically the axes could be at different positions.
# However, we should be keeping all the axes of a lockstep in sync,
# so returning the position of one axis is adequate
return self._axes[0].get_position(Units.LENGTH_METRES)
@property
@with_thread_lock
def position(self):
"""Position of the stage in meters."""
if self._fake_device:
update_fake_position(self._fake_properties)
position = self._fake_properties['position']
else:
position = self._get_true_position()
if self.flip_axis:
position = self.travel_range - position
return position
@position.setter
@with_thread_lock
def position(self, value):
self.set_position(value, wait=True)
@property
@with_thread_lock
def position_mm(self):
"""Current position of the stage in millimeters."""
return self.position * 1000
@position_mm.setter
@with_thread_lock
def position_mm(self, value):
self.position = value / 1000
@property
@with_thread_lock
def maximum_position(self):
if not self.flip_axis:
return self._get_true_maximum_position()
else:
return (self.travel_range - self._get_true_minimum_position())
@with_thread_lock
def _get_true_maximum_position(self):
if self._fake_device:
return self._fake_properties['maximum_position']
# assuming both axes in a lockstep have the same limit
axis = self._axes[0]
return axis.settings.get(
SettingConstants.LIMIT_MAX, unit=Units.LENGTH_METRES)
@property
@with_thread_lock
def minimum_position(self):
if not self.flip_axis:
return self._get_true_minimum_position()
else:
return (self.travel_range - self._get_true_maximum_position())
@with_thread_lock
def _get_true_minimum_position(self):
if self._fake_device:
return self._fake_properties['minimum_position']
axis = self._axes[0]
return axis.settings.get(
SettingConstants.LIMIT_MIN, unit=Units.LENGTH_METRES)
@property
@with_thread_lock
def maximum_speed(self):
"""Maximum speed setpoint for the stage in meters per second."""
if self._fake_device:
return self._fake_properties['maximum_speed']
axis = self._axes[0]
return axis.settings.get(
SettingConstants.MAXSPEED, unit=Units.VELOCITY_METRES_PER_SECOND)
@maximum_speed.setter
@with_thread_lock
def maximum_speed(self, value):
self.set_maximum_speed(value)
@with_thread_lock
def set_maximum_speed(self, value):
if self._fake_device:
if value <= 0:
raise ValueError(f"Invalid speed {value} m/s")
self._fake_properties['maximum_speed'] = value
return
for axis in self._axes:
axis.settings.set(
SettingConstants.MAXSPEED, value=value, unit=Units.VELOCITY_METRES_PER_SECOND)
@property
@with_thread_lock
def approach_maximum_speed(self):
if self._fake_device:
return self._fake_properties['approach_maximum_speed']
axis = self._axes[0]
return axis.settings.get(
SettingConstants.LIMIT_APPROACH_MAXSPEED, unit=Units.VELOCITY_METRES_PER_SECOND)
@approach_maximum_speed.setter
@with_thread_lock
def approach_maximum_speed(self, value):
if self._fake_device:
self._fake_properties['approach_maximum_speed'] = value
return
for axis in self._axes:
axis.settings.set(
SettingConstants.LIMIT_APPROACH_MAXSPEED,
unit=Units.VELOCITY_METRES_PER_SECOND,
value=value)
@property
@with_thread_lock
def stage_serial_number(self):
"""The serial number of the stage as a string."""
if self._fake_device:
return self._fake_properties['stage_serial_number']
# Default serial number is an int, but it is nice to consider
# string-like serial numbers since they often contains numbers and
# symbols
return str(self._device.serial_number)
@property
@with_thread_lock
def communication_adapter_serial_number(self):
"""The serial number of the FTDI USB->Serial adapter for the stage."""
if self._fake_device:
return self._fake_properties['communication_adapter_serial_number']
return self.connection_manager.serial_number
@property
def ftdi_serial_number(self):
warn("The ftdi_serial_number property is deprecated, please use "
"communication_adapter_serial_number.",
stacklevel=2)
return self.communication_adapter_serial_number
@property
@with_thread_lock
def firmware_version(self):
"""Firmware version of the stage controller."""
return self._firmware_version
@property
@with_thread_lock
def led_enable(self):
if self._fake_device:
return self._fake_properties['led_enable']
return bool(self._device.settings.get(SettingConstants.SYSTEM_LED_ENABLE))
@led_enable.setter
@with_thread_lock
def led_enable(self, value):
if self._fake_device:
self._fake_properties['led_enable'] = value
return
self._device.settings.set(SettingConstants.SYSTEM_LED_ENABLE, value)
@property
def flip_axis(self):
return self._flip_axis
# 2021/07/29 Clare
# This requires that a stage have a travel range recorded in the database,
# since the "maximum_position" can be changed in software, and is not necessarily reliable.
# This could be changed in the future if we have reason to believe that the maximum position
# is not being modified.
@flip_axis.setter
def flip_axis(self, value):
if self.travel_range is None:
raise RuntimeError("'flip_axis' property cannot be used for stages"
"without a known travel range.")
self._flip_axis = value
[docs]
@with_thread_lock
def move_sin(self, *, amplitude, frequency, duration=1, count=None):
"""Move the stage in a sinusoidal fashion.
Examples
--------
>>> from owl.instruments import X_LSM_E
>>> stage = X_LSM_E()
>>> stage.move_sin(amplitude=1E-6, frequency=300)
"""
period = 1 / frequency
if count is None:
count = round(duration / period)
period_ms = period * 1E3
if self._fake_device:
return
# https://www.zaber.com/software/docs/motion-library/ascii/references/python/#convert95to95native95units
axis = self._axes[0]
amplitude_native_units = axis.settings.convert_to_native_units(
"pos", amplitude, Units.LENGTH_METRES)
axis.generic_command(f"move sin {amplitude_native_units:.0f} {period_ms:.2f} {count}")
def _stream_validate_is_capable(self, stream_index, buffer_index):
# The X_LSM_E was measured to have 4 streams
# and 100 buffers
num_streams = self.stream_num_streams
if num_streams == 0:
raise RuntimeError(
"Your system does not support the stream feature required for this "
"kind of control. Please contact Ramona Optics at help@ramonaoptics.com "
"with a screenshot of your screen including this error message to learn more."
)
# handle negative indexing with the modulor operator
if stream_index < 0:
stream_index = int(stream_index % num_streams)
else:
stream_index = int(stream_index)
if stream_index >= num_streams:
raise RuntimeError(
f"You requested to use stream index {stream_index}, however this system "
f"only supports {num_streams} streams. "
)
num_buffers = self.stream_num_buffers
if num_buffers < 1:
raise RuntimeError(
"Your system does not support the stream feature required for this "
"kind of control. Please contact Ramona Optics at help@ramonaoptics.com "
"to learn more."
)
if buffer_index < 0:
buffer_index = int(buffer_index % num_buffers)
else:
buffer_index = int(buffer_index)
if stream_index >= num_streams:
raise RuntimeError(
f"You requested to use stream index {stream_index}, however this system "
f"only supports {num_streams} streams. "
)
# They start their stream index at 1, but I (Mark) hates 1 based indexing
# Thus we start at 0
return stream_index + 1, buffer_index + 1
[docs]
def stream_call(
self,
*,
stream_index=0,
buffer_index=0,
):
"""Initiate a streamed motion.
See also
--------
stream_prepare_smoothed_vibration, stream_num_buffers, stream_num_streams
"""
duration = self._expected_stream_durations.get(buffer_index, None)
if duration is None:
raise RuntimeError(
f"Stream buffer index {buffer_index} was not previous setup."
)
if self._fake_device:
sleep(duration)
# Not implemented, lets just say we moved around and not...
return
stream_index, buffer_index = self._stream_validate_is_capable(
stream_index, buffer_index
)
stream = self._device.streams.get_stream(stream_index)
if not stream.check_disabled():
stream.disable()
stream_buffer = self._device.streams.get_buffer(buffer_index)
# TODO: this doesn't work well for tandem devices
stream.setup_live(self._axes[0].axis_number)
stream.call(stream_buffer)
stream.wait_until_idle()
stream.disable()
[docs]
def stream_prepare_smoothed_vibration(
self,
*,
amplitude,
frequency,
duration=1,
stream_index=0,
buffer_index=0,
tqdm=None,
):
"""Prepare a precisely control vibration motion on the stage.
The motion uses a Tukey Window to create a smooth transition
from the resting position for the maximum amplitude.
The first and last 10 oscillations are appodized using a raised cosine
to smoothly increase the amplitude of the motion from 0 to the desired
amplitude.
Examples
--------
>>> from owl.instruments import MCAM, X_LSM_E
>>> from tqdm import tqdm
>>> stage = X_LSM_E.by_axis('z')
>>> # Streams can be customized but each should be unique
>>> # to a given buffer_index
>>> stage.stream_prepare_smoothed_vibration(
... amplitude=1E-6,
... frequency=300,
... duration=1,
... buffer_index=1,
... tqdm=tqdm,
... )
>>> # The creation of streams can be a lengthy process
>>> # A tqdm constructor can be used to show an indication of progress
>>> stage.stream_prepare_smoothed_vibration(
... amplitude=1E-6,
... frequency=250,
... duration=2,
... buffer_index=0,
... tqdm=tqdm,
... )
>>> stage.stream_prepare_smoothed_vibration(
... amplitude=1E-6,
... frequency=350,
... duration=0.5,
... buffer_index=2,
... tqdm=tqdm,
... )
>>> # Stream can be played back in any order
>>> print("250 Hz")
>>> stage.stream_call(buffer_index=0)
>>> print("350 Hz")
>>> stage.stream_call(buffer_index=2)
>>> print("300 Hz")
>>> stage.stream_call(buffer_index=1)
>>> stage.close()
Parameters
----------
amplitude: float
In meters, the amplitude of the oscillations at maximum
intensity. The peak to peak amplitude is twice this value. Typical
values range between 0.25E-6 and 1E-6.
frequency: float
In Hertz, the frequency of oscillation of the movement.
duration: float
In seconds, the duration of the oscillation from start to finish
(including the ramp up time from the window function).
stream_index: int
The stream index to use for this movement. Typically this parameter
is left at 0.
buffer_index: int
The index where to store the motion in the device. Typical values
range between 0 and 99 inclusively but the legal value can range
from device to device.
"""
if duration <= 0:
raise ValueError(
f"duration must be a positive quantity greater than 0. Got {duration}."
)
if tqdm is None:
def tqdm(x, *args, **kwargs):
return x
from scipy.signal import windows
# Zaber counts indices at 1
N_cycles = int(frequency * duration)
N_cycles_window = 21
N_cycles_window = min(N_cycles_window, N_cycles)
if N_cycles_window % 2 == 0:
N_cycles_window += 1
# 2 cycles, one on either end will be identically 0 always
N_full_cycles = N_cycles - (N_cycles_window - 2)
# The first point will be identically 0, skip it,
half_window = windows.hann(N_cycles_window)[1:N_cycles_window // 2]
window = np.concatenate(
[half_window, np.ones(N_full_cycles, dtype=half_window.dtype), half_window[::-1]]
)
amplitude_peak_to_peak = amplitude * 2
# https://www.zaber.com/articles/cycling-applications-creating-precision-frequency-control-and-vibration
acceleration = amplitude_peak_to_peak * (frequency * 4) ** 2
self._expected_stream_durations[buffer_index] = duration
if self._fake_device:
# Not implemented, lets just say we prepared and not
for w in tqdm(window, desc="Preparing motion"):
# There seems to be the ability to write about 30 motions per second
# thus simulate the amount of time it takes for this
# just so that we can design higher level interfaces
sleep(0.03)
return
stream_index, buffer_index = self._stream_validate_is_capable(
stream_index, buffer_index,
)
axis = self._axes[0]
amplitude_native_units = axis.settings.convert_to_native_units(
"pos", amplitude_peak_to_peak,
Units.LENGTH_METRES
)
acceleration_native_units = axis.settings.convert_to_native_units(
"accel", acceleration,
Units.ACCELERATION_METRES_PER_SECOND_SQUARED
)
stream = self._device.streams.get_stream(stream_index)
stream.disable()
stream_buffer = self._device.streams.get_buffer(buffer_index)
stream_buffer.erase()
# TODO: this doesn't work well for tandem devices
stream.setup_store(stream_buffer, self._axes[0].axis_number)
previous_acceleration = None
for w in tqdm(window, desc="Preparing motion"):
# We want to take into account rounding here therefore,
# We get the integer values of the amplitude, and scale the
# acceleration by these values
this_amplitude = int(w * amplitude_native_units)
# We anticipate, that for very small motion, more than 1 or two
if this_amplitude == 0:
continue
this_acceleration = int(
acceleration_native_units * this_amplitude / amplitude_native_units
)
# Avoid writing acceleration all the time, this saves quite a bit
if previous_acceleration != this_acceleration:
stream.set_max_tangential_acceleration(this_acceleration)
previous_acceleration = this_acceleration
# None = native units
stream.line_relative(Measurement( this_amplitude, None)) # noqa
stream.line_relative(Measurement(-this_amplitude, None))
stream.disable()
@property
def stream_num_buffers(self):
"""The number of allowable buffers used to define stream sequences."""
# The settings.get functions seem to return floats
if self._fake_device:
return 100
return int(self._device.settings.get('stream.numbufs'))
@property
def stream_num_streams(self):
"""The number of allowable streams."""
# The settings.get functions seem to return floats
if self._fake_device:
return 4
return int(self._device.settings.get('stream.numstreams'))
def update_fake_position(fake_properties):
if fake_properties['moving']:
current_time = perf_counter()
start_time = fake_properties['motion_start_time']
elapsed_time = current_time - start_time
displacement_for_motion = (
fake_properties['motion_target_position'] -
fake_properties['motion_starting_position']
)
distance_for_motion = abs(displacement_for_motion)
# aint nobody got time to wait for fake devices to move around
# move them 100 times faster...
maximum_speed = fake_properties['maximum_speed'] * 100
time_required_for_motion = distance_for_motion / maximum_speed
if elapsed_time >= time_required_for_motion:
fake_properties['moving'] = False
fake_properties['position'] = fake_properties['motion_target_position']
if fake_properties['position'] == 0.:
fake_properties['homed'] = True
else:
fake_properties['position'] = (
fake_properties['motion_starting_position'] +
displacement_for_motion * elapsed_time / time_required_for_motion
)
# We decided that with clamping, 500 micron antibacklash is sufficient as a default
def find_anti_backlash(positions, anti_backlash=500E-6):
if len(positions) < 2:
return 0
anti_backlash = abs(anti_backlash)
if positions[-1] < positions[0]:
return -anti_backlash
else:
return anti_backlash