diff --git a/aaa_155749_20260209.json b/aaa_155749_20260209.json new file mode 100644 index 0000000..d0359f5 --- /dev/null +++ b/aaa_155749_20260209.json @@ -0,0 +1,144 @@ +{ + "scan_info": { + "friendly_name": "stest", + "waveform_prefix": "aaa", + "data_directory": "/opt/scanengine-3", + "timestamp": "2026-02-09T15:57:49.753094", + "scan_finished": true, + "completion_timestamp": "2026-02-09T15:57:58.301940" + }, + "scan_parameters": { + "number_of_angles": 4, + "row_spacing_mm": 0.5, + "scan_type": "standalone" + }, + "scan_area": { + "x_start_mm": 48.5, + "x_delta_mm": 15.83, + "y_start_mm": 42.17, + "y_delta_mm": -15.67, + "sample_size": "1.25\"" + }, + "scan_boxes": [ + { + "angle_index": 0, + "angle_degrees": 0.0, + "start": [ + 48.5, + 26.5 + ], + "end": [ + 64.33, + 42.17 + ], + "original_corners": [ + [ + 48.5, + 42.17 + ], + [ + 64.33, + 42.17 + ], + [ + 64.33, + 26.5 + ], + [ + 48.5, + 26.5 + ] + ] + }, + { + "angle_index": 1, + "angle_degrees": 45.0, + "start": [ + 45.3339, + 24.3934 + ], + "end": [ + 67.6077, + 46.6673 + ], + "original_corners": [ + [ + 45.3339, + 35.4738 + ], + [ + 56.5274, + 46.6673 + ], + [ + 67.6077, + 35.5869 + ], + [ + 56.4142, + 24.3934 + ] + ] + }, + { + "angle_index": 2, + "angle_degrees": 90.0, + "start": [ + 47.83, + 28.5 + ], + "end": [ + 63.5, + 44.33 + ], + "original_corners": [ + [ + 47.83, + 28.5 + ], + [ + 47.83, + 44.33 + ], + [ + 63.5, + 44.33 + ], + [ + 63.5, + 28.5 + ] + ] + }, + { + "angle_index": 3, + "angle_degrees": 135.0, + "start": [ + 43.3327, + 25.3339 + ], + "end": [ + 65.6066, + 47.6077 + ], + "original_corners": [ + [ + 54.5262, + 25.3339 + ], + [ + 43.3327, + 36.5274 + ], + [ + 54.4131, + 47.6077 + ], + [ + 65.6066, + 36.4142 + ] + ] + } + ] +} \ No newline at end of file diff --git a/aaa_160628_20260209.json b/aaa_160628_20260209.json new file mode 100644 index 0000000..8760a78 --- /dev/null +++ b/aaa_160628_20260209.json @@ -0,0 +1,144 @@ +{ + "scan_info": { + "friendly_name": "test2", + "waveform_prefix": "aaa", + "data_directory": "/opt/scanengine-3", + "timestamp": "2026-02-09T16:06:28.364606", + "scan_finished": true, + "completion_timestamp": "2026-02-09T16:27:37.121723" + }, + "scan_parameters": { + "number_of_angles": 4, + "row_spacing_mm": 0.25, + "scan_type": "standalone" + }, + "scan_area": { + "x_start_mm": 54.83, + "x_delta_mm": 9.83, + "y_start_mm": 43.83, + "y_delta_mm": -9.17, + "sample_size": "1.25\"" + }, + "scan_boxes": [ + { + "angle_index": 0, + "angle_degrees": 0.0, + "start": [ + 54.83, + 34.66 + ], + "end": [ + 64.66, + 43.83 + ], + "original_corners": [ + [ + 54.83, + 43.83 + ], + [ + 64.66, + 43.83 + ], + [ + 64.66, + 34.66 + ], + [ + 54.83, + 34.66 + ] + ] + }, + { + "angle_index": 1, + "angle_degrees": 45.0, + "start": [ + 48.636, + 34.6394 + ], + "end": [ + 62.0711, + 48.0744 + ], + "original_corners": [ + [ + 48.636, + 41.1235 + ], + [ + 55.5869, + 48.0744 + ], + [ + 62.0711, + 41.5902 + ], + [ + 55.1202, + 34.6394 + ] + ] + }, + { + "angle_index": 2, + "angle_degrees": 90.0, + "start": [ + 46.17, + 34.83 + ], + "end": [ + 55.34, + 44.66 + ], + "original_corners": [ + [ + 46.17, + 34.83 + ], + [ + 46.17, + 44.66 + ], + [ + 55.34, + 44.66 + ], + [ + 55.34, + 34.83 + ] + ] + }, + { + "angle_index": 3, + "angle_degrees": 135.0, + "start": [ + 41.9256, + 28.636 + ], + "end": [ + 55.3606, + 42.0711 + ], + "original_corners": [ + [ + 48.8765, + 28.636 + ], + [ + 41.9256, + 35.5869 + ], + [ + 48.4098, + 42.0711 + ], + [ + 55.3606, + 35.1202 + ] + ] + } + ] +} \ No newline at end of file diff --git a/config.json b/config.json index 83f928a..fe6d4d7 100644 --- a/config.json +++ b/config.json @@ -11,8 +11,8 @@ }, "scanning_stage": { "scan_velocity_mm_s": "200", - "scan_acceleration_mm_s2": "500", - "x_trigger_mode": 5, + "scan_acceleration_mm_s2": "1500", + "x_trigger_mode": 6, "y_trigger_mode": 0, "optical_axis_x_mm": "55", "optical_axis_y_mm": "37.5" @@ -25,4 +25,4 @@ "scratch_directory": "/opt/", "save_location": "pc" } -} +} \ No newline at end of file diff --git a/hardware/__init__.py b/hardware/__init__.py index e40b793..08762d3 100644 --- a/hardware/__init__.py +++ b/hardware/__init__.py @@ -1,5 +1,5 @@ """Hardware driver modules for ScanEngine-3""" -from .bbd202 import * +from .pybbd202 import ThorlabsServoDriver, TriggerBitsServo, AXIS_X, AXIS_Y, CONTROLLER from .uc480_camera import * from .tektronix_base import * from .coherent_hops_laser import * diff --git a/hardware/bbd202.py b/hardware/bbd202.py deleted file mode 100644 index eb8eadf..0000000 --- a/hardware/bbd202.py +++ /dev/null @@ -1,3147 +0,0 @@ -''' - MLS203/BBD202 Thorlabs Controller Driver - Thomas Ales | Rev 0 | Jan 2026 - This does contain LLM generated code. Cause if you think - I am gonna copy 443 pages of the protocol manual by hand - you're crazy. - - Only supports two axes, but easily extendable to 3 if you have a 203 controller -''' - - -from pyftdi.ftdi import Ftdi -import time -import struct -import threading -import queue -from typing import Union, Optional, Callable -from dataclasses import dataclass -from enum import IntEnum, IntFlag - - -class MsgId(IntEnum): - """APT message IDs.""" - HW_DISCONNECT = 0x0002 - HW_REQ_INFO = 0x0005 - HW_GET_INFO = 0x0006 - HW_STOP_UPDATEMSGS = 0x0012 - HW_START_UPDATEMSGS = 0x0011 - MOD_IDENTIFY = 0x0223 - MOD_SET_CHANENABLESTATE = 0x0210 - MOD_REQ_CHANENABLESTATE = 0x0211 - MOD_GET_CHANENABLESTATE = 0x0212 - MOT_MOVE_HOME = 0x0443 - MOT_MOVE_HOMED = 0x0444 - MOT_MOVE_ABSOLUTE = 0x0453 - MOT_MOVE_COMPLETED = 0x0464 - MOT_REQ_USTATUSUPDATE = 0x0490 - MOT_GET_USTATUSUPDATE = 0x0491 - MOT_GET_DCSTATUSUPDATE = 0x0491 # Alias for USTATUSUPDATE - MOT_ACK_USTATUSUPDATE = 0x0492 - MOT_SET_ENCCOUNTER = 0x0409 - MOT_REQ_ENCCOUNTER = 0x040A - MOT_GET_ENCCOUNTER = 0x040B - MOT_REQ_POSCOUNTER = 0x0411 - MOT_GET_POSCOUNTER = 0x0412 - MOT_SET_VELPARAMS = 0x0413 - MOT_REQ_VELPARAMS = 0x0414 - MOT_GET_VELPARAMS = 0x0415 - MOT_SET_JOGPARAMS = 0x0416 - MOT_REQ_JOGPARAMS = 0x0417 - MOT_GET_JOGPARAMS = 0x0418 - MOT_SET_MOVERELPARAMS = 0x0445 - MOT_REQ_MOVERELPARAMS = 0x0446 - MOT_GET_MOVERELPARAMS = 0x0447 - MOT_MOVE_RELATIVE = 0x0448 - MOT_SET_MOVEABSPARAMS = 0x0450 - MOT_REQ_MOVEABSPARAMS = 0x0451 - MOT_GET_MOVEABSPARAMS = 0x0452 - MOT_MOVE_STOP = 0x0465 - MOT_MOVE_STOPPED = 0x0466 - MOT_SET_TRIGGER = 0x0500 - MOT_REQ_TRIGGER = 0x0501 - MOT_GET_TRIGGER = 0x0502 - - -class ChannelEnableState(IntEnum): - """Channel enable state values.""" - DISABLED = 0x02 - ENABLED = 0x01 - - -class JogMode(IntEnum): - """Jog mode values.""" - CONTINUOUS = 0x01 - SINGLE_STEP = 0x02 - - -class StopMode(IntEnum): - """Stop mode values.""" - IMMEDIATE = 0x01 - CONTROLLED = 0x02 - - -class TriggerMode(IntEnum): - """Trigger mode values.""" - DISABLED = 0x00 - IN_OUT_RELATIVE_MOVE = 0x01 - IN_OUT_ABSOLUTE_MOVE = 0x02 - IN_OUT_HOME = 0x03 - IN_OUT_STOP = 0x04 - OUT_ONLY = 0x10 - - -class MotorStatusBits(IntFlag): - """Motor status bit flags.""" - CWHARDLIMIT = 0x00000001 # Clockwise hard limit triggered - CCWHARDLIMIT = 0x00000002 # Counter-clockwise hard limit triggered - CWSOFTLIMIT = 0x00000004 # Clockwise software limit triggered - CCWSOFTLIMIT = 0x00000008 # Counter-clockwise software limit triggered - INMOTIONCW = 0x00000010 # Stage is in motion clockwise - INMOTIONCCW = 0x00000020 # Stage is in motion counter-clockwise - CONNECTED = 0x00000100 # Motor recognized by controller - HOMING = 0x00000200 # Motor is homing - HOMED = 0x00000400 # Motor completed homing, position valid - INITIALIZING = 0x00000800 # Motor is performing phase initialization - TRACKING = 0x00001000 # Position is within tracking window - SETTLED = 0x00002000 # Motor not moving and settled at target - POSITIONERROR = 0x00004000 # Actual position outside tracking window - OVERTEMP = 0x00200000 # Overtemperature error - BUSVOLTFAULT = 0x00400000 # Supply voltage too low - COMMUTATIONERROR = 0x00800000 # Motor commutation error (power cycle required) - OVERLOAD = 0x01000000 # Motor overload/overcurrent condition - ENCODERFAULT = 0x02000000 # Encoder error - OVERCURRENT = 0x04000000 # Continuous current limit exceeded - POWEROK = 0x10000000 # Power supply is OK - ACTIVE = 0x20000000 # Controller executing motion command - ERROR = 0x40000000 # Other errors - ENABLED = 0x80000000 # Motor output enabled, maintaining position - - -@dataclass -class AptMessage: - """Represents a parsed APT message.""" - msg_id: int - param1: int - param2: int - dest: int - source: int - data: bytes - raw: bytes - timestamp: float - - -class MotionController: - """Controller class for Thorlabs APT motion controllers via FTDI interface.""" - - TYPE_FORMATS = { - 'word': ' encoder count (int) - self._stage_positions = {} # dest -> position in mm (float) - self._position_lock = threading.Lock() - - self._velocity_params = {} # dest -> {'min_velocity': float, 'acceleration': float, 'max_velocity': float} - self._velocity_lock = threading.Lock() - - self._jog_params = {} # dest -> {'jog_mode': int, 'step_size': int, 'min_velocity': float, 'acceleration': float, 'max_velocity': float, 'stop_mode': int} - self._jog_lock = threading.Lock() - - self._move_rel_params = {} # dest -> {'relative_distance': float, 'relative_distance_counts': int} - self._move_rel_lock = threading.Lock() - - self._move_abs_params = {} # dest -> {'absolute_position': float, 'absolute_position_counts': int} - self._move_abs_lock = threading.Lock() - - self._move_completed = {} # dest -> {'position': float, 'position_counts': int, 'velocity': int, 'motor_current': int, 'status_bits': int} - self._move_completed_lock = threading.Lock() - - self._move_stopped = {} # dest -> {'position': float, 'position_counts': int, 'velocity': int, 'motor_current': int, 'status_bits': int} - self._move_stopped_lock = threading.Lock() - - self._status_bits = {} # dest -> int (updated on every MOVE_COMPLETED or MOVE_STOPPED) - self._status_bits_lock = threading.Lock() - - # TX queue for serialized command sending - all outgoing data goes through this queue - self._tx_queue: queue.Queue = queue.Queue() - self._tx_thread: Optional[threading.Thread] = None - self._tx_running = False - - self._rx_thread: Optional[threading.Thread] = None - self._rx_running = False - self._rx_queue: queue.Queue[AptMessage] = queue.Queue() - self._rx_lock = threading.Lock() - - self._callbacks: dict[int, list[Callable[[AptMessage], None]]] = {} - - self._waiters: dict[int, tuple[threading.Event, list]] = {} - self._waiter_lock = threading.Lock() - - # Pending move tracking for high-level motion control - self._pending_moves = {} # dest -> target_position (float) - self._pending_moves_lock = threading.Lock() - - # Last error tracking - self._last_error: Optional[str] = None - self._last_error_lock = threading.Lock() - - # Auto-ACK control - when True, ACKs are sent automatically in RX loop - # Set to False via stop_status_ack() for manual ACK mode during scanning - self._auto_ack_enabled = True - self._status_update_count_x = 0 # Track X status updates - self._status_update_count_y = 0 # Track Y status updates - - def connect(self, enable_updates: bool = True) -> None: - """ - Open connection to the controller. - - Args: - enable_updates: If True, enable status update messages. ACKs are sent - reactively when status updates are received. - - Note: - When enable_updates=True, the controller will send periodic status updates. - ACKs are automatically sent in response to each status update through - the TX queue, ensuring proper serialization with other commands. - """ - self.ftdi.open_from_url(self.url) - self.ftdi.set_baudrate(self.baudrate) - self.ftdi.set_line_property(8, 1, 'N') - self.ftdi.set_flowctrl('hw') - self.ftdi.set_rts(True) - self.ftdi.purge_buffers() - self._connected = True - time.sleep(0.1) - - # Start TX thread for serialized command sending - self._tx_running = True - self._tx_thread = threading.Thread(target=self._tx_loop, daemon=True) - self._tx_thread.start() - - self._rx_running = True - self._rx_thread = threading.Thread(target=self._rx_loop, daemon=True) - self._rx_thread.start() - - if enable_updates: - # Send to both axis channels to enable updates from each - self._send_raw(self._build_short_message(MsgId.HW_START_UPDATEMSGS, dest=self.DEST_X_AXIS)) - self._send_raw(self._build_short_message(MsgId.HW_START_UPDATEMSGS, dest=self.DEST_Y_AXIS)) - else: - self._send_raw(self._build_short_message(MsgId.HW_STOP_UPDATEMSGS, dest=self.DEST_X_AXIS)) - self._send_raw(self._build_short_message(MsgId.HW_STOP_UPDATEMSGS, dest=self.DEST_Y_AXIS)) - - time.sleep(0.1) - - def disconnect(self) -> None: - """Close connection to the controller.""" - if self._connected: - self._rx_running = False - if self._rx_thread: - self._rx_thread.join(timeout=1.0) - self._rx_thread = None - - try: - # Send disconnect directly to FTDI since we're shutting down TX thread - with self._rx_lock: - self.ftdi.write_data(self._build_short_message(MsgId.HW_DISCONNECT)) - time.sleep(0.05) - except: - pass - - # Stop TX thread after sending disconnect - self._tx_running = False - if self._tx_thread: - self._tx_thread.join(timeout=1.0) - self._tx_thread = None - - self.ftdi.close() - self._connected = False - self._hw_info = None - - def _build_short_message(self, msg_id: int, param1: int = 0, param2: int = 0, - dest: int = 0x11, source: int = 0x01) -> bytes: - """Build a 6-byte APT header-only message.""" - return struct.pack(' bytes: - """Build an APT message with data payload.""" - data_len = len(data) - header = struct.pack(' int: - """Queue raw bytes to be sent to the controller via the TX thread.""" - if not self._connected: - raise ConnectionError("Not connected to controller") - self._tx_queue.put(data) - return len(data) - - def _tx_loop(self) -> None: - """TX thread - processes all outgoing commands from the queue.""" - while self._tx_running: - try: - data = self._tx_queue.get(timeout=0.1) - if data is not None: - with self._rx_lock: - self.ftdi.write_data(data) - except queue.Empty: - pass - except Exception as e: - if self._tx_running: - print(f"TX error: {e}") - - def _rx_loop(self) -> None: - """Receiver thread loop - continuously reads and parses messages.""" - buffer = b'' - - while self._rx_running: - try: - with self._rx_lock: - chunk = self.ftdi.read_data(512) - - if chunk: - buffer += chunk - buffer = self._parse_buffer(buffer) - else: - time.sleep(0.01) - except Exception as e: - if self._rx_running: - print(f"RX error: {e}") - time.sleep(0.1) - - def _parse_buffer(self, buffer: bytes) -> bytes: - """Parse complete messages from buffer, return remaining bytes.""" - while len(buffer) >= 6: - msg_id = struct.unpack('= 12: - encoder_counts = struct.unpack(' 0 and count % 10 == 0: - axis = "X" if msg.source == 0x21 else "Y" - if len(msg.data) >= 14: - pos_counts = struct.unpack('= 14: - position_counts = struct.unpack('= 20: - min_vel_counts = struct.unpack('= 28: - jog_mode = struct.unpack('= 12: - relative_distance_counts = struct.unpack('= 12: - absolute_position_counts = struct.unpack('= 20: - position_counts = struct.unpack('= 20: - position_counts = struct.unpack(' None: - """ - Register a callback for a specific message ID. - - Args: - msg_id: Message ID to listen for - callback: Function to call when message is received - """ - if msg_id not in self._callbacks: - self._callbacks[msg_id] = [] - self._callbacks[msg_id].append(callback) - - def unregister_callback(self, msg_id: int, callback: Callable[[AptMessage], None]) -> None: - """Remove a callback for a specific message ID.""" - if msg_id in self._callbacks: - self._callbacks[msg_id].remove(callback) - - def wait_for_message(self, msg_id: int, timeout: float = 5.0) -> Optional[AptMessage]: - """ - Wait for a specific message ID. - - Args: - msg_id: Message ID to wait for - timeout: Timeout in seconds - - Returns: - The received message, or None if timeout - """ - event = threading.Event() - responses = [] - - with self._waiter_lock: - self._waiters[msg_id] = (event, responses) - - try: - if event.wait(timeout=timeout): - return responses[0] if responses else None - return None - finally: - with self._waiter_lock: - del self._waiters[msg_id] - - def send_command(self, msg_id: int, param1: int = 0, param2: int = 0, - data: Optional[bytes] = None, dest: int = 0x11, source: int = 0x01) -> None: - """ - Send a command to the controller. - - Args: - msg_id: Message ID - param1: Parameter 1 (for short messages) - param2: Parameter 2 (for short messages) - data: Data payload (for long messages) - dest: Destination address - source: Source address - """ - if data is not None: - msg = self._build_long_message(msg_id, data, dest, source) - else: - msg = self._build_short_message(msg_id, param1, param2, dest, source) - self._send_raw(msg) - - def send_and_wait(self, msg_id: int, response_id: int, param1: int = 0, param2: int = 0, - data: Optional[bytes] = None, timeout: float = 5.0) -> Optional[AptMessage]: - """ - Send a command and wait for a response. - - Args: - msg_id: Message ID to send - response_id: Message ID to wait for - param1: Parameter 1 (for short messages) - param2: Parameter 2 (for short messages) - data: Data payload (for long messages) - timeout: Timeout in seconds - - Returns: - The response message, or None if timeout - """ - event = threading.Event() - responses = [] - - with self._waiter_lock: - self._waiters[response_id] = (event, responses) - - try: - self.send_command(msg_id, param1, param2, data) - if event.wait(timeout=timeout): - return responses[0] if responses else None - return None - finally: - with self._waiter_lock: - del self._waiters[response_id] - - def get_message(self, timeout: float = 0.1) -> Optional[AptMessage]: - """ - Get the next message from the receive queue. - - Args: - timeout: Timeout in seconds - - Returns: - The next message, or None if queue is empty - """ - try: - return self._rx_queue.get(timeout=timeout) - except queue.Empty: - return None - - def get_all_messages(self) -> list[AptMessage]: - """Get all messages currently in the receive queue.""" - messages = [] - while True: - try: - messages.append(self._rx_queue.get_nowait()) - except queue.Empty: - break - return messages - - def convert(self, data: bytes, dtype: str) -> Union[int, str]: - """ - Convert a byte slice to the specified type. - - Args: - data: Byte slice to convert - dtype: Type specification: - - 'word': unsigned 16-bit integer - - 'short': signed 16-bit integer (2's complement) - - 'dword': unsigned 32-bit integer - - 'long': signed 32-bit integer (2's complement) - - 'char': single byte - - 'char[n]': string of n characters (e.g., 'char[8]') - - Returns: - Converted value (int for numeric types, str for char[n]) - """ - if dtype.startswith('char[') and dtype.endswith(']'): - n = int(dtype[5:-1]) - if len(data) < n: - raise ValueError(f"Data length {len(data)} < expected {n} for {dtype}") - return data[:n].decode('ascii').rstrip('\x00') - - if dtype not in self.TYPE_FORMATS: - raise ValueError(f"Unknown type: {dtype}") - - fmt = self.TYPE_FORMATS[dtype] - expected_size = struct.calcsize(fmt) - - if len(data) < expected_size: - raise ValueError(f"Data length {len(data)} < expected {expected_size} for {dtype}") - - return struct.unpack(fmt, data[:expected_size])[0] - - # Hardware info methods - - def _refresh_hw_info(self) -> None: - """Fetch and cache hardware info from the controller.""" - response = self.send_and_wait(MsgId.HW_REQ_INFO, MsgId.HW_GET_INFO, timeout=2.0) - - if response is None: - raise IOError("No response to HW_REQ_INFO") - - if len(response.raw) < 90: - raise IOError(f"Invalid HW_GET_INFO response: expected 90 bytes, got {len(response.raw)}") - - self._hw_info = response.raw - - def get_serial_number(self) -> int: - """Get the controller serial number.""" - if self._hw_info is None: - self._refresh_hw_info() - return self.convert(self._hw_info[6:10], 'dword') - - def get_model(self) -> str: - """Get the controller model name.""" - if self._hw_info is None: - self._refresh_hw_info() - return self.convert(self._hw_info[10:18], 'char[8]') - - def get_hw_type(self) -> int: - """Get the hardware type identifier.""" - if self._hw_info is None: - self._refresh_hw_info() - return self.convert(self._hw_info[18:20], 'word') - - def get_firmware_version(self) -> str: - """Get the firmware version as a string (major.interim.minor).""" - if self._hw_info is None: - self._refresh_hw_info() - - fw_minor = self.convert(self._hw_info[20:21], 'char') - fw_interim = self.convert(self._hw_info[21:22], 'char') - fw_major = self.convert(self._hw_info[22:23], 'char') - - return f"{fw_major}.{fw_interim}.{fw_minor}" - - def get_firmware_version_tuple(self) -> tuple: - """Get the firmware version as a tuple (major, interim, minor).""" - if self._hw_info is None: - self._refresh_hw_info() - - fw_minor = self.convert(self._hw_info[20:21], 'char') - fw_interim = self.convert(self._hw_info[21:22], 'char') - fw_major = self.convert(self._hw_info[22:23], 'char') - - return (fw_major, fw_interim, fw_minor) - - def get_notes(self) -> str: - """Get the controller notes/description string.""" - if self._hw_info is None: - self._refresh_hw_info() - return self.convert(self._hw_info[24:84], 'char[60]') - - def get_hw_version(self) -> int: - """Get the hardware version number.""" - if self._hw_info is None: - self._refresh_hw_info() - return self.convert(self._hw_info[84:86], 'word') - - def get_mod_state(self) -> int: - """Get the module state.""" - if self._hw_info is None: - self._refresh_hw_info() - return self.convert(self._hw_info[86:88], 'word') - - def get_num_channels(self) -> int: - """Get the number of channels.""" - if self._hw_info is None: - self._refresh_hw_info() - return self.convert(self._hw_info[88:90], 'word') - - def get_hw_info(self) -> dict: - """ - Get all hardware info as a dictionary. - - Returns: - Dictionary containing all hardware info fields - """ - if self._hw_info is None: - self._refresh_hw_info() - - return { - 'serial_number': self.get_serial_number(), - 'model': self.get_model(), - 'hw_type': self.get_hw_type(), - 'firmware_version': self.get_firmware_version(), - 'notes': self.get_notes(), - 'hw_version': self.get_hw_version(), - 'mod_state': self.get_mod_state(), - 'num_channels': self.get_num_channels(), - } - - # Position methods - - def get_position(self, dest: int, timeout: float = 5.0) -> Optional[float]: - """ - Get the current position of an axis by sending REQ and waiting for GET response. - - Args: - dest: Destination address (0x21 for X-axis, 0x22 for Y-axis) - timeout: Timeout in seconds - - Returns: - The position in mm as a float, or None if timeout - - Raises: - ValueError: If dest is invalid - """ - if dest not in (self.DEST_X_AXIS, self.DEST_Y_AXIS): - raise ValueError(f"Invalid destination: 0x{dest:02X}. Must be 0x21 (X-axis) or 0x22 (Y-axis)") - - # Create a unique waiter key combining message ID and expected source - waiter_key = (MsgId.MOT_GET_ENCCOUNTER, dest) - event = threading.Event() - responses = [] - - with self._waiter_lock: - self._waiters[waiter_key] = (event, responses) - - try: - # Send REQ message to the axis, device will respond with GET - self.send_command( - MsgId.MOT_REQ_ENCCOUNTER, - param1=0x01, - param2=0x00, - dest=dest, # Send to the axis - source=0x01 # From us - ) - - # Wait for GET response from device - if event.wait(timeout=timeout): - response = responses[0] if responses else None - if response is None: - return None - # Bytes 8-11 of message contain encoder counts as 32-bit signed little-endian integer - if len(response.raw) >= 12: - encoder_counts = struct.unpack(' Optional[float]: - """ - Get the current cached position of the X-axis. - - Returns: - The position in mm as a float, or None if not yet known - """ - with self._position_lock: - return self._stage_positions.get(self.DEST_X_AXIS) - - def get_stage_position_y(self) -> Optional[float]: - """ - Get the current cached position of the Y-axis. - - Returns: - The position in mm as a float, or None if not yet known - """ - with self._position_lock: - return self._stage_positions.get(self.DEST_Y_AXIS) - - def get_stage_positions(self) -> dict: - """ - Get all cached stage positions. - - Returns: - Dictionary mapping axis destination (0x21, 0x22) to position in mm - """ - with self._position_lock: - return dict(self._stage_positions) - - # Encoder count properties - - @property - def encoder_count_x(self) -> Optional[int]: - """ - Get the current cached encoder count of the X-axis. - - Returns: - The encoder count as an int, or None if not yet known - """ - with self._position_lock: - return self._encoder_counts.get(self.DEST_X_AXIS) - - @property - def encoder_count_y(self) -> Optional[int]: - """ - Get the current cached encoder count of the Y-axis. - - Returns: - The encoder count as an int, or None if not yet known - """ - with self._position_lock: - return self._encoder_counts.get(self.DEST_Y_AXIS) - - @property - def position_x(self) -> Optional[float]: - """ - Get the current cached position of the X-axis in mm. - - Returns: - The position in mm as a float, or None if not yet known - """ - with self._position_lock: - return self._stage_positions.get(self.DEST_X_AXIS) - - @property - def position_y(self) -> Optional[float]: - """ - Get the current cached position of the Y-axis in mm. - - Returns: - The position in mm as a float, or None if not yet known - """ - with self._position_lock: - return self._stage_positions.get(self.DEST_Y_AXIS) - - def poll_status(self) -> None: - """Request status updates from all axes (non-blocking).""" - # Request status from both axes - self.send_command(MsgId.MOT_REQ_USTATUSUPDATE, param1=0x01, param2=0x00, - dest=self.DEST_X_AXIS, source=0x01) - self.send_command(MsgId.MOT_REQ_USTATUSUPDATE, param1=0x01, param2=0x00, - dest=self.DEST_Y_AXIS, source=0x01) - - def request_status_update(self, dest: int) -> None: - """Request status update from specific axis (non-blocking).""" - self.send_command(MsgId.MOT_REQ_USTATUSUPDATE, param1=0x01, param2=0x00, - dest=dest, source=0x01) - - def check_for_errors(self) -> Optional[str]: - """Check if any axis has error flags set and store as last error.""" - error_msg = None - with self._status_bits_lock: - x_bits = self._status_bits.get(self.DEST_X_AXIS, 0) - y_bits = self._status_bits.get(self.DEST_Y_AXIS, 0) - - if x_bits & MotorStatusBits.ERROR: - error_msg = "X-axis error detected" - elif y_bits & MotorStatusBits.ERROR: - error_msg = "Y-axis error detected" - - if error_msg: - with self._last_error_lock: - self._last_error = error_msg - - return error_msg - - def clear_last_error(self) -> None: - """Clear the stored last error message.""" - with self._last_error_lock: - self._last_error = None - - @property - def last_error(self) -> Optional[str]: - """Get the last error message that was detected.""" - with self._last_error_lock: - return self._last_error - - def is_move_pending(self) -> bool: - """Check if any moves are pending.""" - with self._pending_moves_lock: - return len(self._pending_moves) > 0 - - def get_pending_targets(self) -> dict: - """Get pending move targets as dict mapping dest -> target_position.""" - with self._pending_moves_lock: - return dict(self._pending_moves) - - # Velocity parameter properties - - @property - def velocity_params_x(self) -> Optional[dict]: - """ - Get the current cached velocity parameters for the X-axis. - - Returns: - Dictionary with 'min_velocity', 'acceleration', 'max_velocity' (mm/s or mm/sĀ²), - or None if not yet known - """ - with self._velocity_lock: - params = self._velocity_params.get(self.DEST_X_AXIS) - return dict(params) if params else None - - @property - def velocity_params_y(self) -> Optional[dict]: - """ - Get the current cached velocity parameters for the Y-axis. - - Returns: - Dictionary with 'min_velocity', 'acceleration', 'max_velocity' (mm/s or mm/sĀ²), - or None if not yet known - """ - with self._velocity_lock: - params = self._velocity_params.get(self.DEST_Y_AXIS) - return dict(params) if params else None - - @property - def min_velocity_x(self) -> Optional[float]: - """Get the cached minimum velocity for X-axis in mm/s.""" - with self._velocity_lock: - params = self._velocity_params.get(self.DEST_X_AXIS) - return params['min_velocity'] if params else None - - @property - def min_velocity_y(self) -> Optional[float]: - """Get the cached minimum velocity for Y-axis in mm/s.""" - with self._velocity_lock: - params = self._velocity_params.get(self.DEST_Y_AXIS) - return params['min_velocity'] if params else None - - @property - def acceleration_x(self) -> Optional[float]: - """Get the cached acceleration for X-axis in mm/sĀ².""" - with self._velocity_lock: - params = self._velocity_params.get(self.DEST_X_AXIS) - return params['acceleration'] if params else None - - @property - def acceleration_y(self) -> Optional[float]: - """Get the cached acceleration for Y-axis in mm/sĀ².""" - with self._velocity_lock: - params = self._velocity_params.get(self.DEST_Y_AXIS) - return params['acceleration'] if params else None - - @property - def max_velocity_x(self) -> Optional[float]: - """Get the cached maximum velocity for X-axis in mm/s.""" - with self._velocity_lock: - params = self._velocity_params.get(self.DEST_X_AXIS) - return params['max_velocity'] if params else None - - @property - def max_velocity_y(self) -> Optional[float]: - """Get the cached maximum velocity for Y-axis in mm/s.""" - with self._velocity_lock: - params = self._velocity_params.get(self.DEST_Y_AXIS) - return params['max_velocity'] if params else None - - # Jog parameter properties - - @property - def jog_params_x(self) -> Optional[dict]: - """ - Get the current cached jog parameters for the X-axis. - - Returns: - Dictionary with 'jog_mode', 'step_size', 'min_velocity', 'acceleration', - 'max_velocity', 'stop_mode', or None if not yet known - """ - with self._jog_lock: - params = self._jog_params.get(self.DEST_X_AXIS) - return dict(params) if params else None - - @property - def jog_params_y(self) -> Optional[dict]: - """ - Get the current cached jog parameters for the Y-axis. - - Returns: - Dictionary with 'jog_mode', 'step_size', 'min_velocity', 'acceleration', - 'max_velocity', 'stop_mode', or None if not yet known - """ - with self._jog_lock: - params = self._jog_params.get(self.DEST_Y_AXIS) - return dict(params) if params else None - - # Move relative parameter properties - - @property - def move_rel_params_x(self) -> Optional[dict]: - """ - Get the current cached move relative parameters for the X-axis. - - Returns: - Dictionary with 'relative_distance' (mm) and 'relative_distance_counts', - or None if not yet known - """ - with self._move_rel_lock: - params = self._move_rel_params.get(self.DEST_X_AXIS) - return dict(params) if params else None - - @property - def move_rel_params_y(self) -> Optional[dict]: - """ - Get the current cached move relative parameters for the Y-axis. - - Returns: - Dictionary with 'relative_distance' (mm) and 'relative_distance_counts', - or None if not yet known - """ - with self._move_rel_lock: - params = self._move_rel_params.get(self.DEST_Y_AXIS) - return dict(params) if params else None - - @property - def relative_distance_x(self) -> Optional[float]: - """Get the cached relative move distance for X-axis in mm.""" - with self._move_rel_lock: - params = self._move_rel_params.get(self.DEST_X_AXIS) - return params['relative_distance'] if params else None - - @property - def relative_distance_y(self) -> Optional[float]: - """Get the cached relative move distance for Y-axis in mm.""" - with self._move_rel_lock: - params = self._move_rel_params.get(self.DEST_Y_AXIS) - return params['relative_distance'] if params else None - - # Move absolute parameter properties - - @property - def move_abs_params_x(self) -> Optional[dict]: - """ - Get the current cached move absolute parameters for the X-axis. - - Returns: - Dictionary with 'absolute_position' (mm) and 'absolute_position_counts', - or None if not yet known - """ - with self._move_abs_lock: - params = self._move_abs_params.get(self.DEST_X_AXIS) - return dict(params) if params else None - - @property - def move_abs_params_y(self) -> Optional[dict]: - """ - Get the current cached move absolute parameters for the Y-axis. - - Returns: - Dictionary with 'absolute_position' (mm) and 'absolute_position_counts', - or None if not yet known - """ - with self._move_abs_lock: - params = self._move_abs_params.get(self.DEST_Y_AXIS) - return dict(params) if params else None - - @property - def absolute_position_x(self) -> Optional[float]: - """Get the cached absolute move position for X-axis in mm.""" - with self._move_abs_lock: - params = self._move_abs_params.get(self.DEST_X_AXIS) - return params['absolute_position'] if params else None - - @property - def absolute_position_y(self) -> Optional[float]: - """Get the cached absolute move position for Y-axis in mm.""" - with self._move_abs_lock: - params = self._move_abs_params.get(self.DEST_Y_AXIS) - return params['absolute_position'] if params else None - - # Move completed status properties - - @property - def move_completed_x(self) -> Optional[dict]: - """ - Get the most recent move completion status for the X-axis. - - Returns: - Dictionary with 'position' (mm), 'position_counts', 'velocity', - 'motor_current', 'status_bits', or None if no move completed yet - """ - with self._move_completed_lock: - status = self._move_completed.get(self.DEST_X_AXIS) - return dict(status) if status else None - - @property - def move_completed_y(self) -> Optional[dict]: - """ - Get the most recent move completion status for the Y-axis. - - Returns: - Dictionary with 'position' (mm), 'position_counts', 'velocity', - 'motor_current', 'status_bits', or None if no move completed yet - """ - with self._move_completed_lock: - status = self._move_completed.get(self.DEST_Y_AXIS) - return dict(status) if status else None - - # Move stopped status properties - - @property - def move_stopped_x(self) -> Optional[dict]: - """ - Get the most recent move stopped status for the X-axis. - - Returns: - Dictionary with 'position' (mm), 'position_counts', 'velocity', - 'motor_current', 'status_bits', or None if no move stopped event yet - """ - with self._move_stopped_lock: - status = self._move_stopped.get(self.DEST_X_AXIS) - return dict(status) if status else None - - @property - def move_stopped_y(self) -> Optional[dict]: - """ - Get the most recent move stopped status for the Y-axis. - - Returns: - Dictionary with 'position' (mm), 'position_counts', 'velocity', - 'motor_current', 'status_bits', or None if no move stopped event yet - """ - with self._move_stopped_lock: - status = self._move_stopped.get(self.DEST_Y_AXIS) - return dict(status) if status else None - - # Status bits properties - - @property - def status_bits_x(self) -> Optional[int]: - """ - Get the current status bits for the X-axis. - - Returns: - Current status bits as int, or None if not yet known - """ - with self._status_bits_lock: - return self._status_bits.get(self.DEST_X_AXIS) - - @property - def status_bits_y(self) -> Optional[int]: - """ - Get the current status bits for the Y-axis. - - Returns: - Current status bits as int, or None if not yet known - """ - with self._status_bits_lock: - return self._status_bits.get(self.DEST_Y_AXIS) - - # X-axis status bit properties - - @property - def is_enabled_x(self) -> bool: - """Check if X-axis motor output is enabled.""" - bits = self.status_bits_x - return bool(bits and (MotorStatusBits.ENABLED in MotorStatusBits(bits))) - - @property - def is_homed_x(self) -> bool: - """Check if X-axis has been homed (position count is valid).""" - bits = self.status_bits_x - return bool(bits and (MotorStatusBits.HOMED in MotorStatusBits(bits))) - - @property - def is_homing_x(self) -> bool: - """Check if X-axis is currently homing.""" - bits = self.status_bits_x - return bool(bits and (MotorStatusBits.HOMING in MotorStatusBits(bits))) - - @property - def is_in_motion_x(self) -> bool: - """Check if X-axis is currently in motion.""" - bits = self.status_bits_x - return bool(bits and self.is_in_motion(bits)) - - @property - def is_settled_x(self) -> bool: - """Check if X-axis is settled at target position.""" - bits = self.status_bits_x - return bool(bits and (MotorStatusBits.SETTLED in MotorStatusBits(bits))) - - @property - def is_tracking_x(self) -> bool: - """Check if X-axis position is within tracking window.""" - bits = self.status_bits_x - return bool(bits and (MotorStatusBits.TRACKING in MotorStatusBits(bits))) - - @property - def is_connected_x(self) -> bool: - """Check if X-axis motor is recognized by controller.""" - bits = self.status_bits_x - return bool(bits and (MotorStatusBits.CONNECTED in MotorStatusBits(bits))) - - @property - def has_errors_x(self) -> bool: - """Check if X-axis has any error conditions.""" - bits = self.status_bits_x - return bool(bits and self.has_errors(bits)) - - @property - def power_ok_x(self) -> bool: - """Check if X-axis power supply is OK.""" - bits = self.status_bits_x - return bool(bits and (MotorStatusBits.POWEROK in MotorStatusBits(bits))) - - @property - def is_active_x(self) -> bool: - """Check if X-axis controller is executing a motion command.""" - bits = self.status_bits_x - return bool(bits and (MotorStatusBits.ACTIVE in MotorStatusBits(bits))) - - @property - def at_cw_limit_x(self) -> bool: - """Check if X-axis is at clockwise limit (hard or soft).""" - bits = self.status_bits_x - if not bits: - return False - status = MotorStatusBits(bits) - return bool((MotorStatusBits.CWHARDLIMIT in status) or (MotorStatusBits.CWSOFTLIMIT in status)) - - @property - def at_ccw_limit_x(self) -> bool: - """Check if X-axis is at counter-clockwise limit (hard or soft).""" - bits = self.status_bits_x - if not bits: - return False - status = MotorStatusBits(bits) - return bool((MotorStatusBits.CCWHARDLIMIT in status) or (MotorStatusBits.CCWSOFTLIMIT in status)) - - # Y-axis status bit properties - - @property - def is_enabled_y(self) -> bool: - """Check if Y-axis motor output is enabled.""" - bits = self.status_bits_y - return bool(bits and (MotorStatusBits.ENABLED in MotorStatusBits(bits))) - - @property - def is_homed_y(self) -> bool: - """Check if Y-axis has been homed (position count is valid).""" - bits = self.status_bits_y - return bool(bits and (MotorStatusBits.HOMED in MotorStatusBits(bits))) - - @property - def is_homing_y(self) -> bool: - """Check if Y-axis is currently homing.""" - bits = self.status_bits_y - return bool(bits and (MotorStatusBits.HOMING in MotorStatusBits(bits))) - - @property - def is_in_motion_y(self) -> bool: - """Check if Y-axis is currently in motion.""" - bits = self.status_bits_y - return bool(bits and self.is_in_motion(bits)) - - @property - def is_settled_y(self) -> bool: - """Check if Y-axis is settled at target position.""" - bits = self.status_bits_y - return bool(bits and (MotorStatusBits.SETTLED in MotorStatusBits(bits))) - - @property - def is_tracking_y(self) -> bool: - """Check if Y-axis position is within tracking window.""" - bits = self.status_bits_y - return bool(bits and (MotorStatusBits.TRACKING in MotorStatusBits(bits))) - - @property - def is_connected_y(self) -> bool: - """Check if Y-axis motor is recognized by controller.""" - bits = self.status_bits_y - return bool(bits and (MotorStatusBits.CONNECTED in MotorStatusBits(bits))) - - @property - def has_errors_y(self) -> bool: - """Check if Y-axis has any error conditions.""" - bits = self.status_bits_y - return bool(bits and self.has_errors(bits)) - - @property - def power_ok_y(self) -> bool: - """Check if Y-axis power supply is OK.""" - bits = self.status_bits_y - return bool(bits and (MotorStatusBits.POWEROK in MotorStatusBits(bits))) - - @property - def is_active_y(self) -> bool: - """Check if Y-axis controller is executing a motion command.""" - bits = self.status_bits_y - return bool(bits and (MotorStatusBits.ACTIVE in MotorStatusBits(bits))) - - @property - def at_cw_limit_y(self) -> bool: - """Check if Y-axis is at clockwise limit (hard or soft).""" - bits = self.status_bits_y - if not bits: - return False - status = MotorStatusBits(bits) - return bool((MotorStatusBits.CWHARDLIMIT in status) or (MotorStatusBits.CWSOFTLIMIT in status)) - - @property - def at_ccw_limit_y(self) -> bool: - """Check if Y-axis is at counter-clockwise limit (hard or soft).""" - bits = self.status_bits_y - if not bits: - return False - status = MotorStatusBits(bits) - return bool((MotorStatusBits.CCWHARDLIMIT in status) or (MotorStatusBits.CCWSOFTLIMIT in status)) - - def start_update_messages(self) -> None: - """ - Start automatic status update messages from the controller. - - Once started, the controller will periodically send status update messages - containing position, velocity, and status information. These can be captured - by registering a callback for the update message type. - - Note: On BBD202/BBD203, HW_START_UPDATEMSGS must be sent to each axis - channel (0x21, 0x22) rather than the generic destination (0x11). - """ - # Send to both axis channels to enable updates from each - self.send_command( - MsgId.HW_START_UPDATEMSGS, - param1=0x00, - param2=0x00, - dest=self.DEST_X_AXIS, # 0x21 - source=0x01 - ) - self.send_command( - MsgId.HW_START_UPDATEMSGS, - param1=0x00, - param2=0x00, - dest=self.DEST_Y_AXIS, # 0x22 - source=0x01 - ) - print("Started automatic status update messages") - - def stop_update_messages(self) -> None: - """ - Stop automatic status update messages from the controller. - - This stops the periodic status updates that were started with - start_update_messages(). - """ - self.send_command( - MsgId.HW_STOP_UPDATEMSGS, - param1=0x00, - param2=0x00, - dest=0x11, # Generic destination - source=0x01 - ) - print("Stopped automatic status update messages") - - def start_status_ack(self) -> None: - """ - Enable automatic ACK mode. - - When enabled, ACKs are sent automatically in the RX loop whenever - a status update message is received from the controller. - - Use this after scanning operations that require manual ACK mode. - """ - self._auto_ack_enabled = True - - def stop_status_ack(self) -> None: - """ - Disable automatic ACK mode (switch to manual ACK mode). - - When disabled, ACKs are NOT sent automatically. You must call - ack_status_update() manually after each move completes. - - Use this during scanning operations for more precise control over - when ACKs are sent. The snake test pattern is: - mc.stop_status_ack() - for each move: - mc.move_to_fast(...) - while not mc.poll_until_idle(...): - ... - mc.ack_status_update() # Manual ACK after move completes - mc.start_status_ack() - """ - self._auto_ack_enabled = False - - def ack_status_update(self) -> None: - """ - Send a single status update ACK manually. - - Note: ACKs are now sent automatically in response to status update messages. - This method is retained for debugging or manual control if needed. - """ - self.send_command( - MsgId.MOT_ACK_USTATUSUPDATE, - param1=0x00, - param2=0x00, - dest=0x11, - source=0x01 - ) - - def set_channel_enable_state(self, dest: int, enabled: bool) -> None: - """ - Set the channel enable state. - - Args: - dest: Destination address (0x11 for X-axis, 0x22 for Y-axis) - enabled: True to enable, False to disable - - Raises: - ValueError: If dest is invalid - """ - if dest not in (self.DEST_X_AXIS, self.DEST_Y_AXIS): - raise ValueError(f"Invalid destination: 0x{dest:02X}. Must be 0x11 (X-axis) or 0x22 (Y-axis)") - - en_state = ChannelEnableState.ENABLED if enabled else ChannelEnableState.DISABLED - - self.send_command( - MsgId.MOD_SET_CHANENABLESTATE, - param1=0x02, # Reserved byte - param2=en_state, - dest=dest, - source=0x01 - ) - - def get_channel_enable_state(self, dest: int, timeout: float = 5.0) -> Optional[bool]: - """ - Get the channel enable state by sending REQ and waiting for GET response. - - Args: - dest: Destination address (0x21 for X-axis, 0x22 for Y-axis) - timeout: Timeout in seconds - - Returns: - True if enabled, False if disabled, None if timeout - - Raises: - ValueError: If dest is invalid - """ - if dest not in (self.DEST_X_AXIS, self.DEST_Y_AXIS): - raise ValueError(f"Invalid destination: 0x{dest:02X}. Must be 0x21 (X-axis) or 0x22 (Y-axis)") - - # Create a unique waiter key combining message ID and expected source - waiter_key = (MsgId.MOD_GET_CHANENABLESTATE, dest) - event = threading.Event() - responses = [] - - with self._waiter_lock: - self._waiters[waiter_key] = (event, responses) - - try: - # Send REQ message to the axis, device will respond with GET - self.send_command( - MsgId.MOD_REQ_CHANENABLESTATE, - param1=0x01, - param2=0x00, - dest=dest, # Send to the axis - source=0x01 # From us - ) - - # Wait for GET response from device - if event.wait(timeout=timeout): - response = responses[0] if responses else None - if response is None: - return None - en_state = response.param2 - return en_state == ChannelEnableState.ENABLED - return None - finally: - with self._waiter_lock: - if waiter_key in self._waiters: - del self._waiters[waiter_key] - - # Homing methods - - def home_axis(self, dest: int, timeout: float = 20.0) -> bool: - """ - Home an axis by sending MOVE_HOME and waiting for MOVE_HOMED response. - - The controller will not respond until the axis is fully homed, which can - take 10-15 seconds or more depending on the axis position. After homing, - the position automatically resets to 0. - - Args: - dest: Destination address (0x21 for X-axis, 0x22 for Y-axis) - timeout: Timeout in seconds (default 20s to account for homing time) - - Returns: - True if homing succeeded, False if timeout - - Raises: - ValueError: If dest is invalid - """ - if dest not in (self.DEST_X_AXIS, self.DEST_Y_AXIS): - raise ValueError(f"Invalid destination: 0x{dest:02X}. Must be 0x21 (X-axis) or 0x22 (Y-axis)") - - # Create a unique waiter key combining message ID and expected source - waiter_key = (MsgId.MOT_MOVE_HOMED, dest) - event = threading.Event() - responses = [] - - with self._waiter_lock: - self._waiters[waiter_key] = (event, responses) - - try: - # Send MOVE_HOME message to the axis - self.send_command( - MsgId.MOT_MOVE_HOME, - param1=0x01, # Channel ID - param2=0x00, - dest=dest, # Send to the axis - source=0x01 # From us - ) - - print(f"Homing axis 0x{dest:02X}... (this may take 10-15 seconds)") - - # Wait for MOVE_HOMED response from device - if event.wait(timeout=timeout): - return True - else: - print(f"Timeout waiting for homing response from axis 0x{dest:02X}") - return False - finally: - with self._waiter_lock: - if waiter_key in self._waiters: - del self._waiters[waiter_key] - - def home_x_axis(self, timeout: float = 20.0) -> bool: - """ - Home the X-axis. Position will reset to 0 after homing. - - Args: - timeout: Timeout in seconds (default 20s) - - Returns: - True if homing succeeded, False if timeout - """ - return self.home_axis(self.DEST_X_AXIS, timeout=timeout) - - def home_y_axis(self, timeout: float = 20.0) -> bool: - """ - Home the Y-axis. Position will reset to 0 after homing. - - Args: - timeout: Timeout in seconds (default 20s) - - Returns: - True if homing succeeded, False if timeout - """ - return self.home_axis(self.DEST_Y_AXIS, timeout=timeout) - - # Velocity parameter methods - - def set_velocity_params(self, dest: int, min_velocity: float, acceleration: float, - max_velocity: float) -> None: - """ - Set velocity parameters for an axis. - - Args: - dest: Destination address (0x21 for X-axis, 0x22 for Y-axis) - min_velocity: Minimum velocity in mm/s - acceleration: Acceleration in mm/sĀ² - max_velocity: Maximum velocity in mm/s - - Raises: - ValueError: If dest is invalid - """ - if dest not in (self.DEST_X_AXIS, self.DEST_Y_AXIS): - raise ValueError(f"Invalid destination: 0x{dest:02X}. Must be 0x21 (X-axis) or 0x22 (Y-axis)") - - # Convert from mm/s and mm/sĀ² to encoder counts - min_vel_counts = int(min_velocity * self.VELOCITY_SCALING) - accel_counts = int(acceleration * self.ACCELERATION_SCALING) - max_vel_counts = int(max_velocity * self.VELOCITY_SCALING) - - # Build the 14-byte data payload - data = struct.pack(' Optional[dict]: - """ - Get velocity parameters for an axis. - - Args: - dest: Destination address (0x21 for X-axis, 0x22 for Y-axis) - timeout: Timeout in seconds - - Returns: - Dictionary with keys: 'min_velocity', 'acceleration', 'max_velocity' (all in mm/s or mm/sĀ²), - or None if timeout - - Raises: - ValueError: If dest is invalid - """ - if dest not in (self.DEST_X_AXIS, self.DEST_Y_AXIS): - raise ValueError(f"Invalid destination: 0x{dest:02X}. Must be 0x21 (X-axis) or 0x22 (Y-axis)") - - # Create a unique waiter key combining message ID and expected source - waiter_key = (MsgId.MOT_GET_VELPARAMS, dest) - event = threading.Event() - responses = [] - - with self._waiter_lock: - self._waiters[waiter_key] = (event, responses) - - try: - # Send REQ message to the axis - self.send_command( - MsgId.MOT_REQ_VELPARAMS, - param1=0x01, - param2=0x00, - dest=dest, - source=0x01 - ) - - # Wait for GET response - if event.wait(timeout=timeout): - response = responses[0] if responses else None - if response is None: - return None - - # Parse the 20-byte message - if len(response.raw) >= 20: - channel_id = struct.unpack(' bool: - """ - Set acceleration for an axis while preserving current velocity settings. - - This is a convenience method that retrieves current velocity parameters, - then updates only the acceleration value. - - Args: - dest: Destination address (0x21 for X-axis, 0x22 for Y-axis) - acceleration: Acceleration in mm/sĀ² - timeout: Timeout for retrieving current parameters - - Returns: - True if successful, False if unable to retrieve current parameters - - Raises: - ValueError: If dest is invalid - """ - # Get current velocity parameters - current_params = self.get_velocity_params(dest, timeout=timeout) - if current_params is None: - return False - - # Set velocity parameters with new acceleration - self.set_velocity_params( - dest, - min_velocity=current_params['min_velocity'], - acceleration=acceleration, - max_velocity=current_params['max_velocity'] - ) - return True - - def get_acceleration(self, dest: int, timeout: float = 5.0) -> Optional[float]: - """ - Get acceleration for an axis. - - This is a convenience method that retrieves velocity parameters - and returns only the acceleration value. - - Args: - dest: Destination address (0x21 for X-axis, 0x22 for Y-axis) - timeout: Timeout in seconds - - Returns: - Acceleration in mm/sĀ², or None if timeout - - Raises: - ValueError: If dest is invalid - """ - params = self.get_velocity_params(dest, timeout=timeout) - if params is None: - return None - return params['acceleration'] - - # Jog parameter methods - - def set_jog_params(self, dest: int, jog_mode: int, step_size: float, - min_velocity: float, acceleration: float, max_velocity: float, - stop_mode: int) -> None: - """ - Set jog parameters for an axis. - - Args: - dest: Destination address (0x21 for X-axis, 0x22 for Y-axis) - jog_mode: Jog mode (JogMode.CONTINUOUS=1 or JogMode.SINGLE_STEP=2) - step_size: Step size in mm (for single step mode) - min_velocity: Minimum velocity in mm/s - acceleration: Acceleration in mm/sĀ² - max_velocity: Maximum velocity in mm/s - stop_mode: Stop mode (StopMode.IMMEDIATE=1 or StopMode.CONTROLLED=2) - - Raises: - ValueError: If dest is invalid - """ - if dest not in (self.DEST_X_AXIS, self.DEST_Y_AXIS): - raise ValueError(f"Invalid destination: 0x{dest:02X}. Must be 0x21 (X-axis) or 0x22 (Y-axis)") - - # Convert from mm and mm/s and mm/sĀ² to encoder counts - step_size_counts = int(step_size * self.ENCODER_COUNTS_PER_MM) - min_vel_counts = int(min_velocity * self.VELOCITY_SCALING) - accel_counts = int(acceleration * self.ACCELERATION_SCALING) - max_vel_counts = int(max_velocity * self.VELOCITY_SCALING) - - # Build the 22-byte data payload - data = struct.pack(' Optional[dict]: - """ - Get jog parameters for an axis. - - Args: - dest: Destination address (0x21 for X-axis, 0x22 for Y-axis) - timeout: Timeout in seconds - - Returns: - Dictionary with keys: 'jog_mode', 'step_size' (mm), 'min_velocity' (mm/s), - 'acceleration' (mm/sĀ²), 'max_velocity' (mm/s), 'stop_mode', - or None if timeout - - Raises: - ValueError: If dest is invalid - """ - if dest not in (self.DEST_X_AXIS, self.DEST_Y_AXIS): - raise ValueError(f"Invalid destination: 0x{dest:02X}. Must be 0x21 (X-axis) or 0x22 (Y-axis)") - - # Create a unique waiter key combining message ID and expected source - waiter_key = (MsgId.MOT_GET_JOGPARAMS, dest) - event = threading.Event() - responses = [] - - with self._waiter_lock: - self._waiters[waiter_key] = (event, responses) - - try: - # Send REQ message to the axis - self.send_command( - MsgId.MOT_REQ_JOGPARAMS, - param1=0x01, - param2=0x00, - dest=dest, - source=0x01 - ) - - # Wait for GET response - if event.wait(timeout=timeout): - response = responses[0] if responses else None - if response is None: - return None - - # Parse the 28-byte message - if len(response.raw) >= 28: - channel_id = struct.unpack(' None: - """ - Configure trigger output for an axis. - - Args: - dest: Destination address (0x21 for X-axis, 0x22 for Y-axis) - trigger_mode: Trigger mode from TriggerMode enum - polarity: Trigger polarity (0x01 = active high, 0x02 = active low) - - Raises: - ValueError: If dest is invalid - """ - if dest not in (self.DEST_X_AXIS, self.DEST_Y_AXIS): - raise ValueError(f"Invalid destination: 0x{dest:02X}. Must be 0x21 (X-axis) or 0x22 (Y-axis)") - - # Build 14-byte trigger configuration packet - data = struct.pack(' None: - """ - Set relative move distance for an axis. - - Args: - dest: Destination address (0x21 for X-axis, 0x22 for Y-axis) - relative_distance: Relative distance to move in mm (positive or negative) - - Raises: - ValueError: If dest is invalid - """ - if dest not in (self.DEST_X_AXIS, self.DEST_Y_AXIS): - raise ValueError(f"Invalid destination: 0x{dest:02X}. Must be 0x21 (X-axis) or 0x22 (Y-axis)") - - # Convert from mm to encoder counts (signed) - relative_distance_counts = int(relative_distance * self.ENCODER_COUNTS_PER_MM) - - # Build the 6-byte data payload - data = struct.pack(' Optional[dict]: - """ - Get relative move parameters for an axis. - - Args: - dest: Destination address (0x21 for X-axis, 0x22 for Y-axis) - timeout: Timeout in seconds - - Returns: - Dictionary with keys: 'relative_distance' (mm), 'relative_distance_counts', - or None if timeout - - Raises: - ValueError: If dest is invalid - """ - if dest not in (self.DEST_X_AXIS, self.DEST_Y_AXIS): - raise ValueError(f"Invalid destination: 0x{dest:02X}. Must be 0x21 (X-axis) or 0x22 (Y-axis)") - - # Create a unique waiter key combining message ID and expected source - waiter_key = (MsgId.MOT_GET_MOVERELPARAMS, dest) - event = threading.Event() - responses = [] - - with self._waiter_lock: - self._waiters[waiter_key] = (event, responses) - - try: - # Send REQ message to the axis - self.send_command( - MsgId.MOT_REQ_MOVERELPARAMS, - param1=0x01, - param2=0x00, - dest=dest, - source=0x01 - ) - - # Wait for GET response - if event.wait(timeout=timeout): - response = responses[0] if responses else None - if response is None: - return None - - # Parse the 12-byte message - if len(response.raw) >= 12: - channel_id = struct.unpack(' None: - """ - Set absolute move position for an axis. - - Args: - dest: Destination address (0x21 for X-axis, 0x22 for Y-axis) - absolute_position: Absolute position to move to in mm - - Raises: - ValueError: If dest is invalid - """ - if dest not in (self.DEST_X_AXIS, self.DEST_Y_AXIS): - raise ValueError(f"Invalid destination: 0x{dest:02X}. Must be 0x21 (X-axis) or 0x22 (Y-axis)") - - # Convert from mm to encoder counts (signed) - absolute_position_counts = int(absolute_position * self.ENCODER_COUNTS_PER_MM) - - # Build the 6-byte data payload - data = struct.pack(' Optional[dict]: - """ - Get absolute move parameters for an axis. - - Args: - dest: Destination address (0x21 for X-axis, 0x22 for Y-axis) - timeout: Timeout in seconds - - Returns: - Dictionary with keys: 'absolute_position' (mm), 'absolute_position_counts', - or None if timeout - - Raises: - ValueError: If dest is invalid - """ - if dest not in (self.DEST_X_AXIS, self.DEST_Y_AXIS): - raise ValueError(f"Invalid destination: 0x{dest:02X}. Must be 0x21 (X-axis) or 0x22 (Y-axis)") - - # Create a unique waiter key combining message ID and expected source - waiter_key = (MsgId.MOT_GET_MOVEABSPARAMS, dest) - event = threading.Event() - responses = [] - - with self._waiter_lock: - self._waiters[waiter_key] = (event, responses) - - try: - # Send REQ message to the axis - self.send_command( - MsgId.MOT_REQ_MOVEABSPARAMS, - param1=0x01, - param2=0x00, - dest=dest, - source=0x01 - ) - - # Wait for GET response - if event.wait(timeout=timeout): - response = responses[0] if responses else None - if response is None: - return None - - # Parse the 12-byte message - if len(response.raw) >= 12: - channel_id = struct.unpack(' Optional[dict]: - """ - Execute a relative move using the previously set relative move parameters. - - The controller will not respond until the move is completed. This command - uses the relative distance that was previously set with set_move_rel_params(). - - Args: - dest: Destination address (0x21 for X-axis, 0x22 for Y-axis) - timeout: Timeout in seconds (default 30s to account for move time) - - Returns: - Dictionary with move completion data: 'position' (mm), 'position_counts', - 'velocity', 'motor_current', 'status_bits', or None if timeout - - Raises: - ValueError: If dest is invalid - """ - if dest not in (self.DEST_X_AXIS, self.DEST_Y_AXIS): - raise ValueError(f"Invalid destination: 0x{dest:02X}. Must be 0x21 (X-axis) or 0x22 (Y-axis)") - - # Create a unique waiter key combining message ID and expected source - waiter_key = (MsgId.MOT_MOVE_COMPLETED, dest) - event = threading.Event() - responses = [] - - with self._waiter_lock: - self._waiters[waiter_key] = (event, responses) - - try: - # Send MOVE_RELATIVE command (header only) - self.send_command( - MsgId.MOT_MOVE_RELATIVE, - param1=0x01, - param2=0x00, - dest=dest, - source=0x01 - ) - - print(f"Executing relative move on axis 0x{dest:02X}... (this may take some time)") - - # Wait for MOVE_COMPLETED response from device - if event.wait(timeout=timeout): - response = responses[0] if responses else None - if response is None: - return None - # The move completion data should already be stored in _move_completed by _parse_buffer - with self._move_completed_lock: - return self._move_completed.get(dest) - else: - print(f"Timeout waiting for move completion from axis 0x{dest:02X}") - return None - finally: - with self._waiter_lock: - if waiter_key in self._waiters: - del self._waiters[waiter_key] - - def move_absolute(self, dest: int, timeout: float = 30.0) -> Optional[dict]: - """ - Execute an absolute move using the previously set absolute move parameters. - - The controller will not respond until the move is completed. This command - uses the absolute position that was previously set with set_move_abs_params(). - - Args: - dest: Destination address (0x21 for X-axis, 0x22 for Y-axis) - timeout: Timeout in seconds (default 30s to account for move time) - - Returns: - Dictionary with move completion data: 'position' (mm), 'position_counts', - 'velocity', 'motor_current', 'status_bits', or None if timeout - - Raises: - ValueError: If dest is invalid - """ - if dest not in (self.DEST_X_AXIS, self.DEST_Y_AXIS): - raise ValueError(f"Invalid destination: 0x{dest:02X}. Must be 0x21 (X-axis) or 0x22 (Y-axis)") - - # Create a unique waiter key combining message ID and expected source - waiter_key = (MsgId.MOT_MOVE_COMPLETED, dest) - event = threading.Event() - responses = [] - - with self._waiter_lock: - self._waiters[waiter_key] = (event, responses) - - try: - # Send MOVE_ABSOLUTE command (header only) - self.send_command( - MsgId.MOT_MOVE_ABSOLUTE, - param1=0x01, - param2=0x00, - dest=dest, - source=0x01 - ) - - print(f"Executing absolute move on axis 0x{dest:02X}... (this may take some time)") - - # Wait for MOVE_COMPLETED response from device - if event.wait(timeout=timeout): - response = responses[0] if responses else None - if response is None: - return None - # The move completion data should already be stored in _move_completed by _parse_buffer - with self._move_completed_lock: - return self._move_completed.get(dest) - else: - print(f"Timeout waiting for move completion from axis 0x{dest:02X}") - return None - finally: - with self._waiter_lock: - if waiter_key in self._waiters: - del self._waiters[waiter_key] - - def stop_move(self, dest: int, stop_mode: int = StopMode.CONTROLLED, - wait_for_stopped: bool = True, timeout: float = 5.0) -> Optional[dict]: - """ - Stop a moving axis immediately or with controlled deceleration. - - Args: - dest: Destination address (0x21 for X-axis, 0x22 for Y-axis) - stop_mode: StopMode.IMMEDIATE (1) for instant stop, - StopMode.CONTROLLED (2) for controlled deceleration (default) - wait_for_stopped: If True, wait for MOVE_STOPPED message (default True) - timeout: Timeout in seconds when waiting for MOVE_STOPPED (default 5s) - - Returns: - If wait_for_stopped=True: Dictionary with stopped status data, or None if timeout - If wait_for_stopped=False: None immediately after sending command - - Raises: - ValueError: If dest or stop_mode is invalid - """ - if dest not in (self.DEST_X_AXIS, self.DEST_Y_AXIS): - raise ValueError(f"Invalid destination: 0x{dest:02X}. Must be 0x21 (X-axis) or 0x22 (Y-axis)") - - if stop_mode not in (StopMode.IMMEDIATE, StopMode.CONTROLLED): - raise ValueError(f"Invalid stop_mode: {stop_mode}. Must be StopMode.IMMEDIATE (1) or StopMode.CONTROLLED (2)") - - if wait_for_stopped: - # Create a unique waiter key combining message ID and expected source - waiter_key = (MsgId.MOT_MOVE_STOPPED, dest) - event = threading.Event() - responses = [] - - with self._waiter_lock: - self._waiters[waiter_key] = (event, responses) - - try: - # Send MOVE_STOP command (header only) - self.send_command( - MsgId.MOT_MOVE_STOP, - param1=0x01, # Channel ID (always 1) - param2=stop_mode, # Stop mode - dest=dest, - source=0x01 - ) - - print(f"Stopping axis 0x{dest:02X} with {'immediate' if stop_mode == StopMode.IMMEDIATE else 'controlled'} stop...") - - # Wait for MOVE_STOPPED response from device - if event.wait(timeout=timeout): - response = responses[0] if responses else None - if response is None: - return None - # The move stopped data should already be stored in _move_stopped by _parse_buffer - with self._move_stopped_lock: - return self._move_stopped.get(dest) - else: - print(f"Timeout waiting for stop confirmation from axis 0x{dest:02X}") - return None - finally: - with self._waiter_lock: - if waiter_key in self._waiters: - del self._waiters[waiter_key] - else: - # Just send the command without waiting - self.send_command( - MsgId.MOT_MOVE_STOP, - param1=0x01, # Channel ID (always 1) - param2=stop_mode, # Stop mode - dest=dest, - source=0x01 - ) - return None - - def stop_x_axis(self, stop_mode: int = StopMode.CONTROLLED, - wait_for_stopped: bool = True, timeout: float = 5.0) -> Optional[dict]: - """ - Stop the X-axis. - - Args: - stop_mode: StopMode.IMMEDIATE (1) or StopMode.CONTROLLED (2, default) - wait_for_stopped: If True, wait for MOVE_STOPPED message (default True) - timeout: Timeout in seconds when waiting (default 5s) - - Returns: - Dictionary with stopped status data if wait_for_stopped=True, None otherwise - """ - return self.stop_move(self.DEST_X_AXIS, stop_mode, wait_for_stopped, timeout) - - def stop_y_axis(self, stop_mode: int = StopMode.CONTROLLED, - wait_for_stopped: bool = True, timeout: float = 5.0) -> Optional[dict]: - """ - Stop the Y-axis. - - Args: - stop_mode: StopMode.IMMEDIATE (1) or StopMode.CONTROLLED (2, default) - wait_for_stopped: If True, wait for MOVE_STOPPED message (default True) - timeout: Timeout in seconds when waiting (default 5s) - - Returns: - Dictionary with stopped status data if wait_for_stopped=True, None otherwise - """ - return self.stop_move(self.DEST_Y_AXIS, stop_mode, wait_for_stopped, timeout) - - def stop_all_axes(self, stop_mode: int = StopMode.CONTROLLED, - wait_for_stopped: bool = True, timeout: float = 5.0) -> dict: - """ - Stop all axes simultaneously. - - Args: - stop_mode: StopMode.IMMEDIATE (1) or StopMode.CONTROLLED (2, default) - wait_for_stopped: If True, wait for MOVE_STOPPED messages (default True) - timeout: Timeout in seconds when waiting (default 5s) - - Returns: - Dictionary with keys 'x' and 'y' containing stopped status data or None - """ - if wait_for_stopped: - # Use threading to stop both axes simultaneously - x_result = [None] - y_result = [None] - - def stop_x(): - x_result[0] = self.stop_move(self.DEST_X_AXIS, stop_mode, True, timeout) - - def stop_y(): - y_result[0] = self.stop_move(self.DEST_Y_AXIS, stop_mode, True, timeout) - - x_thread = threading.Thread(target=stop_x) - y_thread = threading.Thread(target=stop_y) - - x_thread.start() - y_thread.start() - - x_thread.join() - y_thread.join() - - return {'x': x_result[0], 'y': y_result[0]} - else: - # Just send commands without waiting - self.stop_move(self.DEST_X_AXIS, stop_mode, False) - self.stop_move(self.DEST_Y_AXIS, stop_mode, False) - return {'x': None, 'y': None} - - # High-level motion control methods - - def poll_positions(self) -> None: - """ - Request position updates for both X and Y axes. - - This actively queries the controller for the current positions of both axes. - The positions are stored internally and can be accessed via position_x and position_y properties. - """ - try: - self.get_position(self.DEST_X_AXIS, timeout=1.5) - except Exception as e: - print(f"Error polling X position: {e}") - - try: - self.get_position(self.DEST_Y_AXIS, timeout=1.5) - except Exception as e: - print(f"Error polling Y position: {e}") - - def move_to_fast(self, x: Optional[float] = None, y: Optional[float] = None) -> None: - """ - Move to absolute position(s) using fast non-blocking moves. - - Args: - x: Target X position in mm (None to leave X unchanged) - y: Target Y position in mm (None to leave Y unchanged) - """ - # Track pending moves - with self._pending_moves_lock: - if x is not None: - self._pending_moves[self.DEST_X_AXIS] = x - if y is not None: - self._pending_moves[self.DEST_Y_AXIS] = y - - if y is not None: - y_counts = int(y * self.ENCODER_COUNTS_PER_MM) - y_data = struct.pack(' bool: - """ - Move to position and wait for MOT_MOVE_COMPLETED message(s). - - This method sends absolute move commands and polls for MOT_MOVE_COMPLETED - messages by checking the _move_completed dictionary which is updated by - the RX thread when completion messages arrive. - - Args: - x: Target X position in mm, or None to skip X axis - y: Target Y position in mm, or None to skip Y axis - timeout: Maximum wait time in seconds (default 30s) - - Returns: - True if all moves completed successfully, False if timeout - """ - if x is None and y is None: - return True # Nothing to do - - # Determine which axes we're moving - axes_to_move = [] - if x is not None: - axes_to_move.append(self.DEST_X_AXIS) - if y is not None: - axes_to_move.append(self.DEST_Y_AXIS) - - # Clear any previous completion data for axes we're about to move - with self._move_completed_lock: - for dest in axes_to_move: - if dest in self._move_completed: - del self._move_completed[dest] - - # Track pending moves - with self._pending_moves_lock: - if x is not None: - self._pending_moves[self.DEST_X_AXIS] = x - if y is not None: - self._pending_moves[self.DEST_Y_AXIS] = y - - move_type = "X+Y" if (x is not None and y is not None) else "X-only" if x is not None else "Y-only" - print(f" [move_and_wait] {move_type}: x={x}, y={y}") - - # Send Y move first if needed - if y is not None: - y_counts = int(y * self.ENCODER_COUNTS_PER_MM) - y_data = struct.pack(' timeout: - missing = [("X" if d == self.DEST_X_AXIS else "Y") for d in axes_to_move if d not in completed_axes] - print(f" [move_and_wait] Timeout waiting for {', '.join(missing)} MOT_MOVE_COMPLETED") - return False - - # Check for completions - with self._move_completed_lock: - for dest in axes_to_move: - if dest not in completed_axes and dest in self._move_completed: - axis_name = "X" if dest == self.DEST_X_AXIS else "Y" - print(f" [move_and_wait] {axis_name}-axis move completed") - completed_axes.add(dest) - # Clear pending move - with self._pending_moves_lock: - if dest in self._pending_moves: - del self._pending_moves[dest] - - if len(completed_axes) < len(axes_to_move): - time.sleep(0.010) # 10ms poll interval - - return True - - def poll_until_idle(self, tolerance: float = 0.005, timeout: float = 0.1) -> bool: - """ - Poll positions and check if all pending moves are complete within tolerance. - - This method checks if the stage has settled at the target positions for all pending moves. - A move is considered complete when the current position is within 'tolerance' of the target. - - Args: - tolerance: Position tolerance in mm (default 0.005mm = 5 microns) - timeout: Time to spend polling in this call (default 0.1s) - - Returns: - True if all pending moves are complete (within tolerance), False otherwise - """ - # Get current pending targets (snapshot) - with self._pending_moves_lock: - pending = dict(self._pending_moves) - - if not pending: - # No pending moves - return True - - # Check if all axes are within tolerance - all_settled = True - completed_axes = [] - - # Query positions using get_position which properly waits for responses - # Use short timeout to avoid blocking too long - query_timeout = min(timeout, 0.5) - - for dest, target in pending.items(): - try: - # Get current position with timeout - current = self.get_position(dest, timeout=query_timeout) - - if current is None: - # Position query failed or timed out - not settled - all_settled = False - continue - - error = abs(current - target) - - if error <= tolerance: - # This axis is settled - completed_axes.append(dest) - else: - # Still moving - all_settled = False - - except Exception as e: - # Position query failed - assume not settled - print(f"Error querying position for axis 0x{dest:02X}: {e}") - all_settled = False - - # Remove completed axes from pending list - if completed_axes: - with self._pending_moves_lock: - for dest in completed_axes: - if dest in self._pending_moves: - del self._pending_moves[dest] - - return all_settled - - def poll_until_idle_passive(self, tolerance: float = 0.005, debug: bool = False) -> bool: - """ - Check if all pending moves are complete using cached positions from automatic updates. - - Unlike poll_until_idle(), this does NOT send position request commands. It only - reads the cached positions that are updated from automatic status updates - (MOT_GET_USTATUSUPDATE). This avoids command traffic that might interfere with - move execution. - - Args: - tolerance: Position tolerance in mm (default 0.005mm = 5 microns) - debug: If True, print debug info about positions - - Returns: - True if all pending moves are complete (within tolerance), False otherwise - """ - # Get current pending targets (snapshot) - with self._pending_moves_lock: - pending = dict(self._pending_moves) - - if not pending: - return True - - # Check if all axes are within tolerance using cached positions - all_settled = True - completed_axes = [] - - for dest, target in pending.items(): - # Get cached position (updated by automatic status updates) - with self._position_lock: - current = self._stage_positions.get(dest) - - axis_name = "X" if dest == self.DEST_X_AXIS else "Y" - - if current is None: - # No cached position yet - not settled - if debug: - print(f" [poll] {axis_name}: no cached position") - all_settled = False - continue - - error = abs(current - target) - - if debug: - print(f" [poll] {axis_name}: current={current:.3f}, target={target:.3f}, error={error*1000:.1f}um") - - if error <= tolerance: - completed_axes.append(dest) - else: - all_settled = False - - # Remove completed axes from pending list - if completed_axes: - with self._pending_moves_lock: - for dest in completed_axes: - if dest in self._pending_moves: - del self._pending_moves[dest] - - return all_settled - - def clear_pending_moves(self) -> None: - """ - Clear all pending move targets. - - This clears the internal tracking of pending moves. It does NOT stop the motors - - use stop_all_axes() if you want to halt motion. - """ - with self._pending_moves_lock: - self._pending_moves.clear() - - # Status bit decoding methods - - @staticmethod - def decode_status_bits(status_bits: int) -> MotorStatusBits: - """ - Decode status bits into a MotorStatusBits IntFlag. - - Args: - status_bits: Raw 32-bit status bits value - - Returns: - MotorStatusBits IntFlag with all active bits set - """ - return MotorStatusBits(status_bits) - - @staticmethod - def get_status_description(status_bits: int) -> str: - """ - Get a human-readable description of all active status bits. - - Args: - status_bits: Raw 32-bit status bits value - - Returns: - Multi-line string describing all active status bits - """ - status = MotorStatusBits(status_bits) - descriptions = [] - - # Error conditions (highest priority) - if MotorStatusBits.COMMUTATIONERROR in status: - descriptions.append("ERROR: Motor commutation error - power cycle required") - if MotorStatusBits.OVERTEMP in status: - descriptions.append("ERROR: Overtemperature detected") - if MotorStatusBits.BUSVOLTFAULT in status: - descriptions.append("ERROR: Supply voltage too low") - if MotorStatusBits.OVERLOAD in status: - descriptions.append("ERROR: Motor overload/overcurrent") - if MotorStatusBits.ENCODERFAULT in status: - descriptions.append("ERROR: Encoder fault") - if MotorStatusBits.OVERCURRENT in status: - descriptions.append("ERROR: Continuous current limit exceeded") - if MotorStatusBits.ERROR in status: - descriptions.append("ERROR: Other error condition") - - # Limit switches - if MotorStatusBits.CWHARDLIMIT in status: - descriptions.append("WARNING: Clockwise hard limit triggered") - if MotorStatusBits.CCWHARDLIMIT in status: - descriptions.append("WARNING: Counter-clockwise hard limit triggered") - if MotorStatusBits.CWSOFTLIMIT in status: - descriptions.append("WARNING: Clockwise software limit triggered") - if MotorStatusBits.CCWSOFTLIMIT in status: - descriptions.append("WARNING: Counter-clockwise software limit triggered") - if MotorStatusBits.POSITIONERROR in status: - descriptions.append("WARNING: Position error - outside tracking window") - - # Motion state - if MotorStatusBits.HOMING in status: - descriptions.append("STATUS: Motor is homing") - elif MotorStatusBits.INMOTIONCW in status: - descriptions.append("STATUS: Moving clockwise") - elif MotorStatusBits.INMOTIONCCW in status: - descriptions.append("STATUS: Moving counter-clockwise") - elif MotorStatusBits.ACTIVE in status: - descriptions.append("STATUS: Executing motion command") - elif MotorStatusBits.SETTLED in status: - descriptions.append("STATUS: Settled at target position") - - # Operational state - if MotorStatusBits.INITIALIZING in status: - descriptions.append("STATUS: Performing phase initialization") - if MotorStatusBits.HOMED in status: - descriptions.append("STATUS: Homed (position count valid)") - if MotorStatusBits.TRACKING in status: - descriptions.append("STATUS: Position within tracking window") - if MotorStatusBits.CONNECTED in status: - descriptions.append("STATUS: Motor recognized by controller") - if MotorStatusBits.ENABLED in status: - descriptions.append("STATUS: Motor output enabled") - if MotorStatusBits.POWEROK in status: - descriptions.append("STATUS: Power supply OK") - - if not descriptions: - descriptions.append("STATUS: No status bits set") - - return "\n".join(descriptions) - - @staticmethod - def has_errors(status_bits: int) -> bool: - """ - Check if any error conditions are present in status bits. - - Args: - status_bits: Raw 32-bit status bits value - - Returns: - True if any error bits are set, False otherwise - """ - status = MotorStatusBits(status_bits) - error_flags = ( - MotorStatusBits.OVERTEMP | - MotorStatusBits.BUSVOLTFAULT | - MotorStatusBits.COMMUTATIONERROR | - MotorStatusBits.OVERLOAD | - MotorStatusBits.ENCODERFAULT | - MotorStatusBits.OVERCURRENT | - MotorStatusBits.ERROR - ) - return bool(status & error_flags) - - @staticmethod - def is_in_motion(status_bits: int) -> bool: - """ - Check if motor is currently in motion. - - Args: - status_bits: Raw 32-bit status bits value - - Returns: - True if motor is moving, False otherwise - """ - status = MotorStatusBits(status_bits) - motion_flags = ( - MotorStatusBits.INMOTIONCW | - MotorStatusBits.INMOTIONCCW | - MotorStatusBits.HOMING | - MotorStatusBits.ACTIVE - ) - return bool(status & motion_flags) - - @staticmethod - def is_settled(status_bits: int) -> bool: - """ - Check if motor is settled at target position. - - Args: - status_bits: Raw 32-bit status bits value - - Returns: - True if motor is settled, False otherwise - """ - status = MotorStatusBits(status_bits) - return MotorStatusBits.SETTLED in status - - def __enter__(self): - self.connect() - return self - - def __exit__(self, exc_type, exc_val, exc_tb): - self.disconnect() - return False - - -if __name__ == '__main__': - # Example: monitor status updates and control channel enable state - def on_status_update(msg: AptMessage): - print(f"Status update: msg_id=0x{msg.msg_id:04X}, data={msg.data.hex() if msg.data else 'none'}") - - def on_move_stopped(msg: AptMessage): - """Callback for unexpected motor stop events.""" - print(f"\n!!! MOVE STOPPED EVENT DETECTED !!!") - print(f" Source: 0x{msg.source:02X} ({'X-axis' if msg.source == 0x21 else 'Y-axis' if msg.source == 0x22 else 'Unknown'})") - # The status will already be parsed and stored in _move_stopped - # Note: You can access mc.move_stopped_x or mc.move_stopped_y properties - # after this callback to get the full stopped status including status bits - - with MotionController() as mc: - # Register callbacks - mc.register_callback(MsgId.MOT_GET_DCSTATUSUPDATE, on_status_update) - mc.register_callback(MsgId.MOT_MOVE_STOPPED, on_move_stopped) - - # Print hardware info - print("Hardware Info:") - print(f" Serial Number: {mc.get_serial_number()}") - print(f" Model: {mc.get_model()}") - print(f" Firmware: {mc.get_firmware_version()}") - print(f" Channels: {mc.get_num_channels()}") - - print("\n=== Homing Test ===") - print("Homing X-axis...") - try: - print(" Enabling X-axis channel...") - mc.set_channel_enable_state(mc.DEST_X_AXIS, enabled=True) - time.sleep(0.5) - - if mc.home_x_axis(timeout=20.0): - print(f" X-axis homed successfully") - print(f" Position: {mc.position_x} mm (encoder count: {mc.encoder_count_x})") - else: - print(f" X-axis homing timed out") - except Exception as e: - print(f" Error homing X-axis: {e}") - - print("\nHoming Y-axis...") - try: - print(" Enabling Y-axis channel...") - mc.set_channel_enable_state(mc.DEST_Y_AXIS, enabled=True) - time.sleep(0.5) - - if mc.home_y_axis(timeout=20.0): - print(f" Y-axis homed successfully") - print(f" Position: {mc.position_y} mm (encoder count: {mc.encoder_count_y})") - else: - print(f" Y-axis homing timed out") - except Exception as e: - print(f" Error homing Y-axis: {e}") - - print("\n=== Jog Parameters Test ===") - print("Getting current jog parameters for X-axis...") - try: - jog_params = mc.get_jog_params(mc.DEST_X_AXIS, timeout=2.0) - if jog_params: - print(f" Jog mode: {jog_params['jog_mode']} ({'Continuous' if jog_params['jog_mode'] == JogMode.CONTINUOUS else 'Single Step'})") - print(f" Step size: {jog_params['step_size']:.3f} mm ({jog_params['step_size_counts']} counts)") - print(f" Min velocity: {jog_params['min_velocity']:.3f} mm/s") - print(f" Acceleration: {jog_params['acceleration']:.3f} mm/sĀ²") - print(f" Max velocity: {jog_params['max_velocity']:.3f} mm/s") - print(f" Stop mode: {jog_params['stop_mode']} ({'Immediate' if jog_params['stop_mode'] == StopMode.IMMEDIATE else 'Controlled'})") - else: - print(f" Timeout getting jog parameters") - except Exception as e: - print(f" Error getting jog parameters: {e}") - - print("\nUsing jog parameter properties:") - print(f" X-axis jog params: {mc.jog_params_x}") - - print("\nSetting new jog parameters for X-axis...") - try: - # Set: single step mode, 1mm step, min_vel=0.5 mm/s, accel=50 mm/sĀ², max_vel=10 mm/s, controlled stop - mc.set_jog_params( - mc.DEST_X_AXIS, - jog_mode=JogMode.SINGLE_STEP, - step_size=1.0, - min_velocity=0.5, - acceleration=50.0, - max_velocity=10.0, - stop_mode=StopMode.CONTROLLED - ) - time.sleep(0.2) - - # Read back the parameters - jog_params = mc.get_jog_params(mc.DEST_X_AXIS, timeout=2.0) - if jog_params: - print(f" New jog mode: {jog_params['jog_mode']} ({'Continuous' if jog_params['jog_mode'] == JogMode.CONTINUOUS else 'Single Step'})") - print(f" New step size: {jog_params['step_size']:.3f} mm") - print(f" New min velocity: {jog_params['min_velocity']:.3f} mm/s") - print(f" New acceleration: {jog_params['acceleration']:.3f} mm/sĀ²") - print(f" New max velocity: {jog_params['max_velocity']:.3f} mm/s") - print(f" New stop mode: {jog_params['stop_mode']} ({'Immediate' if jog_params['stop_mode'] == StopMode.IMMEDIATE else 'Controlled'})") - else: - print(f" Timeout reading back jog parameters") - - # Show cached properties updated - print(f"\n Cached properties after update:") - print(f" X jog params: {mc.jog_params_x}") - except Exception as e: - print(f" Error setting jog parameters: {e}") - - print("\n=== Move Relative Parameters Test ===") - print("Getting current move relative parameters for X-axis...") - try: - move_rel_params = mc.get_move_rel_params(mc.DEST_X_AXIS, timeout=2.0) - if move_rel_params: - print(f" Relative distance: {move_rel_params['relative_distance']:.3f} mm ({move_rel_params['relative_distance_counts']} counts)") - else: - print(f" Timeout getting move relative parameters") - except Exception as e: - print(f" Error getting move relative parameters: {e}") - - print("\nUsing move relative parameter properties:") - print(f" X-axis move rel params: {mc.move_rel_params_x}") - print(f" X relative distance: {mc.relative_distance_x} mm") - - print("\nSetting new move relative parameters for X-axis...") - try: - # Set: relative distance of 5mm - mc.set_move_rel_params(mc.DEST_X_AXIS, relative_distance=5.0) - time.sleep(0.2) - - # Read back the parameters - move_rel_params = mc.get_move_rel_params(mc.DEST_X_AXIS, timeout=2.0) - if move_rel_params: - print(f" New relative distance: {move_rel_params['relative_distance']:.3f} mm ({move_rel_params['relative_distance_counts']} counts)") - else: - print(f" Timeout reading back move relative parameters") - - # Show cached properties updated - print(f"\n Cached properties after update:") - print(f" X relative distance: {mc.relative_distance_x} mm") - except Exception as e: - print(f" Error setting move relative parameters: {e}") - - print("\nTesting negative relative move...") - try: - # Set: relative distance of -2.5mm - mc.set_move_rel_params(mc.DEST_X_AXIS, relative_distance=-2.5) - time.sleep(0.2) - - # Read back the parameters - move_rel_params = mc.get_move_rel_params(mc.DEST_X_AXIS, timeout=2.0) - if move_rel_params: - print(f" New relative distance: {move_rel_params['relative_distance']:.3f} mm ({move_rel_params['relative_distance_counts']} counts)") - print(f" Cached X relative distance: {mc.relative_distance_x} mm") - else: - print(f" Timeout reading back move relative parameters") - except Exception as e: - print(f" Error setting negative move relative parameters: {e}") - - print("\n=== Move Absolute Parameters Test ===") - print("Getting current move absolute parameters for X-axis...") - try: - move_abs_params = mc.get_move_abs_params(mc.DEST_X_AXIS, timeout=2.0) - if move_abs_params: - print(f" Absolute position: {move_abs_params['absolute_position']:.3f} mm ({move_abs_params['absolute_position_counts']} counts)") - else: - print(f" Timeout getting move absolute parameters") - except Exception as e: - print(f" Error getting move absolute parameters: {e}") - - print("\nUsing move absolute parameter properties:") - print(f" X-axis move abs params: {mc.move_abs_params_x}") - print(f" X absolute position: {mc.absolute_position_x} mm") - - print("\nSetting new move absolute parameters for X-axis...") - try: - # Set: absolute position of 10mm - mc.set_move_abs_params(mc.DEST_X_AXIS, absolute_position=10.0) - time.sleep(0.2) - - # Read back the parameters - move_abs_params = mc.get_move_abs_params(mc.DEST_X_AXIS, timeout=2.0) - if move_abs_params: - print(f" New absolute position: {move_abs_params['absolute_position']:.3f} mm ({move_abs_params['absolute_position_counts']} counts)") - else: - print(f" Timeout reading back move absolute parameters") - - # Show cached properties updated - print(f"\n Cached properties after update:") - print(f" X absolute position: {mc.absolute_position_x} mm") - except Exception as e: - print(f" Error setting move absolute parameters: {e}") - - print("\nTesting different absolute position...") - try: - # Set: absolute position of 45mm (near the home position) - mc.set_move_abs_params(mc.DEST_X_AXIS, absolute_position=45.0) - time.sleep(0.2) - - # Read back the parameters - move_abs_params = mc.get_move_abs_params(mc.DEST_X_AXIS, timeout=2.0) - if move_abs_params: - print(f" New absolute position: {move_abs_params['absolute_position']:.3f} mm ({move_abs_params['absolute_position_counts']} counts)") - print(f" Cached X absolute position: {mc.absolute_position_x} mm") - else: - print(f" Timeout reading back move absolute parameters") - except Exception as e: - print(f" Error setting absolute position: {e}") - - print("\n=== Move Relative Execution Test ===") - print("Setting up relative move of 2mm for X-axis...") - try: - # First, set the relative move parameters - mc.set_move_rel_params(mc.DEST_X_AXIS, relative_distance=2.0) - time.sleep(0.2) - - # Verify parameters were set - rel_params = mc.get_move_rel_params(mc.DEST_X_AXIS, timeout=2.0) - if rel_params: - print(f" Relative move parameters set: {rel_params['relative_distance']:.3f} mm") - else: - print(f" Failed to verify relative move parameters") - - # Execute the relative move - print("\n Executing relative move...") - move_result = mc.move_relative(mc.DEST_X_AXIS, timeout=30.0) - - if move_result: - print(f" Move completed successfully!") - print(f" Final position: {move_result['position']:.3f} mm ({move_result['position_counts']} counts)") - print(f" Final velocity: {move_result['velocity']} encoder units") - print(f" Motor current: {move_result['motor_current']}") - print(f" Status bits: 0x{move_result['status_bits']:08X}") - - # Decode and display status bits - print(f"\n Status bit analysis:") - status_desc = MotionController.get_status_description(move_result['status_bits']) - for line in status_desc.split('\n'): - print(f" {line}") - - print(f"\n Status bit checks:") - print(f" Has errors: {MotionController.has_errors(move_result['status_bits'])}") - print(f" Is in motion: {MotionController.is_in_motion(move_result['status_bits'])}") - print(f" Is settled: {MotionController.is_settled(move_result['status_bits'])}") - - # Show cached property - print(f"\n Cached move completed status:") - print(f" X: {mc.move_completed_x}") - - # Demonstrate status bit properties - print(f"\n Status bit properties (automatically updated):") - print(f" Raw status bits: 0x{mc.status_bits_x:08X}" if mc.status_bits_x else " Raw status bits: Not available") - print(f" Is enabled: {mc.is_enabled_x}") - print(f" Is homed: {mc.is_homed_x}") - print(f" Is in motion: {mc.is_in_motion_x}") - print(f" Is settled: {mc.is_settled_x}") - print(f" Is tracking: {mc.is_tracking_x}") - print(f" Is connected: {mc.is_connected_x}") - print(f" Power OK: {mc.power_ok_x}") - print(f" Is active: {mc.is_active_x}") - print(f" Has errors: {mc.has_errors_x}") - print(f" At CW limit: {mc.at_cw_limit_x}") - print(f" At CCW limit: {mc.at_ccw_limit_x}") - else: - print(f" Move timed out or failed") - except Exception as e: - print(f" Error executing relative move: {e}") - - print("\n=== Status Bit Properties Summary ===") - print("X-axis status:") - if mc.status_bits_x: - print(f" Raw bits: 0x{mc.status_bits_x:08X}") - print(f" Enabled: {mc.is_enabled_x}, Homed: {mc.is_homed_x}, Connected: {mc.is_connected_x}") - print(f" In motion: {mc.is_in_motion_x}, Settled: {mc.is_settled_x}, Active: {mc.is_active_x}") - print(f" Has errors: {mc.has_errors_x}, Power OK: {mc.power_ok_x}") - else: - print(" No status available yet") - - print("\nY-axis status:") - if mc.status_bits_y: - print(f" Raw bits: 0x{mc.status_bits_y:08X}") - print(f" Enabled: {mc.is_enabled_y}, Homed: {mc.is_homed_y}, Connected: {mc.is_connected_y}") - print(f" In motion: {mc.is_in_motion_y}, Settled: {mc.is_settled_y}, Active: {mc.is_active_y}") - print(f" Has errors: {mc.has_errors_y}, Power OK: {mc.power_ok_y}") - else: - print(" No status available yet") - - print("\n=== Test Complete ===") - print("Waiting 1 second before exit to allow final messages...") - time.sleep(1.0) diff --git a/hardware/pybbd202/__init__.py b/hardware/pybbd202/__init__.py new file mode 100644 index 0000000..030d3b7 --- /dev/null +++ b/hardware/pybbd202/__init__.py @@ -0,0 +1,9 @@ +"""pybbd202 - Thorlabs BBD202 servo stage driver (pyserial-based)""" + +from .bbd20x import ThorlabsServoDriver +from .apt_constants import TriggerBitsServo, StatusBits + +# Axis address constants +AXIS_X = 0x21 +AXIS_Y = 0x22 +CONTROLLER = 0x11 diff --git a/hardware/pybbd202/apt_constants.py b/hardware/pybbd202/apt_constants.py new file mode 100644 index 0000000..882ad6b --- /dev/null +++ b/hardware/pybbd202/apt_constants.py @@ -0,0 +1,59 @@ +''' + APT Constants and Registries + Thomas Ales | Feb 2026 +''' +from enum import IntFlag + +class StatusBits(IntFlag): + MOT_SB_CWHARDLIMIT = 0x01 + MOT_SB_CCWHARDLIMIT = 0x02 + MOT_SB_INMOTIONCW = 0x10 + MOT_SB_INMOTIONCCW = 0x20 + MOT_SB_HOMING = 0x200 + MOT_SB_HOMED = 0x400 + MOT_SB_TRACKING = 0x1000 + MOT_SB_SETTLED = 0x2000 + MOT_SB_POSITIONERROR = 0x4000 + MOT_SB_INSTRERROR = 0x8000 + MOT_SB_INTERLOCK = 0x10000 + MOT_SB_OVERTEMP = 0x20000 + MOT_SB_BUSVOLTFAULT = 0x40000 + MOT_SB_COMMUTATIONERROR = 0x80000 + MOT_SB_DIGIP1 = 0x100000 + MOT_SB_OVERLOAD = 0x1000000 + MOT_SB_POWEROK = 0x10000000 + MOT_SB_ERROR = 0x40000000 + MOT_SB_ENABLED = 0x80000000 + + # combined masks for checking various + # states + MOT_ANY_MOVE = MOT_SB_INMOTIONCW | MOT_SB_INMOTIONCCW + MOT_ANY_ERR = (MOT_SB_OVERTEMP | MOT_SB_BUSVOLTFAULT | + MOT_SB_COMMUTATIONERROR | MOT_SB_OVERLOAD | + MOT_SB_ERROR | MOT_SB_INSTRERROR) + +class TriggerBitsStepper(IntFlag): + TRIGIN_ENABLE = 0x01, + TRIGOUT_ENABLE = 0x02, + TRIGOUT_MODEFOLLOW = 0x04, + TRIGOUT_MODEMOVEEND = 0x08, + TRIG_RELMOVE = 0x10, + TRIG_ABSMOVE = 0x20, + TRIG_HOMEMOVE = 0x40, + TRIGOUT_NOTRIGIN = 0x80 + +class TriggerBitsServo(IntFlag): + TRIGIN_HIGH = 0x01 + TRIGIN_RELMOVE = 0x02 + TRIGIN_ABSMOVE = 0x04 + TRIGIN_HOMEMOVE = 0x08 + TRIGOUT_HIGH = 0x10 + TRIGOUT_INMOTION = 0x20 + TRIGOUT_MOTIONCOMPLETE = 0x40 + TRIGOUT_MAXVELOCITY = 0x80 + + TRIGOUT_MAXV = TRIGOUT_HIGH | TRIGOUT_MAXVELOCITY + + + + diff --git a/hardware/pybbd202/apt_messages.py b/hardware/pybbd202/apt_messages.py new file mode 100644 index 0000000..a96704f --- /dev/null +++ b/hardware/pybbd202/apt_messages.py @@ -0,0 +1,318 @@ +''' + ThorLABS APT Protocol Message Registry + Thomas Ales | Feb 2026 + Version 1 +''' +import struct +from .apt_constants import StatusBits as sb + +class APTProtocol(): + ADDRESSES = { 'HOST_PC': 0x01, 'CONTROLLER': 0x11, + 'X_AXIS': 0x21, 'Y_AXIS': 0x22 + } + + # N.B.: If the format is anything other than the following + # two, it is considered a 'long' message. In this case + # the format key refers only to the payload part of the + # message. + # BB - Short Message, Only Source/Destination + # BBBB - Short Message, Using Parameters 1 & 2 + + # Dictionary entries MUST BE in the order they + # appear in the thorlabs documentation, if you don't + # the unpacking logic goes all to šŸ’© and 'fun' things + # happen. + + MSGS = { + 0x0002: { + 'name': 'MGMSG_HW_DISCONNECT', + 'format': 'BB', + 'response': None, + 'fields': ['destination', 'source'], + 'data_fields': None + }, + 0x0005: { + 'name': 'MGMSG_HW_REQ_INFO', + 'format': 'BB', + 'response': 0x0006, + 'fields': ['destination', 'source'], + 'data_fields': None + }, + 0x0006: { + 'name': 'MGMSG_HW_GET_INFO', + 'format': ' 0 else 0x00 + p2 = kwargs.get(param_fields[1], 0x00) if len(param_fields) > 1 else 0x00 + message = struct.pack(' 0: + data[param_fields[0]] = bdata[2] + if len(param_fields) > 1: + data[param_fields[1]] = bdata[3] + return msg_id, data + else: + raise ValueError(f"Message format {fmt_string} isn't defined/valid!") + + @classmethod + def get_name(cls, msg_id): + return cls.MSGS.get(msg_id, {}).get('name', f'UNKNOWN_{hex(msg_id)}') + diff --git a/hardware/pybbd202/bbd20x.py b/hardware/pybbd202/bbd20x.py new file mode 100644 index 0000000..0e49428 --- /dev/null +++ b/hardware/pybbd202/bbd20x.py @@ -0,0 +1,505 @@ +''' + BBD20X Stage Driver for SRAS + Thomas Ales | Feb 2026 + Version 2 +''' + +import time +from threading import Thread, Event +from queue import Queue, Empty +from .apt_constants import StatusBits, TriggerBitsServo +from .apt_messages import APTProtocol +from .serial_comms import SerialSnooper + + +class ThorlabsServoDriver(): + # These are specific to the MLS203-1 + # change for a different application + counts_per_mm = 20000 + accel_scaling = 13.744 + velocity_scaling = 134217.73 + + def __init__(self): + self.am_connected = False + self.am_enabled = [False, False] + self.am_homed = [False, False] + self.am_moving = [False, False] + self.am_error = [False, False] + self.serial_port = "/dev/ttyUSB1" + self.serial_spd = 115200 + self.max_velocities = [100.0, 100.0] # mm/s + self.max_accels = [500.0, 500.0] # mm/s2 + self.positions = [-1.0, -1.0] + self.act_velocities = [-1.0, -1.0] + self.current_demand = [0, 0] + self.serial_snoop = None + self.am_listening = False + self.pending_responses = {} # msg_id -> {'event': Event, 'data': None} + self.bays_present = [] # list of bay addresses that responded + self._tx_queue = Queue() + self._polling_active = False + self._poll_interval = 0.2 # seconds between poll cycles + + def connect(self, port=None, spd=None): + if port is None: + port = self.serial_port + else: + self.serial_port = port + + if spd is None: + spd = self.serial_spd + else: + self.serial_spd = spd + + self.serial_snoop = SerialSnooper(port, spd) + self.serial_snoop.start() + self.am_listening = True + + # Start worker threads + self._tx_thread = Thread(target=self._tx_worker, daemon=True) + self._tx_thread.start() + self._rx_thread = Thread(target=self._rx_worker, daemon=True) + self._rx_thread.start() + self._poll_thread = Thread(target=self._poll_worker, daemon=True) + self._poll_thread.start() + + # Give threads time to start + time.sleep(0.3) + + # Query which bays are present + self.bays_present = [] + for bay_id in range(10): # bays 0-9 + try: + result = self.send_and_wait(0x0060, timeout=0.5, + bay_id=bay_id, + destination=0x11, + source=0x01) + if result and result.get('bay_state') == 0x01: + bay_addr = 0x21 + bay_id + self.bays_present.append(bay_addr) + + except TimeoutError: + # Bay not present or not responding + pass + + if not self.bays_present: + print(" [WARN] No bays detected!") + + self.am_connected = True + + # ā”€ā”€ Worker threads ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ + + def _tx_worker(self): + '''Single serial writer. All outbound messages flow through + _tx_queue so there are no write races on the serial port.''' + while self.am_listening: + try: + msg = self._tx_queue.get(timeout=0.05) + self.serial_snoop.serial_connection.write(msg) + self.serial_snoop.serial_connection.flush() + except Empty: + continue + except (OSError, TypeError): + break + + def _rx_worker(self): + '''Listens for messages on the serial RX queue and dispatches them.''' + while self.am_listening: + try: + _msg = self.serial_snoop.rx_msg_queue.get(timeout=0.05) + + # Try to parse the message, skip if unknown + try: + _msgid, data = APTProtocol.unpack_message(_msg) + except ValueError: + print(f" [WARN] Unknown message: {_msg.hex()}") + continue + + # Check if someone is waiting for this response + if _msgid in self.pending_responses: + self.pending_responses[_msgid]['data'] = data + self.pending_responses[_msgid]['event'].set() + + # Status update ā†’ ACK via TX queue, then update state + if _msgid == 0x0491: + if self.am_listening: + _ack = APTProtocol.build_message(0x0492, source=0x01, + destination=_msg[5]) + self._tx_queue.put(_ack) + self._update0x491(data) + # Move completed ā†’ update state + elif _msgid == 0x0464: + self._update0x0464(data) + except Empty: + continue + + return + + def _poll_worker(self): + '''Periodically sends REQ_USTATUSUPDATE (0x0490) to each bay. + The 0x0491 responses are ACKed by _rx_worker through the + TX queue, which keeps the controller's comms watchdog alive.''' + while self.am_listening: + if self._polling_active: + for axis_addr in (self.bays_present or [0x21, 0x22]): + if not self.am_listening: + break + msg = APTProtocol.build_message(0x0490, source=0x01, + destination=axis_addr) + self._tx_queue.put(msg) + time.sleep(self._poll_interval) + else: + time.sleep(0.05) + + # ā”€ā”€ Polling control ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ + + def start_polling(self, interval=0.2): + '''Start periodic status polling (interval in seconds).''' + self._poll_interval = interval + self._polling_active = True + + def stop_polling(self): + '''Stop periodic status polling.''' + self._polling_active = False + + # ā”€ā”€ Core messaging ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ + + def send_and_wait(self, msg_id, timeout=10.0, retries=1, **kwargs): + """Send a message and wait for its expected response. + On timeout, drains serial buffer and retries up to `retries` times.""" + msg_spec = APTProtocol.MSGS.get(msg_id) + if not msg_spec: + raise ValueError(f"Unknown message: {hex(msg_id)}") + + expected = msg_spec.get('response') + msg = APTProtocol.build_message(msg_id, **kwargs) + + for attempt in range(1 + retries): + # Set up listener before sending + if expected: + evt = Event() + self.pending_responses[expected] = {'event': evt, 'data': None} + + self._tx_queue.put(msg) + + if not expected: + return None # no response expected + + # Wait for response + if evt.wait(timeout=timeout): + data = self.pending_responses[expected]['data'] + del self.pending_responses[expected] + return data + else: + del self.pending_responses[expected] + if attempt < retries: + # Drain serial input buffer and message queue, then retry + self.serial_snoop.serial_connection.reset_input_buffer() + time.sleep(0.05) + while not self.serial_snoop.rx_msg_queue.empty(): + try: + self.serial_snoop.rx_msg_queue.get_nowait() + except Empty: + break + print(f" [RETRY] {APTProtocol.get_name(msg_id)} attempt {attempt+2}") + + raise TimeoutError(f"Timeout waiting for {hex(expected)}") + + def send_message(self, msg_id, **kwargs): + '''Build and queue a message for transmission (fire-and-forget).''' + msg = APTProtocol.build_message(msg_id, **kwargs) + self._tx_queue.put(msg) + + # ā”€ā”€ Connection management ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ + + def disconnect(self): + if (self.serial_snoop and + self.am_listening is True): + # Stop polling first + self._polling_active = False + # Queue disconnect messages for the TX worker to send + for addr in [0x11, 0x21, 0x22]: + self._tx_queue.put( + APTProtocol.build_message(0x0002, destination=addr, + source=0x01)) + time.sleep(0.2) # let TX worker flush them out + # Stop all worker loops, then wait for threads to exit + self.am_listening = False + self.serial_snoop.stop() + self._rx_thread.join() + self._tx_thread.join() + self._poll_thread.join() + self.serial_snoop.join() + # Close port only after all threads are done + self.serial_snoop.close() + + # ā”€ā”€ State update handlers ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ + + def _update0x491(self, msg): + ''' + _update0x491 - Internal function for handling USTATUSUPDATE + messages and updating the data for that particular axis sending + the message. + ''' + if msg['source'] == 0x21: + ch = 0 + elif msg['source'] == 0x22: + ch = 1 + else: + return + + self.positions[ch] = msg['position'] / self.counts_per_mm + self.act_velocities[ch] = msg['velocity'] / self.velocity_scaling + self.current_demand[ch] = msg['motor_current'] + + if(msg['status_bits'] & StatusBits.MOT_ANY_ERR): + self.am_error[ch] = True + else: + self.am_error[ch] = False + + if(msg['status_bits'] & StatusBits.MOT_ANY_MOVE): + self.am_moving[ch] = True + else: + self.am_moving[ch] = False + + if(msg['status_bits'] & StatusBits.MOT_SB_HOMED): + self.am_homed[ch] = True + + def _update0x0464(self, msg): + ''' + _update0x0464 - internal function for MOVE_COMPLETED messages. + Updates position, velocity, and moving state. + ''' + if msg['source'] == 0x21: + ch = 0 + elif msg['source'] == 0x22: + ch = 1 + else: + return + + self.positions[ch] = msg['position'] / self.counts_per_mm + self.act_velocities[ch] = msg['velocity'] / self.velocity_scaling + self.current_demand[ch] = msg['motor_current'] + self.am_moving[ch] = False + return + + def _update0x0212(self, msg): + ''' + _update0x0212 - internal function that listens for CHANENABLESTATE + messages. + ''' + if msg['source'] == 0x21: + ch = 0 + elif msg['source'] == 0x22: + ch = 1 + else: + raise ValueError("Wherever this message came from, it's WRONG!") + + if msg['enable_state'] == 0x01: + self.am_enabled[ch] = True # enabled + elif msg['enable_state'] == 0x02: + self.am_enabled[ch] = False # disabled + else: + raise ValueError("Am I a joke to you? WTF did this even come from?!") + + # ā”€ā”€ Axis control ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ + + def enable_axis(self, axis): + '''Enable the specified axis (0x21 = X, 0x22 = Y).''' + self.send_message(0x0210, chan_ident=1, enable_state=0x01, + destination=axis, source=0x01) + + def disable_axis(self, axis): + '''Disable the specified axis (0x21 = X, 0x22 = Y).''' + self.send_message(0x0210, chan_ident=1, enable_state=0x02, + destination=axis, source=0x01) + + def toggle_enabled_state(self, axis): + ''' + toggle_enabled_state(axis) - enables the axis if disabled. disables + if enabled. not much more to it. + ''' + if axis == 0x21: + ch = 0 + elif axis == 0x22: + ch = 1 + else: + raise ValueError("I don't know that axis!") + # get the old state and flip it like a sample + new_state = not self.am_enabled[ch] + self.send_message(0x0210, chan_ident=1, + enable_state=0x01 if new_state else 0x02, + destination=axis, source=0x01) + + def home_axis(self, axis, timeout=60.0): + ''' + home_axis(axis, timeout=60): Blocking home command. Required at + power up. Default timeout is 60s, but 20-30s is fine as well if + you're in that much of a hurry. + ''' + if axis == 0x21: + ch = 0 + elif axis == 0x22: + ch = 1 + else: + raise ValueError("I don't know that axis!") + + self.send_and_wait(0x0443, timeout=timeout, chan_ident=1, + destination=axis, source=0x01) + return + + def move_axis_relative(self, axis, distance_in_mm, timeout=10.0): + ''' + move_axis_relative(axis, distance_in_mm, timeout=10): + moves the specified axis a specified distance in mm. + Timeout defaults to ten seconds. + ''' + # sanity check + if axis == 0x21 and abs(distance_in_mm) > 110.0: + raise ValueError("You can't move farther than the stage is long.") + elif axis == 0x22 and abs(distance_in_mm) > 75.0: + raise ValueError("You can't move farther than the stage is wide.") + + _distance_in_encoder = int(round(distance_in_mm * self.counts_per_mm)) + self.send_and_wait(0x0448, timeout=timeout, chan_ident=1, + relative_distance=_distance_in_encoder, + destination=axis, source=0x01) + return + + def move_axis_absolute(self, axis, position_in_mm, timeout=10.0): + ''' + move_axis_absolute(axis, position_in_mm, timeout=10): + moves the specified axis to an absolute position in mm. + Timeout defaults to ten seconds. + ''' + if axis == 0x21 and (position_in_mm < 0.0 or position_in_mm > 110.0): + raise ValueError("Position out of range for X axis (0-110 mm).") + elif axis == 0x22 and (position_in_mm < 0.0 or position_in_mm > 75.0): + raise ValueError("Position out of range for Y axis (0-75 mm).") + + _position_in_encoder = int(round(position_in_mm * self.counts_per_mm)) + self.send_and_wait(0x0453, timeout=timeout, chan_ident=1, + absolute_distance=_position_in_encoder, + destination=axis, source=0x01) + return + + # ā”€ā”€ Velocity parameters ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ + + def get_velocity_params(self, axis, timeout=5.0): + ''' + get_velocity_params(axis): Queries the current velocity parameters + for the specified axis. Returns a dict with keys: + min_velocity (mm/s), acceleration (mm/s2), max_velocity (mm/s) + ''' + if axis == 0x21: + ch = 0 + elif axis == 0x22: + ch = 1 + else: + raise ValueError("I don't know that axis!") + + result = self.send_and_wait(0x0414, timeout=timeout, chan_ident=1, + zero_this=0x00, destination=axis, + source=0x01) + params = { + 'min_velocity': result['min_velocity'] / self.velocity_scaling, + 'acceleration': result['acceleration'] / self.accel_scaling, + 'max_velocity': result['max_velocity'] / self.velocity_scaling, + } + + self.max_velocities[ch] = params['max_velocity'] + self.max_accels[ch] = params['acceleration'] + + return params + + def set_velocity_params(self, axis, max_velocity=None, acceleration=None): + ''' + set_velocity_params(axis, max_velocity=None, acceleration=None): + Sets velocity and/or acceleration for the specified axis. + Values are in mm/s and mm/s2 respectively. Any parameter + left as None keeps its current value. + ''' + if axis == 0x21: + ch = 0 + elif axis == 0x22: + ch = 1 + else: + raise ValueError("I don't know that axis!") + + # Only query current params if we need to fill in a missing value + if max_velocity is None or acceleration is None: + current = self.get_velocity_params(axis) + if max_velocity is None: + max_velocity = current['max_velocity'] + if acceleration is None: + acceleration = current['acceleration'] + + # Update the cached values + self.max_velocities[ch] = max_velocity + self.max_accels[ch] = acceleration + + _min_v = 0 + _accel = int(round(acceleration * self.accel_scaling)) + _max_v = int(round(max_velocity * self.velocity_scaling)) + + self.send_message(0x0413, chan_ident=1, + min_velocity=_min_v, + acceleration=_accel, + max_velocity=_max_v, + destination=axis, source=0x01) + + # ā”€ā”€ Trigger control ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ + + def set_trigger(self, axis, mode): + ''' + set_trigger(axis, mode): Sets the trigger mode for the specified + axis. Mode should be a TriggerBitsServo value or combination. + ''' + self.send_message(0x0500, chan_ident=1, mode=int(mode), + destination=axis, source=0x01) + + def get_trigger(self, axis, timeout=5.0): + ''' + get_trigger(axis): Queries the current trigger mode for the + specified axis. Returns the mode byte as a TriggerBitsServo. + ''' + result = self.send_and_wait(0x0501, timeout=timeout, + chan_ident=1, mode=0x00, + destination=axis, source=0x01) + return TriggerBitsServo(result['mode']) + + def set_trigger_trigin_high(self, axis): + '''Set trigger input to logic high.''' + self.set_trigger(axis, TriggerBitsServo.TRIGIN_HIGH) + + def set_trigger_trigin_relmove(self, axis): + '''Set trigger input to initiate a relative move.''' + self.set_trigger(axis, TriggerBitsServo.TRIGIN_RELMOVE) + + def set_trigger_trigin_absmove(self, axis): + '''Set trigger input to initiate an absolute move.''' + self.set_trigger(axis, TriggerBitsServo.TRIGIN_ABSMOVE) + + def set_trigger_trigin_homemove(self, axis): + '''Set trigger input to initiate a home move.''' + self.set_trigger(axis, TriggerBitsServo.TRIGIN_HOMEMOVE) + + def set_trigger_trigout_high(self, axis): + '''Set trigger output to logic high.''' + self.set_trigger(axis, TriggerBitsServo.TRIGOUT_HIGH) + + def set_trigger_trigout_inmotion(self, axis): + '''Set trigger output high while axis is in motion.''' + self.set_trigger(axis, TriggerBitsServo.TRIGOUT_INMOTION) + + def set_trigger_trigout_motioncomplete(self, axis): + '''Set trigger output to pulse when motion completes.''' + self.set_trigger(axis, TriggerBitsServo.TRIGOUT_MOTIONCOMPLETE) + + def set_trigger_trigout_maxvelocity(self, axis): + '''Set trigger output to pulse at max velocity.''' + self.set_trigger(axis, TriggerBitsServo.TRIGOUT_MAXVELOCITY) + + def set_trigger_trigout_maxv(self, axis): + '''Set trigger output high + pulse at max velocity (TRIGOUT_MAXV).''' + self.set_trigger(axis, TriggerBitsServo.TRIGOUT_MAXV) diff --git a/hardware/pybbd202/serial_comms.py b/hardware/pybbd202/serial_comms.py new file mode 100644 index 0000000..7fc615f --- /dev/null +++ b/hardware/pybbd202/serial_comms.py @@ -0,0 +1,94 @@ +''' + SRAS Serial Communication Tools + Thomas Ales | Feb 2026 + Version 1 +''' +from threading import Thread +from queue import Queue +import serial +import struct +import time + +class SerialSnooper(Thread): + + def __init__(self, _port, _spd): + super().__init__() + self.serial_connection = None + self.serial_port = _port + self.serial_speed = _spd + self.am_listening = False + self.rx_msg_queue = Queue() + + def run(self): + ''' + run() - Starts up the serial listener, checks if the port was + opened successfully, and if so begins + listening for APT messages. + ''' + self.serial_connection = serial.Serial(port=self.serial_port, + baudrate=self.serial_speed, + rtscts=True, timeout=0.05) + if self.serial_connection.is_open is True: + # Send disconnect to stop any ongoing auto-updates from + # a previous session + for addr in [0x11, 0x21, 0x22]: + disconnect_msg = struct.pack(' 0: + self.serial_connection.read(self.serial_connection.in_waiting) + + self.am_listening = True + _rxbuf = bytearray() + while self.am_listening is True: + try: + # Read whatever is available (or wait up to timeout) + _chunk = self.serial_connection.read( + max(1, self.serial_connection.in_waiting)) + if _chunk: + _rxbuf.extend(_chunk) + + # Process complete messages from the buffer + while len(_rxbuf) >= 6: + # Check if this is a long message (bit 7 of byte 4) + if _rxbuf[4] & 0x80: + _msglen = struct.unpack(" y_end: + y_current = y_end + + # Alternate scan direction for snake pattern + if line_idx % 2 == 0: + scan_start_x, scan_end_x = x_start, x_end + else: + scan_start_x, scan_end_x = x_end, x_start + + self.line_started.emit(line_idx, num_lines, y_current) + + # Position to line start (diagonal move - X and Y simultaneously) + # Y move issued first (non-blocking from stage perspective), + # then X move; both complete before scan line begins + controller.move_axis_absolute(AXIS_Y, y_current, timeout=20.0) + controller.move_axis_absolute(AXIS_X, scan_start_x, timeout=20.0) + + # Perform scan line (trigger is HIGH during this move) + scan_distance = abs(scan_end_x - scan_start_x) + scan_timeout = max(10.0, scan_distance / scan_velocity * 3) + controller.move_axis_absolute(AXIS_X, scan_end_x, timeout=scan_timeout) + + # Update progress + line_progress = int(100 * (line_idx + 1) / num_lines) + self.current_progress.emit(line_progress) + + overall = int(100 * (angle_idx + (line_idx + 1) / num_lines) / num_angles) + self.overall_progress.emit(overall) + + # Cleanup: disable triggers and restore original velocity + controller.set_trigger(AXIS_X, 0) + controller.set_velocity_params(AXIS_X, max_velocity=orig_x_velocity, acceleration=orig_x_accel) + controller.set_velocity_params(AXIS_Y, max_velocity=orig_y_velocity, acceleration=orig_y_accel) + + self.status_message.emit("Scan complete") + self.scan_completed.emit() except Exception as e: print(f"ERROR in scan worker: {e}") import traceback traceback.print_exc() + # Best-effort cleanup on error + try: + if self.motion_worker and self.motion_worker.controller: + self.motion_worker.controller.set_trigger(AXIS_X, 0) + except Exception: + pass self.scan_failed.emit(str(e)) finally: # Re-enable motion worker polling @@ -2395,8 +2443,8 @@ class ScanVisualizationDialog(QtWidgets.QDialog): self.current_scan_label.setText("Checking home status...") # Get home status for both axes - x_homed = controller.is_homed_x - y_homed = controller.is_homed_y + x_homed = controller.am_homed[0] + y_homed = controller.am_homed[1] print(f"Home status - X: {x_homed}, Y: {y_homed}") @@ -2430,26 +2478,25 @@ class ScanVisualizationDialog(QtWidgets.QDialog): QtWidgets.QApplication.processEvents() try: - success = controller.home_axis(controller.DEST_X_AXIS, timeout=60.0) - if success: - print("X-axis homed successfully") - self.current_scan_progress.setValue(50) + controller.home_axis(AXIS_X, timeout=60.0) + print("X-axis homed successfully") + self.current_scan_progress.setValue(50) - # Small delay to let system stabilize - import time - time.sleep(0.5) - else: - print("ERROR: X-axis homing timeout") - QtWidgets.QMessageBox.critical( - self, - "Homing Error", - "X-axis homing timed out after 60 seconds.\n\n" - "Please check:\n" - "ā€¢ Stage can move freely\n" - "ā€¢ No obstructions\n" - "ā€¢ Stage is connected properly" - ) - return False + # Small delay to let system stabilize + import time + time.sleep(0.5) + except TimeoutError: + print("ERROR: X-axis homing timeout") + QtWidgets.QMessageBox.critical( + self, + "Homing Error", + "X-axis homing timed out after 60 seconds.\n\n" + "Please check:\n" + "ā€¢ Stage can move freely\n" + "ā€¢ No obstructions\n" + "ā€¢ Stage is connected properly" + ) + return False except Exception as e: print(f"ERROR: Failed to home X-axis: {e}") QtWidgets.QMessageBox.critical( @@ -2467,26 +2514,25 @@ class ScanVisualizationDialog(QtWidgets.QDialog): QtWidgets.QApplication.processEvents() try: - success = controller.home_axis(controller.DEST_Y_AXIS, timeout=60.0) - if success: - print("Y-axis homed successfully") - self.current_scan_progress.setValue(90) + controller.home_axis(AXIS_Y, timeout=60.0) + print("Y-axis homed successfully") + self.current_scan_progress.setValue(90) - # Small delay to let system stabilize - import time - time.sleep(0.5) - else: - print("ERROR: Y-axis homing timeout") - QtWidgets.QMessageBox.critical( - self, - "Homing Error", - "Y-axis homing timed out after 60 seconds.\n\n" - "Please check:\n" - "ā€¢ Stage can move freely\n" - "ā€¢ No obstructions\n" - "ā€¢ Stage is connected properly" - ) - return False + # Small delay to let system stabilize + import time + time.sleep(0.5) + except TimeoutError: + print("ERROR: Y-axis homing timeout") + QtWidgets.QMessageBox.critical( + self, + "Homing Error", + "Y-axis homing timed out after 60 seconds.\n\n" + "Please check:\n" + "ā€¢ Stage can move freely\n" + "ā€¢ No obstructions\n" + "ā€¢ Stage is connected properly" + ) + return False except Exception as e: print(f"ERROR: Failed to home Y-axis: {e}") QtWidgets.QMessageBox.critical( @@ -2689,16 +2735,16 @@ class OptionsDialog(QtWidgets.QDialog): def populate_trigger_modes(self): """Populate the trigger mode combo boxes with available options""" - from hardware.bbd202 import TriggerMode + from hardware.pybbd202 import TriggerBitsServo trigger_options = [ - ("Disabled", TriggerMode.DISABLED), - ("In/Out Relative Move", TriggerMode.IN_OUT_RELATIVE_MOVE), - ("In/Out Absolute Move", TriggerMode.IN_OUT_ABSOLUTE_MOVE), - ("In/Out Home", TriggerMode.IN_OUT_HOME), - ("In/Out Stop", TriggerMode.IN_OUT_STOP), - ("Out Only (HIGH during motion)", TriggerMode.OUT_ONLY), - ("Out Position", TriggerMode.OUT_POSITION), + ("Disabled", 0), + ("Trigger In: Relative Move", TriggerBitsServo.TRIGIN_RELMOVE), + ("Trigger In: Absolute Move", TriggerBitsServo.TRIGIN_ABSMOVE), + ("Trigger In: Home", TriggerBitsServo.TRIGIN_HOMEMOVE), + ("Trigger Out: In Motion", TriggerBitsServo.TRIGOUT_INMOTION), + ("Trigger Out: Motion Complete", TriggerBitsServo.TRIGOUT_MOTIONCOMPLETE), + ("Trigger Out: Max Velocity", TriggerBitsServo.TRIGOUT_MAXVELOCITY), ] for label, mode in trigger_options: diff --git a/scanengine/motion_worker.py b/scanengine/motion_worker.py index a3c792a..e2f523d 100644 --- a/scanengine/motion_worker.py +++ b/scanengine/motion_worker.py @@ -6,7 +6,7 @@ Provides async command queueing and position updates via Qt signals. """ from PyQt6 import QtCore -from hardware.bbd202 import MotionController +from hardware.pybbd202 import ThorlabsServoDriver, AXIS_X, AXIS_Y import queue import time from typing import Optional, Dict, Any @@ -45,7 +45,7 @@ class MotionWorker(QtCore.QObject): def __init__(self): super().__init__() - self.controller: Optional[MotionController] = None + self.controller: Optional[ThorlabsServoDriver] = None self.is_connected = False self.command_queue = queue.Queue() self.running = True @@ -65,9 +65,9 @@ class MotionWorker(QtCore.QObject): self.last_x_moving = None self.last_y_moving = None - # Position update throttling (active requests can be slow) + # Position update throttling self.last_position_update_time = 0 - self.position_update_interval = 0.2 # seconds between position requests + self.position_update_interval = 0.2 # seconds between position reads # Flag to pause polling during scanning (scan worker handles its own position queries) self.scanning_active = False @@ -133,27 +133,25 @@ class MotionWorker(QtCore.QObject): def do_connect(self): """Connect to the motion controller""" try: - self.controller = MotionController() - self.controller.connect() # ACKs are sent reactively when enable_updates=True + self.controller = ThorlabsServoDriver() + self.controller.connect() # Enable channels - self.controller.set_channel_enable_state(self.controller.DEST_X_AXIS, True) - self.controller.set_channel_enable_state(self.controller.DEST_Y_AXIS, True) + self.controller.enable_axis(AXIS_X) + self.controller.enable_axis(AXIS_Y) - # Request status updates for both axes to populate status bits (including homed state) - # This is necessary because status bits are not sent automatically after connect - self.controller.request_status_update(self.controller.DEST_X_AXIS) - self.controller.request_status_update(self.controller.DEST_Y_AXIS) - # Wait a moment for the asynchronous status update responses - time.sleep(0.2) + # Start polling to populate cached state (positions, homed, moving, errors) + self.controller.start_polling(interval=0.2) + + # Wait for first polling cycle to populate status + time.sleep(0.3) # Set initial velocity parameters - for dest in [self.controller.DEST_X_AXIS, self.controller.DEST_Y_AXIS]: + for dest in [AXIS_X, AXIS_Y]: self.controller.set_velocity_params( dest, - min_velocity=0.0, - acceleration=self.acceleration, - max_velocity=self.jog_speed + max_velocity=self.jog_speed, + acceleration=self.acceleration ) self.is_connected = True @@ -188,29 +186,25 @@ class MotionWorker(QtCore.QObject): return try: - # Determine destination - dest = self.controller.DEST_X_AXIS if axis == 'x' else self.controller.DEST_Y_AXIS + dest = AXIS_X if axis == 'x' else AXIS_Y # Calculate relative distance distance = self.step_size * direction - # Set relative move parameters - self.controller.set_move_rel_params(dest, distance) - - # Execute the move (non-blocking - we don't wait for completion) - # Use a very short timeout since we're doing continuous jogging - self.controller.move_relative(dest, timeout=0.5) + # Execute the move (blocking, with short timeout for continuous jogging) + self.controller.move_axis_relative(dest, distance, timeout=0.5) # Update position self.update_position() self.move_completed.emit(axis) + except TimeoutError: + # Timeout is expected during continuous jog - don't report as error + pass except Exception as e: - # Don't emit errors for timeout - that's expected during continuous jog - if "timeout" not in str(e).lower(): - print(f"Jog error: {e}") - self.error_occurred.emit(f"Jog failed: {str(e)}") + print(f"Jog error: {e}") + self.error_occurred.emit(f"Jog failed: {str(e)}") def do_home(self, axis: str): """Home an axis""" @@ -218,7 +212,7 @@ class MotionWorker(QtCore.QObject): return try: - dest = self.controller.DEST_X_AXIS if axis == 'x' else self.controller.DEST_Y_AXIS + dest = AXIS_X if axis == 'x' else AXIS_Y print(f"Homing {axis.upper()} axis...") self.controller.home_axis(dest, timeout=20.0) @@ -229,6 +223,9 @@ class MotionWorker(QtCore.QObject): print(f"{axis.upper()} axis homed successfully") + except TimeoutError: + print(f"Home timeout: {axis.upper()} axis") + self.error_occurred.emit(f"Homing {axis.upper()} timed out") except Exception as e: print(f"Home error: {e}") self.error_occurred.emit(f"Homing {axis.upper()} failed: {str(e)}") @@ -244,12 +241,11 @@ class MotionWorker(QtCore.QObject): self.jog_speed = speed self.acceleration = accel - for dest in [self.controller.DEST_X_AXIS, self.controller.DEST_Y_AXIS]: + for dest in [AXIS_X, AXIS_Y]: self.controller.set_velocity_params( dest, - min_velocity=0.0, - acceleration=self.acceleration, - max_velocity=self.jog_speed + max_velocity=self.jog_speed, + acceleration=self.acceleration ) except Exception as e: @@ -261,8 +257,11 @@ class MotionWorker(QtCore.QObject): return try: - dest = self.controller.DEST_X_AXIS if axis == 'x' else self.controller.DEST_Y_AXIS - self.controller.set_channel_enable_state(dest, enabled) + dest = AXIS_X if axis == 'x' else AXIS_Y + if enabled: + self.controller.enable_axis(dest) + else: + self.controller.disable_axis(dest) state_str = "enabled" if enabled else "disabled" print(f"{axis.upper()} axis {state_str}") @@ -275,35 +274,25 @@ class MotionWorker(QtCore.QObject): if not self.is_connected or not self.controller: return - # Throttle position requests to avoid slowing down the main loop + # Throttle position reads to avoid excessive signal emission current_time = time.time() if current_time - self.last_position_update_time < self.position_update_interval: return self.last_position_update_time = current_time try: - # Actively request positions from the controller instead of relying on cached values - # This ensures we always have up-to-date position data - # Use longer timeout (1.5s) to accommodate high-speed scanning at 200mm/s - x_pos = self.controller.get_position(self.controller.DEST_X_AXIS, timeout=1.5) - y_pos = self.controller.get_position(self.controller.DEST_Y_AXIS, timeout=1.5) + # Read cached positions (populated by polling worker) + x_pos = self.controller.positions[0] + y_pos = self.controller.positions[1] - if x_pos is not None and y_pos is not None: - # Always emit on first update, or if position changed significantly (> 0.001mm) - if (self.last_x is None or self.last_y is None or - abs(x_pos - self.last_x) > 0.001 or abs(y_pos - self.last_y) > 0.001): - self.last_x = x_pos - self.last_y = y_pos - print(f"Position update: X={x_pos:.3f}mm, Y={y_pos:.3f}mm") - self.position_updated.emit(x_pos, y_pos) - # else: silently skip incomplete position data during busy scanning + # Always emit on first update, or if position changed significantly (> 0.001mm) + if (self.last_x is None or self.last_y is None or + abs(x_pos - self.last_x) > 0.001 or abs(y_pos - self.last_y) > 0.001): + self.last_x = x_pos + self.last_y = y_pos + print(f"Position update: X={x_pos:.3f}mm, Y={y_pos:.3f}mm") + self.position_updated.emit(x_pos, y_pos) - except RuntimeError as e: - # RuntimeError indicates actual hardware error (overtemp, encoder fault, etc.) - print(f"CRITICAL: Motor error detected: {e}") - self.error_occurred.emit(str(e)) - # Stop requesting position updates to avoid spam - self.is_connected = False except Exception as e: print(f"Error updating position: {e}") import traceback @@ -315,8 +304,8 @@ class MotionWorker(QtCore.QObject): return try: - x_homed = self.controller.is_homed_x - y_homed = self.controller.is_homed_y + x_homed = self.controller.am_homed[0] + y_homed = self.controller.am_homed[1] # Only emit if status changed if x_homed != self.last_x_homed or y_homed != self.last_y_homed: @@ -332,42 +321,22 @@ class MotionWorker(QtCore.QObject): def update_motion_status(self): """Update motion status and emit signal if changed. - NOTE: On BBD202 firmware v2.1.5, the is_in_motion_x/y properties - do NOT reliably detect motion - they may always return False even - during active movement. This is a known firmware limitation. - - For scan execution, use the ScanWorker.move_and_wait() method which - uses position-based motion detection instead of status bits. - - This UI status display is best-effort only. + The new driver's polling worker keeps am_moving[], am_error[] + up to date automatically via status update messages. """ if not self.is_connected or not self.controller: return try: - # Actively poll for status updates - self.controller.poll_status() - # Check for any error conditions - error_msg = self.controller.check_for_errors() - if error_msg: - print(f"Motor error detected: {error_msg}") - self.error_occurred.emit(error_msg) - self.controller.clear_last_error() + if self.controller.am_error[0]: + self.error_occurred.emit("X-axis error detected") + if self.controller.am_error[1]: + self.error_occurred.emit("Y-axis error detected") - # Read the cached status (may not accurately reflect motion on some firmware) - x_moving = self.controller.is_in_motion_x - y_moving = self.controller.is_in_motion_y - - # Also check if there are pending moves (more reliable) - if self.controller.is_move_pending(): - # If there are pending moves, we're likely still moving - # This provides a backup indication when status bits fail - pending = self.controller.get_pending_targets() - if self.controller.DEST_X_AXIS in pending: - x_moving = True - if self.controller.DEST_Y_AXIS in pending: - y_moving = True + # Read cached motion status (updated by polling worker) + x_moving = self.controller.am_moving[0] + y_moving = self.controller.am_moving[1] # Only emit if status changed if x_moving != self.last_x_moving or y_moving != self.last_y_moving: