Initial commit: merge nuescan, pymso, pybbd202, and pypewpewhops into scanengine-3
- Merged four separate hardware control projects into unified platform - Created unified requirements.txt with all dependencies - Added comprehensive .gitignore - Added project overview README Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
This commit is contained in:
Thomas Ales [M S E]
@@ -0,0 +1,4 @@
|
||||
"""
|
||||
Hardware communication modules for nueScan
|
||||
Handles communication with ThorLabs stage, T3R device, and microscopes
|
||||
"""
|
||||
@@ -0,0 +1,928 @@
|
||||
"""
|
||||
ThorLabs BBD203 3-Channel Motor Controller Driver
|
||||
Complete implementation of the APT protocol for BBD203
|
||||
|
||||
Copyright (C) 2025 Thomas Ales
|
||||
Licensed under GNU General Public License v2.0
|
||||
"""
|
||||
|
||||
import serial
|
||||
import serial.tools.list_ports
|
||||
import time
|
||||
import threading
|
||||
from typing import Dict, List, Optional, Callable, Tuple
|
||||
from queue import Queue, Empty
|
||||
|
||||
from hardware.bbd203_protocol import (
|
||||
APTProtocol, APTMessage, MessageID, StatusBits, TriggerMode, Destination
|
||||
)
|
||||
|
||||
|
||||
class BBD203Channel:
|
||||
"""Represents a single channel on the BBD203"""
|
||||
|
||||
def __init__(self, channel_num: int):
|
||||
"""
|
||||
Initialize channel
|
||||
|
||||
Args:
|
||||
channel_num: Channel number (1, 2, or 3)
|
||||
"""
|
||||
self.channel_num = channel_num
|
||||
self.enabled = False
|
||||
self.homed = False
|
||||
self.position_mm = 0.0
|
||||
self.encoder_count = 0
|
||||
self.status_bits = 0
|
||||
self.moving = False
|
||||
self.homing = False
|
||||
self.error = False
|
||||
|
||||
def update_from_status(self, position: int, encoder: int, status: int,
|
||||
protocol: APTProtocol):
|
||||
"""Update channel state from status update"""
|
||||
self.position_mm = protocol.apt_to_position(position)
|
||||
self.encoder_count = encoder
|
||||
self.status_bits = status
|
||||
|
||||
# Parse status bits
|
||||
self.homed = bool(status & StatusBits.HOMED)
|
||||
self.homing = bool(status & StatusBits.HOMING)
|
||||
self.enabled = bool(status & StatusBits.MOTOR_ENABLED)
|
||||
self.error = bool(status & StatusBits.MOTION_ERROR)
|
||||
|
||||
# Check if moving
|
||||
self.moving = bool(status & (
|
||||
StatusBits.IN_MOTION_FORWARD |
|
||||
StatusBits.IN_MOTION_REVERSE |
|
||||
StatusBits.JOGGING_FORWARD |
|
||||
StatusBits.JOGGING_REVERSE |
|
||||
StatusBits.HOMING
|
||||
))
|
||||
|
||||
def is_ready(self) -> bool:
|
||||
"""Check if channel is ready for operation"""
|
||||
return self.enabled and self.homed and not self.error
|
||||
|
||||
|
||||
class BBD203Driver:
|
||||
"""
|
||||
Complete driver for ThorLabs BBD203 3-Channel Motor Controller
|
||||
|
||||
Features:
|
||||
- 3 independent motor channels
|
||||
- Binary APT protocol communication
|
||||
- Automatic status updates
|
||||
- Thread-safe operation
|
||||
- Position and velocity control
|
||||
"""
|
||||
|
||||
def __init__(self, encoder_counts_per_mm: int = 20000, timeout: float = 1.0):
|
||||
"""
|
||||
Initialize BBD203 driver
|
||||
|
||||
Args:
|
||||
encoder_counts_per_mm: Encoder resolution (default: 20000 for MLS203)
|
||||
timeout: Serial communication timeout in seconds
|
||||
"""
|
||||
self.protocol = APTProtocol(encoder_counts_per_mm)
|
||||
self.timeout = timeout
|
||||
|
||||
# Serial connection
|
||||
self._serial: Optional[serial.Serial] = None
|
||||
self._port_name = ""
|
||||
self._connected = False
|
||||
|
||||
# Channels
|
||||
self.channels = {
|
||||
1: BBD203Channel(1),
|
||||
2: BBD203Channel(2),
|
||||
3: BBD203Channel(3)
|
||||
}
|
||||
|
||||
# Communication thread
|
||||
self._rx_thread: Optional[threading.Thread] = None
|
||||
self._stop_thread = threading.Event()
|
||||
self._rx_queue = Queue()
|
||||
|
||||
# Callbacks for asynchronous events
|
||||
self._move_complete_callbacks: Dict[int, List[Callable]] = {1: [], 2: [], 3: []}
|
||||
self._home_complete_callbacks: Dict[int, List[Callable]] = {1: [], 2: [], 3: []}
|
||||
|
||||
# Hardware info
|
||||
self._hw_info = {}
|
||||
|
||||
# ==================== Connection Management ====================
|
||||
|
||||
@staticmethod
|
||||
def list_available_ports() -> List[str]:
|
||||
"""
|
||||
List available serial ports
|
||||
|
||||
Returns:
|
||||
list: Available port names
|
||||
"""
|
||||
ports = serial.tools.list_ports.comports()
|
||||
return [port.device for port in ports]
|
||||
|
||||
@staticmethod
|
||||
def list_thorlabs_devices() -> List[Dict[str, str]]:
|
||||
"""
|
||||
List all ThorLabs APT devices connected via USB
|
||||
|
||||
Returns:
|
||||
list: List of dictionaries containing device information
|
||||
Each dict has: 'serial', 'port', 'description', 'vid', 'pid'
|
||||
"""
|
||||
thorlabs_devices = []
|
||||
ports = serial.tools.list_ports.comports()
|
||||
|
||||
# ThorLabs devices typically use FTDI chips
|
||||
# Common VID/PID combinations:
|
||||
# - FTDI: VID=0x0403, various PIDs
|
||||
thorlabs_vids = [0x0403] # FTDI vendor ID
|
||||
|
||||
for port in ports:
|
||||
# Check if this is a ThorLabs device by VID
|
||||
if port.vid in thorlabs_vids:
|
||||
device_info = {
|
||||
'serial': port.serial_number or 'Unknown',
|
||||
'port': port.device,
|
||||
'description': port.description or 'Unknown',
|
||||
'manufacturer': port.manufacturer or 'Unknown',
|
||||
'vid': f"0x{port.vid:04X}" if port.vid else 'Unknown',
|
||||
'pid': f"0x{port.pid:04X}" if port.pid else 'Unknown'
|
||||
}
|
||||
thorlabs_devices.append(device_info)
|
||||
print(f"DEBUG: Found ThorLabs device - Serial: {device_info['serial']}, "
|
||||
f"Port: {device_info['port']}")
|
||||
|
||||
return thorlabs_devices
|
||||
|
||||
@staticmethod
|
||||
def find_device_by_serial(serial_number: str) -> Optional[str]:
|
||||
"""
|
||||
Find ThorLabs device by serial number and return its port
|
||||
|
||||
Args:
|
||||
serial_number: Device serial number (e.g., '83123456')
|
||||
|
||||
Returns:
|
||||
str: COM port name if found, None otherwise
|
||||
"""
|
||||
devices = BBD203Driver.list_thorlabs_devices()
|
||||
|
||||
for device in devices:
|
||||
if device['serial'] == serial_number:
|
||||
print(f"INFO: Found device {serial_number} on port {device['port']}")
|
||||
return device['port']
|
||||
|
||||
print(f"WARNING: Device with serial number {serial_number} not found")
|
||||
print(f"Available devices: {[d['serial'] for d in devices]}")
|
||||
return None
|
||||
|
||||
def connect_by_serial(self, serial_number: str, baudrate: int = 115200) -> bool:
|
||||
"""
|
||||
Connect to BBD203 controller by serial number (auto-find port)
|
||||
|
||||
This is the preferred connection method - automatically finds the
|
||||
device by serial number over USB, similar to Kinesis library.
|
||||
|
||||
Args:
|
||||
serial_number: Device serial number (e.g., '83123456')
|
||||
baudrate: Baud rate (default: 115200)
|
||||
|
||||
Returns:
|
||||
bool: True if connection successful
|
||||
|
||||
Example:
|
||||
driver.connect_by_serial('83123456')
|
||||
"""
|
||||
# Find device port by serial number
|
||||
port = self.find_device_by_serial(serial_number)
|
||||
|
||||
if port is None:
|
||||
print(f"ERROR: Could not find BBD203 with serial number {serial_number}")
|
||||
print("Available ThorLabs devices:")
|
||||
for device in self.list_thorlabs_devices():
|
||||
print(f" Serial: {device['serial']}, Port: {device['port']}, "
|
||||
f"Description: {device['description']}")
|
||||
return False
|
||||
|
||||
# Connect using the found port
|
||||
return self.connect(port, baudrate)
|
||||
|
||||
def connect(self, port: str, baudrate: int = 115200) -> bool:
|
||||
"""
|
||||
Connect to BBD203 controller by port name
|
||||
|
||||
Note: It's recommended to use connect_by_serial() instead, which
|
||||
automatically finds the device by serial number.
|
||||
|
||||
Args:
|
||||
port: Serial port name (e.g., 'COM3' or '/dev/ttyUSB0')
|
||||
baudrate: Baud rate (default: 115200)
|
||||
|
||||
Returns:
|
||||
bool: True if connection successful
|
||||
"""
|
||||
try:
|
||||
print(f"INFO: Connecting to BBD203 on {port}")
|
||||
|
||||
self._serial = serial.Serial(
|
||||
port=port,
|
||||
baudrate=baudrate,
|
||||
bytesize=serial.EIGHTBITS,
|
||||
parity=serial.PARITY_NONE,
|
||||
stopbits=serial.STOPBITS_ONE,
|
||||
timeout=self.timeout,
|
||||
rtscts=False, # Disable hardware flow control
|
||||
xonxoff=False # Disable software flow control
|
||||
)
|
||||
|
||||
# Set DTR and RTS for ThorLabs FTDI devices
|
||||
# For BBD203, RTS should be LOW to enable communication
|
||||
self._serial.dtr = False
|
||||
self._serial.rts = False
|
||||
|
||||
self._port_name = port
|
||||
self._connected = True
|
||||
|
||||
# Give controller time to initialize after connection
|
||||
time.sleep(0.5)
|
||||
|
||||
# Start receive thread
|
||||
self._stop_thread.clear()
|
||||
self._rx_thread = threading.Thread(target=self._receive_loop, daemon=True)
|
||||
self._rx_thread.start()
|
||||
|
||||
# Initialize controller
|
||||
time.sleep(0.5) # Allow thread to start and controller to be ready
|
||||
|
||||
# Request hardware info
|
||||
self._send_command(self.protocol.cmd_req_hw_info())
|
||||
time.sleep(0.5)
|
||||
|
||||
# Start automatic status updates
|
||||
self._send_command(self.protocol.cmd_start_update_msgs())
|
||||
time.sleep(0.5)
|
||||
|
||||
print(f"INFO: Successfully connected to BBD203 on {port}")
|
||||
return True
|
||||
|
||||
except serial.SerialException as e:
|
||||
print(f"ERROR: Failed to connect to {port}: {e}")
|
||||
self._connected = False
|
||||
return False
|
||||
|
||||
def disconnect(self) -> bool:
|
||||
"""
|
||||
Disconnect from BBD203 controller
|
||||
|
||||
Returns:
|
||||
bool: True if disconnection successful
|
||||
"""
|
||||
if not self._connected:
|
||||
return True
|
||||
|
||||
try:
|
||||
print("INFO: Disconnecting from BBD203")
|
||||
|
||||
# Stop status updates
|
||||
self._send_command(self.protocol.cmd_stop_update_msgs())
|
||||
time.sleep(0.1)
|
||||
|
||||
# Stop receive thread
|
||||
self._stop_thread.set()
|
||||
if self._rx_thread:
|
||||
self._rx_thread.join(timeout=2.0)
|
||||
|
||||
# Close serial port
|
||||
if self._serial and self._serial.is_open:
|
||||
self._serial.close()
|
||||
|
||||
self._connected = False
|
||||
print("INFO: Disconnected from BBD203")
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
print(f"ERROR: Error during disconnect: {e}")
|
||||
return False
|
||||
|
||||
def is_connected(self) -> bool:
|
||||
"""Check if controller is connected"""
|
||||
return self._connected and self._serial and self._serial.is_open
|
||||
|
||||
# ==================== Communication Methods ====================
|
||||
|
||||
def _send_command(self, cmd: bytes) -> bool:
|
||||
"""
|
||||
Send command to controller
|
||||
|
||||
Args:
|
||||
cmd: Command bytes to send
|
||||
|
||||
Returns:
|
||||
bool: True if send successful
|
||||
"""
|
||||
if not self.is_connected():
|
||||
print("ERROR: Cannot send command - not connected")
|
||||
return False
|
||||
|
||||
try:
|
||||
print(f"DEBUG: Sending {len(cmd)} bytes: {cmd.hex()}")
|
||||
self._serial.write(cmd)
|
||||
self._serial.flush() # Ensure data is sent
|
||||
return True
|
||||
except serial.SerialException as e:
|
||||
print(f"ERROR: Failed to send command: {e}")
|
||||
return False
|
||||
|
||||
def _receive_loop(self):
|
||||
"""Background thread to receive messages from controller"""
|
||||
buffer = bytearray()
|
||||
print("DEBUG: Receive thread started")
|
||||
|
||||
while not self._stop_thread.is_set():
|
||||
try:
|
||||
if self._serial.in_waiting > 0:
|
||||
data = self._serial.read(self._serial.in_waiting)
|
||||
print(f"DEBUG: Received {len(data)} bytes: {data.hex()}")
|
||||
buffer.extend(data)
|
||||
|
||||
# Process complete messages
|
||||
while len(buffer) >= 6:
|
||||
# Parse header
|
||||
msg_id, data_len, dest, source = APTMessage.parse_header(buffer)
|
||||
|
||||
# Determine total message length
|
||||
if data_len == 0 or data_len > 255:
|
||||
# Header-only message
|
||||
msg_len = 6
|
||||
else:
|
||||
# Message with data
|
||||
msg_len = 6 + data_len
|
||||
|
||||
# Wait for complete message
|
||||
if len(buffer) < msg_len:
|
||||
break
|
||||
|
||||
# Extract message
|
||||
msg = bytes(buffer[:msg_len])
|
||||
buffer = buffer[msg_len:]
|
||||
|
||||
# Process message
|
||||
self._process_message(msg_id, msg)
|
||||
|
||||
else:
|
||||
time.sleep(0.001) # Small delay to prevent busy waiting
|
||||
|
||||
except Exception as e:
|
||||
if not self._stop_thread.is_set():
|
||||
print(f"ERROR: Exception in receive loop: {e}")
|
||||
time.sleep(0.1)
|
||||
|
||||
def _process_message(self, msg_id: int, msg: bytes):
|
||||
"""
|
||||
Process received message
|
||||
|
||||
Args:
|
||||
msg_id: Message ID
|
||||
msg: Complete message bytes
|
||||
"""
|
||||
try:
|
||||
print(f"DEBUG: Processing message ID 0x{msg_id:04X}, len={len(msg)}, data={msg.hex()}")
|
||||
|
||||
if msg_id == MessageID.MGMSG_MOT_GET_STATUSUPDATE:
|
||||
# Status update
|
||||
channel, position, encoder, status = APTMessage.parse_status_update(msg)
|
||||
# Determine which channel this is for (from destination byte)
|
||||
dest = msg[4]
|
||||
channel_num = dest - 0x20 # 0x21->1, 0x22->2, 0x23->3
|
||||
|
||||
if channel_num in self.channels:
|
||||
self.channels[channel_num].update_from_status(
|
||||
position, encoder, status, self.protocol
|
||||
)
|
||||
|
||||
elif msg_id == MessageID.MGMSG_MOT_MOVE_COMPLETED:
|
||||
# Move completed
|
||||
dest = msg[4]
|
||||
channel_num = dest - 0x20
|
||||
|
||||
if channel_num in self.channels:
|
||||
self.channels[channel_num].moving = False
|
||||
|
||||
# Call callbacks
|
||||
for callback in self._move_complete_callbacks.get(channel_num, []):
|
||||
callback(channel_num)
|
||||
|
||||
elif msg_id == MessageID.MGMSG_MOT_MOVE_HOMED:
|
||||
# Homing completed
|
||||
dest = msg[4]
|
||||
channel_num = dest - 0x20
|
||||
|
||||
if channel_num in self.channels:
|
||||
self.channels[channel_num].homed = True
|
||||
self.channels[channel_num].homing = False
|
||||
|
||||
# Call callbacks
|
||||
for callback in self._home_complete_callbacks.get(channel_num, []):
|
||||
callback(channel_num)
|
||||
|
||||
elif msg_id == MessageID.MGMSG_MOT_MOVE_STOPPED:
|
||||
# Motion stopped
|
||||
dest = msg[4]
|
||||
channel_num = dest - 0x20
|
||||
|
||||
if channel_num in self.channels:
|
||||
self.channels[channel_num].moving = False
|
||||
|
||||
elif msg_id == MessageID.MGMSG_MOD_GET_CHANENABLESTATE:
|
||||
# Channel enable state
|
||||
channel, enabled = APTMessage.parse_channel_enable_state(msg)
|
||||
print(f"DEBUG: Received CHANENABLESTATE - parsed channel={channel}, enabled={enabled}, msg={msg.hex()}")
|
||||
|
||||
# Use the channel number from the parsed message
|
||||
if channel in self.channels:
|
||||
self.channels[channel].enabled = enabled
|
||||
print(f"DEBUG: Set channel {channel} enabled={enabled}")
|
||||
|
||||
elif msg_id == MessageID.MGMSG_HW_RESPONSE:
|
||||
# Hardware response (error or acknowledgement)
|
||||
print(f"DEBUG: Received HW_RESPONSE: {msg.hex()}")
|
||||
|
||||
elif msg_id == MessageID.MGMSG_HW_GET_INFO:
|
||||
# Hardware info
|
||||
print(f"DEBUG: Received hardware info")
|
||||
|
||||
except Exception as e:
|
||||
print(f"ERROR: Failed to process message {msg_id:04X}: {e}")
|
||||
|
||||
def _set_and_verify_enable(self, channel: int, enable: bool, retries: int = 3) -> bool:
|
||||
"""
|
||||
Set channel enable state and verify it was set correctly
|
||||
|
||||
Args:
|
||||
channel: Channel number (1, 2, or 3)
|
||||
enable: True to enable, False to disable
|
||||
retries: Number of retry attempts
|
||||
|
||||
Returns:
|
||||
bool: True if value was set and verified
|
||||
"""
|
||||
for attempt in range(retries):
|
||||
# Send enable command
|
||||
cmd = self.protocol.cmd_enable_channel(channel, enable)
|
||||
if not self._send_command(cmd):
|
||||
continue
|
||||
|
||||
time.sleep(0.5) # Wait for controller to process
|
||||
|
||||
# Request channel enable state to verify
|
||||
req_cmd = self.protocol.cmd_req_channel_enable_state(channel)
|
||||
self._send_command(req_cmd)
|
||||
time.sleep(0.5) # Wait for response
|
||||
|
||||
# Check if state matches expected
|
||||
if self.channels[channel].enabled == enable:
|
||||
return True
|
||||
|
||||
if attempt < retries - 1:
|
||||
print(f"DEBUG: Enable verification failed for channel {channel}, "
|
||||
f"retrying ({attempt + 1}/{retries})")
|
||||
time.sleep(0.2)
|
||||
|
||||
print(f"ERROR: Failed to set and verify enable state for channel {channel} "
|
||||
f"after {retries} attempts")
|
||||
return False
|
||||
|
||||
# ==================== Channel Control ====================
|
||||
|
||||
def enable_channel(self, channel: int, enable: bool = True) -> bool:
|
||||
"""
|
||||
Enable or disable a motor channel
|
||||
|
||||
Args:
|
||||
channel: Channel number (1, 2, or 3)
|
||||
enable: True to enable, False to disable
|
||||
|
||||
Returns:
|
||||
bool: True if command sent successfully
|
||||
"""
|
||||
if channel not in [1, 2, 3]:
|
||||
print(f"ERROR: Invalid channel number: {channel}")
|
||||
return False
|
||||
|
||||
action = "Enabling" if enable else "Disabling"
|
||||
print(f"DEBUG: {action} channel {channel}")
|
||||
|
||||
# Use set and verify to ensure command was processed
|
||||
return self._set_and_verify_enable(channel, enable)
|
||||
|
||||
def identify(self, channel: int) -> bool:
|
||||
"""
|
||||
Flash front panel LEDs to identify controller
|
||||
|
||||
Args:
|
||||
channel: Channel number (1, 2, or 3)
|
||||
|
||||
Returns:
|
||||
bool: True if command sent successfully
|
||||
"""
|
||||
print(f"DEBUG: Identifying channel {channel}")
|
||||
cmd = self.protocol.cmd_identify(channel)
|
||||
return self._send_command(cmd)
|
||||
|
||||
# ==================== Homing ====================
|
||||
|
||||
def home_channel(self, channel: int, wait: bool = False, timeout: float = 30.0) -> bool:
|
||||
"""
|
||||
Home a motor channel
|
||||
|
||||
Args:
|
||||
channel: Channel number (1, 2, or 3)
|
||||
wait: If True, block until homing complete
|
||||
timeout: Timeout in seconds if waiting
|
||||
|
||||
Returns:
|
||||
bool: True if homing initiated (or completed if wait=True)
|
||||
"""
|
||||
if channel not in [1, 2, 3]:
|
||||
print(f"ERROR: Invalid channel number: {channel}")
|
||||
return False
|
||||
|
||||
print(f"DEBUG: Homing channel {channel}")
|
||||
|
||||
self.channels[channel].homing = True
|
||||
self.channels[channel].homed = False
|
||||
|
||||
cmd = self.protocol.cmd_move_home(channel)
|
||||
if not self._send_command(cmd):
|
||||
return False
|
||||
|
||||
if wait:
|
||||
# Wait for homing to complete
|
||||
start_time = time.time()
|
||||
while time.time() - start_time < timeout:
|
||||
if self.channels[channel].homed and not self.channels[channel].homing:
|
||||
print(f"INFO: Channel {channel} homing completed")
|
||||
return True
|
||||
time.sleep(0.1)
|
||||
|
||||
print(f"ERROR: Homing timeout for channel {channel}")
|
||||
return False
|
||||
|
||||
return True
|
||||
|
||||
def home_all_channels(self, wait: bool = False, timeout: float = 30.0) -> bool:
|
||||
"""
|
||||
Home all enabled channels
|
||||
|
||||
Args:
|
||||
wait: If True, block until all homing complete
|
||||
timeout: Timeout in seconds if waiting
|
||||
|
||||
Returns:
|
||||
bool: True if all homing operations successful
|
||||
"""
|
||||
success = True
|
||||
for channel in [1, 2, 3]:
|
||||
if self.channels[channel].enabled:
|
||||
if not self.home_channel(channel, wait=False):
|
||||
success = False
|
||||
|
||||
if wait:
|
||||
start_time = time.time()
|
||||
while time.time() - start_time < timeout:
|
||||
all_homed = all(
|
||||
ch.homed for ch in self.channels.values() if ch.enabled
|
||||
)
|
||||
if all_homed:
|
||||
print("INFO: All channels homed successfully")
|
||||
return True
|
||||
time.sleep(0.1)
|
||||
|
||||
print("ERROR: Timeout waiting for all channels to home")
|
||||
return False
|
||||
|
||||
return success
|
||||
|
||||
# ==================== Motion Control ====================
|
||||
|
||||
def move_absolute(self, channel: int, position_mm: float,
|
||||
wait: bool = False, timeout: float = 30.0) -> bool:
|
||||
"""
|
||||
Move to absolute position
|
||||
|
||||
Args:
|
||||
channel: Channel number (1, 2, or 3)
|
||||
position_mm: Target position in mm
|
||||
wait: If True, block until move complete
|
||||
timeout: Timeout in seconds if waiting
|
||||
|
||||
Returns:
|
||||
bool: True if move initiated (or completed if wait=True)
|
||||
"""
|
||||
if channel not in [1, 2, 3]:
|
||||
print(f"ERROR: Invalid channel number: {channel}")
|
||||
return False
|
||||
|
||||
if not self.channels[channel].is_ready():
|
||||
print(f"ERROR: Channel {channel} not ready for movement")
|
||||
return False
|
||||
|
||||
print(f"DEBUG: Moving channel {channel} to {position_mm} mm")
|
||||
|
||||
self.channels[channel].moving = True
|
||||
|
||||
cmd = self.protocol.cmd_move_absolute(channel, position_mm)
|
||||
if not self._send_command(cmd):
|
||||
return False
|
||||
|
||||
if wait:
|
||||
# Wait for move to complete
|
||||
start_time = time.time()
|
||||
while time.time() - start_time < timeout:
|
||||
if not self.channels[channel].moving:
|
||||
print(f"INFO: Channel {channel} move completed")
|
||||
return True
|
||||
time.sleep(0.01)
|
||||
|
||||
print(f"ERROR: Move timeout for channel {channel}")
|
||||
return False
|
||||
|
||||
return True
|
||||
|
||||
def move_relative(self, channel: int, distance_mm: float,
|
||||
wait: bool = False, timeout: float = 30.0) -> bool:
|
||||
"""
|
||||
Move relative distance
|
||||
|
||||
Args:
|
||||
channel: Channel number (1, 2, or 3)
|
||||
distance_mm: Distance to move in mm (positive or negative)
|
||||
wait: If True, block until move complete
|
||||
timeout: Timeout in seconds if waiting
|
||||
|
||||
Returns:
|
||||
bool: True if move initiated (or completed if wait=True)
|
||||
"""
|
||||
if channel not in [1, 2, 3]:
|
||||
print(f"ERROR: Invalid channel number: {channel}")
|
||||
return False
|
||||
|
||||
if not self.channels[channel].is_ready():
|
||||
print(f"ERROR: Channel {channel} not ready for movement")
|
||||
return False
|
||||
|
||||
print(f"DEBUG: Moving channel {channel} by {distance_mm} mm")
|
||||
|
||||
self.channels[channel].moving = True
|
||||
|
||||
cmd = self.protocol.cmd_move_relative(channel, distance_mm)
|
||||
if not self._send_command(cmd):
|
||||
return False
|
||||
|
||||
if wait:
|
||||
# Wait for move to complete
|
||||
start_time = time.time()
|
||||
while time.time() - start_time < timeout:
|
||||
if not self.channels[channel].moving:
|
||||
print(f"INFO: Channel {channel} move completed")
|
||||
return True
|
||||
time.sleep(0.01)
|
||||
|
||||
print(f"ERROR: Move timeout for channel {channel}")
|
||||
return False
|
||||
|
||||
return True
|
||||
|
||||
def stop(self, channel: int, immediate: bool = True) -> bool:
|
||||
"""
|
||||
Stop motion
|
||||
|
||||
Args:
|
||||
channel: Channel number (1, 2, or 3), or 0 for all channels
|
||||
immediate: If True, stop immediately; if False, decelerate
|
||||
|
||||
Returns:
|
||||
bool: True if stop command sent successfully
|
||||
"""
|
||||
if channel == 0:
|
||||
# Stop all channels
|
||||
success = True
|
||||
for ch in [1, 2, 3]:
|
||||
if not self.stop(ch, immediate):
|
||||
success = False
|
||||
return success
|
||||
|
||||
if channel not in [1, 2, 3]:
|
||||
print(f"ERROR: Invalid channel number: {channel}")
|
||||
return False
|
||||
|
||||
print(f"DEBUG: Stopping channel {channel}")
|
||||
|
||||
cmd = self.protocol.cmd_move_stop(channel, immediate)
|
||||
return self._send_command(cmd)
|
||||
|
||||
# ==================== Parameter Setting ====================
|
||||
|
||||
def set_velocity_params(self, channel: int, max_vel_mm_s: float,
|
||||
accel_mm_s2: float) -> bool:
|
||||
"""
|
||||
Set velocity and acceleration parameters
|
||||
|
||||
Args:
|
||||
channel: Channel number (1, 2, or 3)
|
||||
max_vel_mm_s: Maximum velocity in mm/s
|
||||
accel_mm_s2: Acceleration in mm/s²
|
||||
|
||||
Returns:
|
||||
bool: True if parameters set successfully
|
||||
"""
|
||||
if channel not in [1, 2, 3]:
|
||||
print(f"ERROR: Invalid channel number: {channel}")
|
||||
return False
|
||||
|
||||
print(f"DEBUG: Setting velocity params for channel {channel}: "
|
||||
f"vel={max_vel_mm_s} mm/s, accel={accel_mm_s2} mm/s²")
|
||||
|
||||
cmd = self.protocol.cmd_set_velocity_params(channel, max_vel_mm_s, accel_mm_s2)
|
||||
if self._send_command(cmd):
|
||||
time.sleep(0.1) # Wait for controller to process
|
||||
|
||||
# Request status update to confirm parameters were accepted
|
||||
self.request_status_update(channel)
|
||||
time.sleep(0.1) # Wait for status response
|
||||
return True
|
||||
return False
|
||||
|
||||
# ==================== Status and Position ====================
|
||||
|
||||
def get_position(self, channel: int) -> Optional[float]:
|
||||
"""
|
||||
Get current position of channel
|
||||
|
||||
Args:
|
||||
channel: Channel number (1, 2, or 3)
|
||||
|
||||
Returns:
|
||||
float: Current position in mm, or None if unavailable
|
||||
"""
|
||||
if channel not in [1, 2, 3]:
|
||||
return None
|
||||
|
||||
return self.channels[channel].position_mm
|
||||
|
||||
def get_channel_status(self, channel: int) -> Optional[Dict]:
|
||||
"""
|
||||
Get detailed status of channel
|
||||
|
||||
Args:
|
||||
channel: Channel number (1, 2, or 3)
|
||||
|
||||
Returns:
|
||||
dict: Channel status dictionary
|
||||
"""
|
||||
if channel not in [1, 2, 3]:
|
||||
return None
|
||||
|
||||
ch = self.channels[channel]
|
||||
return {
|
||||
'channel': channel,
|
||||
'enabled': ch.enabled,
|
||||
'homed': ch.homed,
|
||||
'homing': ch.homing,
|
||||
'moving': ch.moving,
|
||||
'error': ch.error,
|
||||
'ready': ch.is_ready(),
|
||||
'position_mm': ch.position_mm,
|
||||
'encoder_count': ch.encoder_count,
|
||||
'status_bits': ch.status_bits
|
||||
}
|
||||
|
||||
def request_status_update(self, channel: int) -> bool:
|
||||
"""
|
||||
Request immediate status update for channel
|
||||
|
||||
Args:
|
||||
channel: Channel number (1, 2, or 3)
|
||||
|
||||
Returns:
|
||||
bool: True if request sent successfully
|
||||
"""
|
||||
if channel not in [1, 2, 3]:
|
||||
return False
|
||||
|
||||
cmd = self.protocol.cmd_req_status_update(channel)
|
||||
return self._send_command(cmd)
|
||||
|
||||
# ==================== Callbacks ====================
|
||||
|
||||
def register_move_complete_callback(self, channel: int, callback: Callable):
|
||||
"""Register callback for move complete event"""
|
||||
if channel in [1, 2, 3]:
|
||||
self._move_complete_callbacks[channel].append(callback)
|
||||
|
||||
def register_home_complete_callback(self, channel: int, callback: Callable):
|
||||
"""Register callback for home complete event"""
|
||||
if channel in [1, 2, 3]:
|
||||
self._home_complete_callbacks[channel].append(callback)
|
||||
|
||||
# ==================== Trigger Configuration ====================
|
||||
|
||||
def set_trigger_mode(self, channel: int, mode: int, polarity: int = 0x01,
|
||||
start_pos_fwd: float = 0.0, start_pos_rev: float = 0.0,
|
||||
interval_fwd: float = 0.0, interval_rev: float = 0.0) -> bool:
|
||||
"""
|
||||
Set trigger configuration for a channel
|
||||
|
||||
Args:
|
||||
channel: Channel number (1, 2, or 3)
|
||||
mode: Trigger mode (TriggerMode enum value)
|
||||
polarity: Trigger polarity (0x01 = active high, 0x02 = active low)
|
||||
start_pos_fwd: Start position for forward trigger (mm)
|
||||
start_pos_rev: Start position for reverse trigger (mm)
|
||||
interval_fwd: Interval for forward trigger (mm)
|
||||
interval_rev: Interval for reverse trigger (mm)
|
||||
|
||||
Returns:
|
||||
bool: True if trigger configuration set successfully
|
||||
|
||||
Example:
|
||||
# Disable trigger
|
||||
driver.set_trigger_mode(1, TriggerMode.DISABLED)
|
||||
|
||||
# Enable trigger output on motion
|
||||
driver.set_trigger_mode(1, TriggerMode.OUT_ONLY)
|
||||
|
||||
# Trigger at specific positions
|
||||
driver.set_trigger_mode(1, TriggerMode.OUT_POSITION,
|
||||
start_pos_fwd=10.0, interval_fwd=1.0)
|
||||
"""
|
||||
if channel not in [1, 2, 3]:
|
||||
print(f"ERROR: Invalid channel number: {channel}")
|
||||
return False
|
||||
|
||||
print(f"DEBUG: Setting trigger mode for channel {channel}: mode={mode}")
|
||||
|
||||
cmd = self.protocol.cmd_set_trigger(
|
||||
channel, mode, polarity, start_pos_fwd, start_pos_rev,
|
||||
interval_fwd, interval_rev
|
||||
)
|
||||
|
||||
if self._send_command(cmd):
|
||||
time.sleep(0.1) # Wait for controller to process
|
||||
return True
|
||||
return False
|
||||
|
||||
def get_trigger_config(self, channel: int) -> Optional[Dict]:
|
||||
"""
|
||||
Get current trigger configuration for a channel
|
||||
|
||||
Args:
|
||||
channel: Channel number (1, 2, or 3)
|
||||
|
||||
Returns:
|
||||
dict: Trigger configuration or None if unavailable
|
||||
"""
|
||||
if channel not in [1, 2, 3]:
|
||||
return None
|
||||
|
||||
cmd = self.protocol.cmd_req_trigger(channel)
|
||||
if not self._send_command(cmd):
|
||||
return None
|
||||
|
||||
# Note: In a complete implementation, would wait for response
|
||||
# For now, returning None as response handling would need queue
|
||||
print("WARNING: get_trigger_config not fully implemented (requires response queue)")
|
||||
return None
|
||||
|
||||
# ==================== Digital I/O ====================
|
||||
|
||||
def set_digital_outputs(self, output_bits: int) -> bool:
|
||||
"""
|
||||
Set digital output states
|
||||
|
||||
Args:
|
||||
output_bits: Bit pattern for outputs (0x00 to 0xFF)
|
||||
|
||||
Returns:
|
||||
bool: True if digital outputs set successfully
|
||||
|
||||
Note:
|
||||
Digital outputs share pins with trigger outputs. Ensure
|
||||
trigger mode is disabled before using digital outputs.
|
||||
"""
|
||||
if not (0 <= output_bits <= 0xFF):
|
||||
print(f"ERROR: Invalid output bits: {output_bits}")
|
||||
return False
|
||||
|
||||
print(f"DEBUG: Setting digital outputs: 0x{output_bits:02X}")
|
||||
|
||||
cmd = self.protocol.cmd_set_digital_outputs(output_bits)
|
||||
if self._send_command(cmd):
|
||||
time.sleep(0.05)
|
||||
return True
|
||||
return False
|
||||
@@ -0,0 +1,532 @@
|
||||
"""
|
||||
ThorLabs BBD203 APT Protocol Handler
|
||||
Binary message protocol for BBD203 motor controller
|
||||
|
||||
Copyright (C) 2025 Thomas Ales
|
||||
Licensed under GNU General Public License v2.0
|
||||
"""
|
||||
|
||||
import struct
|
||||
from enum import IntEnum
|
||||
from typing import Tuple, Optional, List
|
||||
|
||||
|
||||
# Message IDs
|
||||
class MessageID(IntEnum):
|
||||
"""APT Protocol Message IDs for BBD203"""
|
||||
|
||||
# Module Control
|
||||
MGMSG_MOD_IDENTIFY = 0x0223
|
||||
MGMSG_MOD_SET_CHANENABLESTATE = 0x0210
|
||||
MGMSG_MOD_REQ_CHANENABLESTATE = 0x0211
|
||||
MGMSG_MOD_GET_CHANENABLESTATE = 0x0212
|
||||
|
||||
# Hardware Control
|
||||
MGMSG_HW_DISCONNECT = 0x0002
|
||||
MGMSG_HW_RESPONSE = 0x0080
|
||||
MGMSG_HW_RICHRESPONSE = 0x0081
|
||||
MGMSG_HW_START_UPDATEMSGS = 0x0011
|
||||
MGMSG_HW_STOP_UPDATEMSGS = 0x0012
|
||||
MGMSG_HW_REQ_INFO = 0x0005
|
||||
MGMSG_HW_GET_INFO = 0x0006
|
||||
|
||||
# Motor Control - Basic
|
||||
MGMSG_MOT_SET_POSCOUNTER = 0x0410
|
||||
MGMSG_MOT_REQ_POSCOUNTER = 0x0411
|
||||
MGMSG_MOT_GET_POSCOUNTER = 0x0412
|
||||
MGMSG_MOT_SET_ENCCOUNTER = 0x0409
|
||||
MGMSG_MOT_REQ_ENCCOUNTER = 0x040A
|
||||
MGMSG_MOT_GET_ENCCOUNTER = 0x040B
|
||||
|
||||
# Motor Control - Homing
|
||||
MGMSG_MOT_SET_HOMEPARAMS = 0x0440
|
||||
MGMSG_MOT_REQ_HOMEPARAMS = 0x0441
|
||||
MGMSG_MOT_GET_HOMEPARAMS = 0x0442
|
||||
MGMSG_MOT_MOVE_HOME = 0x0443
|
||||
MGMSG_MOT_MOVE_HOMED = 0x0444
|
||||
|
||||
# Motor Control - Movement
|
||||
MGMSG_MOT_SET_MOVERELPARAMS = 0x0445
|
||||
MGMSG_MOT_REQ_MOVERELPARAMS = 0x0446
|
||||
MGMSG_MOT_GET_MOVERELPARAMS = 0x0447
|
||||
MGMSG_MOT_MOVE_RELATIVE = 0x0448
|
||||
MGMSG_MOT_SET_MOVEABSPARAMS = 0x0450
|
||||
MGMSG_MOT_REQ_MOVEABSPARAMS = 0x0451
|
||||
MGMSG_MOT_GET_MOVEABSPARAMS = 0x0452
|
||||
MGMSG_MOT_MOVE_ABSOLUTE = 0x0453
|
||||
MGMSG_MOT_MOVE_COMPLETED = 0x0464
|
||||
MGMSG_MOT_MOVE_VELOCITY = 0x0457
|
||||
MGMSG_MOT_MOVE_STOP = 0x0465
|
||||
MGMSG_MOT_MOVE_STOPPED = 0x0466
|
||||
|
||||
# Motor Control - Velocity
|
||||
MGMSG_MOT_SET_VELPARAMS = 0x0413
|
||||
MGMSG_MOT_REQ_VELPARAMS = 0x0414
|
||||
MGMSG_MOT_GET_VELPARAMS = 0x0415
|
||||
|
||||
# Motor Control - Status
|
||||
MGMSG_MOT_REQ_STATUSUPDATE = 0x0480
|
||||
MGMSG_MOT_GET_STATUSUPDATE = 0x0481
|
||||
MGMSG_MOT_REQ_STATUSBITS = 0x0429
|
||||
MGMSG_MOT_GET_STATUSBITS = 0x042A
|
||||
|
||||
# Digital I/O and Trigger
|
||||
MGMSG_RACK_SET_DIGOUTPUTS = 0x0228
|
||||
MGMSG_RACK_REQ_DIGOUTPUTS = 0x0229
|
||||
MGMSG_RACK_GET_DIGOUTPUTS = 0x0230
|
||||
MGMSG_MOT_SET_TRIGGER = 0x0500
|
||||
MGMSG_MOT_REQ_TRIGGER = 0x0501
|
||||
MGMSG_MOT_GET_TRIGGER = 0x0502
|
||||
|
||||
|
||||
# Destination addresses
|
||||
class Destination(IntEnum):
|
||||
"""BBD203 Destination addresses"""
|
||||
USB = 0x50
|
||||
ALL_CHANNELS = 0x11
|
||||
CHANNEL_1 = 0x21
|
||||
CHANNEL_2 = 0x22
|
||||
CHANNEL_3 = 0x23
|
||||
|
||||
|
||||
# Source addresses
|
||||
class Source(IntEnum):
|
||||
"""Source addresses"""
|
||||
HOST = 0x01
|
||||
|
||||
|
||||
# Status bits
|
||||
class StatusBits(IntEnum):
|
||||
"""Motor status bit definitions"""
|
||||
HOMING = 0x00000200
|
||||
HOMED = 0x00000400
|
||||
TRACKING = 0x00001000
|
||||
SETTLED = 0x00002000
|
||||
MOTION_ERROR = 0x00004000
|
||||
MOTOR_ENABLED = 0x80000000
|
||||
FORWARD_LIMIT = 0x00000001
|
||||
REVERSE_LIMIT = 0x00000002
|
||||
IN_MOTION_FORWARD = 0x00000010
|
||||
IN_MOTION_REVERSE = 0x00000020
|
||||
JOGGING_FORWARD = 0x00000040
|
||||
JOGGING_REVERSE = 0x00000080
|
||||
|
||||
|
||||
# Trigger modes
|
||||
class TriggerMode(IntEnum):
|
||||
"""Trigger mode definitions"""
|
||||
DISABLED = 0x00
|
||||
IN_OUT_RELATIVE_MOVE = 0x01
|
||||
IN_OUT_ABSOLUTE_MOVE = 0x02
|
||||
IN_OUT_HOME = 0x03
|
||||
IN_OUT_STOP = 0x04
|
||||
OUT_ONLY = 0x10
|
||||
OUT_POSITION = 0x11
|
||||
|
||||
|
||||
class APTMessage:
|
||||
"""
|
||||
APT Protocol Message Builder and Parser
|
||||
Handles construction and parsing of binary APT messages
|
||||
"""
|
||||
|
||||
@staticmethod
|
||||
def build_header_only(msg_id: int, param1: int, param2: int,
|
||||
dest: int, source: int = Source.HOST) -> bytes:
|
||||
"""
|
||||
Build a 6-byte header-only message
|
||||
|
||||
Args:
|
||||
msg_id: Message ID (16-bit)
|
||||
param1: Parameter 1 (8-bit)
|
||||
param2: Parameter 2 (8-bit)
|
||||
dest: Destination address
|
||||
source: Source address (default: HOST)
|
||||
|
||||
Returns:
|
||||
bytes: 6-byte message
|
||||
"""
|
||||
return struct.pack('<HBBBB', msg_id, param1, param2, dest, source)
|
||||
|
||||
@staticmethod
|
||||
def build_with_data(msg_id: int, dest: int, data: bytes,
|
||||
source: int = Source.HOST) -> bytes:
|
||||
"""
|
||||
Build a message with data packet
|
||||
|
||||
Args:
|
||||
msg_id: Message ID (16-bit)
|
||||
dest: Destination address
|
||||
data: Data packet bytes
|
||||
source: Source address (default: HOST)
|
||||
|
||||
Returns:
|
||||
bytes: Complete message (header + data)
|
||||
"""
|
||||
data_len = len(data)
|
||||
header = struct.pack('<HHBB', msg_id, data_len, dest, source)
|
||||
return header + data
|
||||
|
||||
@staticmethod
|
||||
def parse_header(data: bytes) -> Tuple[int, int, int, int, int]:
|
||||
"""
|
||||
Parse message header
|
||||
|
||||
Args:
|
||||
data: At least 6 bytes of message data
|
||||
|
||||
Returns:
|
||||
tuple: (msg_id, data_len, dest, source, has_data)
|
||||
"""
|
||||
if len(data) < 6:
|
||||
raise ValueError("Insufficient data for header")
|
||||
|
||||
msg_id, byte2, byte3, dest, source = struct.unpack('<HBBBB', data[:6])
|
||||
|
||||
# Determine if this is header-only or has data
|
||||
# Header-only messages use bytes 2-3 as parameters
|
||||
# Messages with data use bytes 2-3 as data length
|
||||
data_len = (byte3 << 8) | byte2
|
||||
|
||||
return msg_id, data_len, dest, source
|
||||
|
||||
@staticmethod
|
||||
def parse_position_counter(data: bytes) -> Tuple[int, int]:
|
||||
"""Parse MGMSG_MOT_GET_POSCOUNTER response"""
|
||||
if len(data) < 12:
|
||||
raise ValueError("Insufficient data for position counter")
|
||||
|
||||
_, channel, position = struct.unpack('<HHI', data[6:12])
|
||||
return channel, position
|
||||
|
||||
@staticmethod
|
||||
def parse_encoder_counter(data: bytes) -> Tuple[int, int]:
|
||||
"""Parse MGMSG_MOT_GET_ENCCOUNTER response"""
|
||||
if len(data) < 12:
|
||||
raise ValueError("Insufficient data for encoder counter")
|
||||
|
||||
_, channel, encoder = struct.unpack('<HHI', data[6:12])
|
||||
return channel, encoder
|
||||
|
||||
@staticmethod
|
||||
def parse_status_update(data: bytes) -> Tuple[int, int, int, int]:
|
||||
"""
|
||||
Parse MGMSG_MOT_GET_STATUSUPDATE response
|
||||
|
||||
Returns:
|
||||
tuple: (channel, position, enc_count, status_bits)
|
||||
"""
|
||||
if len(data) < 20:
|
||||
raise ValueError("Insufficient data for status update")
|
||||
|
||||
# Skip 6-byte header, parse data packet
|
||||
channel, position, enc_count, status = struct.unpack('<HIII', data[6:20])
|
||||
return channel, position, enc_count, status
|
||||
|
||||
@staticmethod
|
||||
def parse_velocity_params(data: bytes) -> Tuple[int, int, int, int]:
|
||||
"""
|
||||
Parse MGMSG_MOT_GET_VELPARAMS response
|
||||
|
||||
Returns:
|
||||
tuple: (channel, min_vel, max_vel, accel)
|
||||
"""
|
||||
if len(data) < 20:
|
||||
raise ValueError("Insufficient data for velocity params")
|
||||
|
||||
channel, min_vel, max_vel, accel = struct.unpack('<HIII', data[6:20])
|
||||
return channel, min_vel, max_vel, accel
|
||||
|
||||
@staticmethod
|
||||
def parse_channel_enable_state(data: bytes) -> Tuple[int, bool]:
|
||||
"""Parse MGMSG_MOD_GET_CHANENABLESTATE response"""
|
||||
if len(data) < 6:
|
||||
raise ValueError("Insufficient data for channel enable state")
|
||||
|
||||
# Header only message, params in bytes 2-3
|
||||
_, enable_state, channel, _, _ = struct.unpack('<HBBBB', data[:6])
|
||||
return channel, (enable_state == 0x01)
|
||||
|
||||
@staticmethod
|
||||
def parse_trigger_config(data: bytes) -> Tuple[int, int, int, int, int, int, int]:
|
||||
"""
|
||||
Parse MGMSG_MOT_GET_TRIGGER response
|
||||
|
||||
Returns:
|
||||
tuple: (channel, trigger_mode, polarity, start_pos_fwd, start_pos_rev,
|
||||
interval_fwd, interval_rev, num_pulses, pulse_width, num_cycles)
|
||||
"""
|
||||
if len(data) < 28:
|
||||
raise ValueError("Insufficient data for trigger config")
|
||||
|
||||
# Parse data packet (22 bytes starting at byte 6)
|
||||
channel, mode, polarity, start_fwd, start_rev, interval_fwd, interval_rev = \
|
||||
struct.unpack('<HBBIIIi', data[6:28])
|
||||
|
||||
# Extended parameters if available
|
||||
num_pulses = 0
|
||||
pulse_width = 0
|
||||
num_cycles = 0
|
||||
if len(data) >= 40:
|
||||
num_pulses, pulse_width, num_cycles = struct.unpack('<III', data[28:40])
|
||||
|
||||
return (channel, mode, polarity, start_fwd, start_rev,
|
||||
interval_fwd, interval_rev, num_pulses, pulse_width, num_cycles)
|
||||
|
||||
@staticmethod
|
||||
def parse_digital_outputs(data: bytes) -> Tuple[int, int]:
|
||||
"""Parse MGMSG_RACK_GET_DIGOUTPUTS response"""
|
||||
if len(data) < 6:
|
||||
raise ValueError("Insufficient data for digital outputs")
|
||||
|
||||
# Header only message, params in bytes 2-3
|
||||
_, output_state, _, _, _ = struct.unpack('<HBBBB', data[:6])
|
||||
return output_state
|
||||
|
||||
|
||||
class APTProtocol:
|
||||
"""
|
||||
High-level APT Protocol interface for BBD203
|
||||
Provides methods to build common command messages
|
||||
"""
|
||||
|
||||
# Scaling constants
|
||||
T_SAMPLE = 102.4e-6 # Controller sample time
|
||||
VELOCITY_SCALE = int(T_SAMPLE * 65536)
|
||||
ACCEL_SCALE = int((T_SAMPLE ** 2) * 65536)
|
||||
|
||||
def __init__(self, encoder_counts_per_mm: int = 20000):
|
||||
"""
|
||||
Initialize APT Protocol handler
|
||||
|
||||
Args:
|
||||
encoder_counts_per_mm: Encoder resolution (default: 20000 for MLS203)
|
||||
"""
|
||||
self.enc_cnt = encoder_counts_per_mm
|
||||
|
||||
def position_to_apt(self, pos_mm: float) -> int:
|
||||
"""Convert position in mm to APT units"""
|
||||
return int(pos_mm * self.enc_cnt)
|
||||
|
||||
def apt_to_position(self, apt_units: int) -> float:
|
||||
"""Convert APT units to position in mm"""
|
||||
return apt_units / self.enc_cnt
|
||||
|
||||
def velocity_to_apt(self, vel_mm_s: float) -> int:
|
||||
"""Convert velocity in mm/s to APT units"""
|
||||
return int(self.enc_cnt * self.T_SAMPLE * 65536 * vel_mm_s)
|
||||
|
||||
def apt_to_velocity(self, apt_units: int) -> float:
|
||||
"""Convert APT units to velocity in mm/s"""
|
||||
return apt_units / (self.enc_cnt * self.T_SAMPLE * 65536)
|
||||
|
||||
def accel_to_apt(self, accel_mm_s2: float) -> int:
|
||||
"""Convert acceleration in mm/s² to APT units"""
|
||||
return int(self.enc_cnt * (self.T_SAMPLE ** 2) * 65536 * accel_mm_s2)
|
||||
|
||||
def apt_to_accel(self, apt_units: int) -> float:
|
||||
"""Convert APT units to acceleration in mm/s²"""
|
||||
return apt_units / (self.enc_cnt * (self.T_SAMPLE ** 2) * 65536)
|
||||
|
||||
# Command builders
|
||||
|
||||
def cmd_identify(self, channel: int) -> bytes:
|
||||
"""Build identify command (flash LEDs)"""
|
||||
dest = Destination.CHANNEL_1 + (channel - 1)
|
||||
return APTMessage.build_header_only(
|
||||
MessageID.MGMSG_MOD_IDENTIFY, 0x00, 0x00, dest
|
||||
)
|
||||
|
||||
def cmd_enable_channel(self, channel: int, enable: bool = True) -> bytes:
|
||||
"""Build enable/disable channel command"""
|
||||
dest = Destination.CHANNEL_1 + (channel - 1)
|
||||
state = 0x01 if enable else 0x02
|
||||
return APTMessage.build_header_only(
|
||||
MessageID.MGMSG_MOD_SET_CHANENABLESTATE, state, channel, dest
|
||||
)
|
||||
|
||||
def cmd_req_channel_enable_state(self, channel: int) -> bytes:
|
||||
"""Build request channel enable state command"""
|
||||
dest = Destination.CHANNEL_1 + (channel - 1)
|
||||
return APTMessage.build_header_only(
|
||||
MessageID.MGMSG_MOD_REQ_CHANENABLESTATE, 0x01, 0x00, dest
|
||||
)
|
||||
|
||||
def cmd_start_update_msgs(self) -> bytes:
|
||||
"""Build start automatic status updates command"""
|
||||
return APTMessage.build_header_only(
|
||||
MessageID.MGMSG_HW_START_UPDATEMSGS, 0x00, 0x00, Destination.USB
|
||||
)
|
||||
|
||||
def cmd_stop_update_msgs(self) -> bytes:
|
||||
"""Build stop automatic status updates command"""
|
||||
return APTMessage.build_header_only(
|
||||
MessageID.MGMSG_HW_STOP_UPDATEMSGS, 0x00, 0x00, Destination.USB
|
||||
)
|
||||
|
||||
def cmd_req_hw_info(self) -> bytes:
|
||||
"""Build request hardware info command"""
|
||||
return APTMessage.build_header_only(
|
||||
MessageID.MGMSG_HW_REQ_INFO, 0x00, 0x00, Destination.USB
|
||||
)
|
||||
|
||||
def cmd_move_home(self, channel: int) -> bytes:
|
||||
"""Build move home command"""
|
||||
dest = Destination.CHANNEL_1 + (channel - 1)
|
||||
return APTMessage.build_header_only(
|
||||
MessageID.MGMSG_MOT_MOVE_HOME, 0x01, 0x00, dest
|
||||
)
|
||||
|
||||
def cmd_move_absolute(self, channel: int, position_mm: float) -> bytes:
|
||||
"""Build move absolute command"""
|
||||
dest = Destination.CHANNEL_1 + (channel - 1)
|
||||
pos_apt = self.position_to_apt(position_mm)
|
||||
data = struct.pack('<HI', channel, pos_apt)
|
||||
return APTMessage.build_with_data(
|
||||
MessageID.MGMSG_MOT_MOVE_ABSOLUTE, dest, data
|
||||
)
|
||||
|
||||
def cmd_move_relative(self, channel: int, distance_mm: float) -> bytes:
|
||||
"""Build move relative command"""
|
||||
dest = Destination.CHANNEL_1 + (channel - 1)
|
||||
dist_apt = self.position_to_apt(distance_mm)
|
||||
data = struct.pack('<Hi', channel, dist_apt)
|
||||
return APTMessage.build_with_data(
|
||||
MessageID.MGMSG_MOT_MOVE_RELATIVE, dest, data
|
||||
)
|
||||
|
||||
def cmd_move_stop(self, channel: int, immediate: bool = True) -> bytes:
|
||||
"""Build stop motion command"""
|
||||
dest = Destination.CHANNEL_1 + (channel - 1)
|
||||
stop_mode = 0x01 if immediate else 0x02
|
||||
return APTMessage.build_header_only(
|
||||
MessageID.MGMSG_MOT_MOVE_STOP, 0x01, stop_mode, dest
|
||||
)
|
||||
|
||||
def cmd_set_velocity_params(self, channel: int, max_vel_mm_s: float,
|
||||
accel_mm_s2: float) -> bytes:
|
||||
"""Build set velocity parameters command"""
|
||||
dest = Destination.CHANNEL_1 + (channel - 1)
|
||||
max_vel_apt = self.velocity_to_apt(max_vel_mm_s)
|
||||
accel_apt = self.accel_to_apt(accel_mm_s2)
|
||||
|
||||
data = struct.pack('<HIII',
|
||||
channel, # Channel number
|
||||
0, # Min velocity (0)
|
||||
max_vel_apt, # Max velocity
|
||||
accel_apt # Acceleration
|
||||
)
|
||||
return APTMessage.build_with_data(
|
||||
MessageID.MGMSG_MOT_SET_VELPARAMS, dest, data
|
||||
)
|
||||
|
||||
def cmd_req_velocity_params(self, channel: int) -> bytes:
|
||||
"""Build request velocity parameters command"""
|
||||
dest = Destination.CHANNEL_1 + (channel - 1)
|
||||
return APTMessage.build_header_only(
|
||||
MessageID.MGMSG_MOT_REQ_VELPARAMS, 0x01, 0x00, dest
|
||||
)
|
||||
|
||||
def cmd_req_position(self, channel: int) -> bytes:
|
||||
"""Build request position counter command"""
|
||||
dest = Destination.CHANNEL_1 + (channel - 1)
|
||||
return APTMessage.build_header_only(
|
||||
MessageID.MGMSG_MOT_REQ_POSCOUNTER, 0x01, 0x00, dest
|
||||
)
|
||||
|
||||
def cmd_req_encoder(self, channel: int) -> bytes:
|
||||
"""Build request encoder counter command"""
|
||||
dest = Destination.CHANNEL_1 + (channel - 1)
|
||||
return APTMessage.build_header_only(
|
||||
MessageID.MGMSG_MOT_REQ_ENCCOUNTER, 0x01, 0x00, dest
|
||||
)
|
||||
|
||||
def cmd_req_status_update(self, channel: int) -> bytes:
|
||||
"""Build request status update command"""
|
||||
dest = Destination.CHANNEL_1 + (channel - 1)
|
||||
return APTMessage.build_header_only(
|
||||
MessageID.MGMSG_MOT_REQ_STATUSUPDATE, 0x01, 0x00, dest
|
||||
)
|
||||
|
||||
def cmd_req_status_bits(self, channel: int) -> bytes:
|
||||
"""Build request status bits command"""
|
||||
dest = Destination.CHANNEL_1 + (channel - 1)
|
||||
return APTMessage.build_header_only(
|
||||
MessageID.MGMSG_MOT_REQ_STATUSBITS, 0x01, 0x00, dest
|
||||
)
|
||||
|
||||
def cmd_set_position_counter(self, channel: int, position_mm: float) -> bytes:
|
||||
"""Build set position counter command"""
|
||||
dest = Destination.CHANNEL_1 + (channel - 1)
|
||||
pos_apt = self.position_to_apt(position_mm)
|
||||
data = struct.pack('<HI', channel, pos_apt)
|
||||
return APTMessage.build_with_data(
|
||||
MessageID.MGMSG_MOT_SET_POSCOUNTER, dest, data
|
||||
)
|
||||
|
||||
def cmd_set_trigger(self, channel: int, mode: int, polarity: int = 0x01,
|
||||
start_pos_fwd: float = 0.0, start_pos_rev: float = 0.0,
|
||||
interval_fwd: float = 0.0, interval_rev: float = 0.0) -> bytes:
|
||||
"""
|
||||
Build set trigger configuration command
|
||||
|
||||
Args:
|
||||
channel: Channel number (1, 2, or 3)
|
||||
mode: Trigger mode (TriggerMode enum value)
|
||||
polarity: Trigger polarity (0x01 = active high, 0x02 = active low)
|
||||
start_pos_fwd: Start position for forward trigger (mm)
|
||||
start_pos_rev: Start position for reverse trigger (mm)
|
||||
interval_fwd: Interval for forward trigger (mm)
|
||||
interval_rev: Interval for reverse trigger (mm)
|
||||
|
||||
Returns:
|
||||
bytes: Complete trigger configuration command
|
||||
"""
|
||||
dest = Destination.CHANNEL_1 + (channel - 1)
|
||||
|
||||
# Convert positions to APT units
|
||||
start_fwd_apt = self.position_to_apt(start_pos_fwd)
|
||||
start_rev_apt = self.position_to_apt(start_pos_rev)
|
||||
interval_fwd_apt = self.position_to_apt(interval_fwd)
|
||||
interval_rev_apt = int(self.position_to_apt(interval_rev)) # Signed
|
||||
|
||||
data = struct.pack('<HBBIIIi',
|
||||
channel, # Channel number
|
||||
mode, # Trigger mode
|
||||
polarity, # Polarity
|
||||
start_fwd_apt, # Start position forward
|
||||
start_rev_apt, # Start position reverse
|
||||
interval_fwd_apt, # Interval forward
|
||||
interval_rev_apt # Interval reverse (signed)
|
||||
)
|
||||
|
||||
return APTMessage.build_with_data(
|
||||
MessageID.MGMSG_MOT_SET_TRIGGER, dest, data
|
||||
)
|
||||
|
||||
def cmd_req_trigger(self, channel: int) -> bytes:
|
||||
"""Build request trigger configuration command"""
|
||||
dest = Destination.CHANNEL_1 + (channel - 1)
|
||||
return APTMessage.build_header_only(
|
||||
MessageID.MGMSG_MOT_REQ_TRIGGER, 0x01, 0x00, dest
|
||||
)
|
||||
|
||||
def cmd_set_digital_outputs(self, output_bits: int) -> bytes:
|
||||
"""
|
||||
Build set digital outputs command
|
||||
|
||||
Args:
|
||||
output_bits: Bit pattern for digital outputs (0x00 to 0xFF)
|
||||
|
||||
Returns:
|
||||
bytes: Digital output command
|
||||
"""
|
||||
return APTMessage.build_header_only(
|
||||
MessageID.MGMSG_RACK_SET_DIGOUTPUTS, output_bits, 0x00, Destination.USB
|
||||
)
|
||||
|
||||
def cmd_req_digital_outputs(self) -> bytes:
|
||||
"""Build request digital outputs command"""
|
||||
return APTMessage.build_header_only(
|
||||
MessageID.MGMSG_RACK_REQ_DIGOUTPUTS, 0x00, 0x00, Destination.USB
|
||||
)
|
||||
@@ -0,0 +1,627 @@
|
||||
"""
|
||||
Helios Laser Driver
|
||||
Complete RS-232 driver for Helios laser systems
|
||||
|
||||
Copyright (C) 2025 Thomas Ales
|
||||
Licensed under GNU General Public License v2.0
|
||||
"""
|
||||
|
||||
import serial
|
||||
import serial.tools.list_ports
|
||||
import time
|
||||
import threading
|
||||
from typing import Dict, List, Optional, Tuple
|
||||
|
||||
from hardware.helios_protocol import (
|
||||
HeliosProtocol, HeliosCommand, HeliosStatus, PulseMode
|
||||
)
|
||||
|
||||
|
||||
class HeliosDriver:
|
||||
"""
|
||||
Complete driver for Helios laser system
|
||||
|
||||
Features:
|
||||
- RS-232 communication at 9600 baud
|
||||
- All protocol commands supported
|
||||
- Thread-safe operation
|
||||
- Temperature monitoring
|
||||
- Status monitoring
|
||||
- Power monitoring
|
||||
"""
|
||||
|
||||
# RS-232 Settings (from protocol document)
|
||||
BAUDRATE = 9600
|
||||
DATABITS = 8
|
||||
PARITY = 'N'
|
||||
STOPBITS = 1
|
||||
|
||||
def __init__(self, timeout: float = 2.0):
|
||||
"""
|
||||
Initialize Helios driver
|
||||
|
||||
Args:
|
||||
timeout: Serial communication timeout in seconds
|
||||
"""
|
||||
self.protocol = HeliosProtocol()
|
||||
self.timeout = timeout
|
||||
|
||||
# Serial connection
|
||||
self._serial: Optional[serial.Serial] = None
|
||||
self._port_name = ""
|
||||
self._connected = False
|
||||
|
||||
# Communication lock for thread safety
|
||||
self._comm_lock = threading.Lock()
|
||||
|
||||
# Cached state
|
||||
self._laser_enabled = False
|
||||
self._pulse_mode = PulseMode.CONTINUOUS_PULSING
|
||||
self._frequency_hz = 10000.0
|
||||
self._current_ma = 500.0
|
||||
self._controller_serial = ""
|
||||
self._head_serial = ""
|
||||
|
||||
# Temperature monitoring (in °C)
|
||||
self._pump_temp_c = 0.0
|
||||
self._resonator_temp_c = 0.0
|
||||
self._qswitch_temp_c = 0.0
|
||||
self._power_stage_temp_c = 0.0
|
||||
|
||||
# Status
|
||||
self._status = HeliosStatus(0)
|
||||
self._operation_hours = 0.0
|
||||
self._power_mw = 0.0
|
||||
|
||||
print("INFO: Helios Laser driver initialized")
|
||||
|
||||
# ==================== Connection Management ====================
|
||||
|
||||
@staticmethod
|
||||
def list_available_ports() -> List[str]:
|
||||
"""
|
||||
List available serial ports
|
||||
|
||||
Returns:
|
||||
list: Available port names
|
||||
"""
|
||||
ports = serial.tools.list_ports.comports()
|
||||
return [port.device for port in ports]
|
||||
|
||||
def connect(self, port: str) -> bool:
|
||||
"""
|
||||
Connect to Helios laser
|
||||
|
||||
Args:
|
||||
port: Serial port name (e.g., 'COM3', '/dev/ttyUSB0')
|
||||
|
||||
Returns:
|
||||
bool: True if connection successful
|
||||
"""
|
||||
try:
|
||||
print(f"INFO: Connecting to Helios laser on {port}")
|
||||
|
||||
self._serial = serial.Serial(
|
||||
port=port,
|
||||
baudrate=self.BAUDRATE,
|
||||
bytesize=self.DATABITS,
|
||||
parity=self.PARITY,
|
||||
stopbits=self.STOPBITS,
|
||||
timeout=self.timeout
|
||||
)
|
||||
|
||||
self._port_name = port
|
||||
self._connected = True
|
||||
|
||||
# Read serial numbers
|
||||
time.sleep(0.5)
|
||||
self._controller_serial = self.query_controller_serial()
|
||||
time.sleep(0.5)
|
||||
self._head_serial = self.query_head_serial()
|
||||
time.sleep(0.5)
|
||||
|
||||
# Read initial state
|
||||
self._update_cached_state()
|
||||
|
||||
print(f"INFO: Connected to Helios laser on {port}")
|
||||
print(f" Controller S/N: {self._controller_serial}")
|
||||
print(f" Head S/N: {self._head_serial}")
|
||||
return True
|
||||
|
||||
except serial.SerialException as e:
|
||||
print(f"ERROR: Failed to connect to {port}: {e}")
|
||||
self._connected = False
|
||||
return False
|
||||
|
||||
def disconnect(self) -> bool:
|
||||
"""
|
||||
Disconnect from Helios laser
|
||||
|
||||
Returns:
|
||||
bool: True if disconnection successful
|
||||
"""
|
||||
if not self._connected:
|
||||
return True
|
||||
|
||||
try:
|
||||
print("INFO: Disconnecting from Helios laser")
|
||||
|
||||
# Turn off laser before disconnecting
|
||||
self.set_laser_enable(False)
|
||||
time.sleep(0.5)
|
||||
|
||||
# Close serial port
|
||||
if self._serial and self._serial.is_open:
|
||||
self._serial.close()
|
||||
|
||||
self._connected = False
|
||||
print("INFO: Disconnected from Helios laser")
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
print(f"ERROR: Error during disconnect: {e}")
|
||||
return False
|
||||
|
||||
def is_connected(self) -> bool:
|
||||
"""Check if laser is connected"""
|
||||
return self._connected and self._serial and self._serial.is_open
|
||||
|
||||
# ==================== Communication Methods ====================
|
||||
|
||||
def _send_command(self, command: bytes) -> bool:
|
||||
"""
|
||||
Send command to laser (no response expected)
|
||||
|
||||
Args:
|
||||
command: Command bytes to send
|
||||
|
||||
Returns:
|
||||
bool: True if send successful
|
||||
"""
|
||||
if not self.is_connected():
|
||||
print("ERROR: Cannot send command - not connected")
|
||||
return False
|
||||
|
||||
try:
|
||||
with self._comm_lock:
|
||||
self._serial.write(command)
|
||||
self._serial.flush()
|
||||
return True
|
||||
except serial.SerialException as e:
|
||||
print(f"ERROR: Failed to send command: {e}")
|
||||
return False
|
||||
|
||||
def _query(self, command: bytes) -> Optional[str]:
|
||||
"""
|
||||
Send query and read response
|
||||
|
||||
Args:
|
||||
command: Query command bytes
|
||||
|
||||
Returns:
|
||||
str: Response string, or None if error
|
||||
"""
|
||||
if not self.is_connected():
|
||||
print("ERROR: Cannot query - not connected")
|
||||
return None
|
||||
|
||||
try:
|
||||
with self._comm_lock:
|
||||
# Clear input buffer
|
||||
self._serial.reset_input_buffer()
|
||||
|
||||
# Send query
|
||||
self._serial.write(command)
|
||||
self._serial.flush()
|
||||
|
||||
# Read response (terminated by CR)
|
||||
response = self._serial.read_until(b'\r')
|
||||
|
||||
if not response:
|
||||
print("ERROR: No response from laser")
|
||||
return None
|
||||
|
||||
return HeliosCommand.parse_response(response)
|
||||
|
||||
except serial.SerialException as e:
|
||||
print(f"ERROR: Query failed: {e}")
|
||||
return None
|
||||
|
||||
def _set_and_verify(self, set_cmd: bytes, query_cmd: bytes,
|
||||
expected_value: str, retries: int = 3) -> bool:
|
||||
"""
|
||||
Set a value and verify it was set correctly
|
||||
|
||||
Args:
|
||||
set_cmd: Command to set value
|
||||
query_cmd: Command to query value
|
||||
expected_value: Expected response
|
||||
retries: Number of retry attempts
|
||||
|
||||
Returns:
|
||||
bool: True if value was set and verified
|
||||
"""
|
||||
for attempt in range(retries):
|
||||
# Send set command
|
||||
if not self._send_command(set_cmd):
|
||||
continue
|
||||
|
||||
time.sleep(0.5) # Wait 500ms for laser to process (per documentation)
|
||||
|
||||
# Query to verify
|
||||
response = self._query(query_cmd)
|
||||
if response and response == expected_value:
|
||||
return True
|
||||
|
||||
if attempt < retries - 1:
|
||||
print(f"DEBUG: Verification failed, retrying ({attempt + 1}/{retries})")
|
||||
time.sleep(0.5)
|
||||
|
||||
print(f"ERROR: Failed to set and verify value after {retries} attempts")
|
||||
return False
|
||||
|
||||
# ==================== Laser Control ====================
|
||||
|
||||
def set_laser_enable(self, enabled: bool) -> bool:
|
||||
"""
|
||||
Enable or disable laser
|
||||
|
||||
Args:
|
||||
enabled: True to enable, False to disable
|
||||
|
||||
Returns:
|
||||
bool: True if command successful
|
||||
"""
|
||||
print(f"DEBUG: {'Enabling' if enabled else 'Disabling'} laser")
|
||||
|
||||
cmd = self.protocol.cmd_set_laser_enable(enabled)
|
||||
query_cmd = self.protocol.cmd_query_laser_enable()
|
||||
expected = "1" if enabled else "0"
|
||||
|
||||
if self._set_and_verify(cmd, query_cmd, expected):
|
||||
self._laser_enabled = enabled
|
||||
return True
|
||||
return False
|
||||
|
||||
def is_laser_enabled(self) -> bool:
|
||||
"""Check if laser is currently enabled"""
|
||||
return self._laser_enabled
|
||||
|
||||
def query_laser_enable(self) -> bool:
|
||||
"""Query laser enable state from hardware"""
|
||||
cmd = self.protocol.cmd_query_laser_enable()
|
||||
response = self._query(cmd)
|
||||
if response:
|
||||
self._laser_enabled = (response == "1")
|
||||
return self._laser_enabled
|
||||
return False
|
||||
|
||||
# ==================== Pulse Mode Control ====================
|
||||
|
||||
def set_pulse_mode(self, mode: PulseMode) -> bool:
|
||||
"""
|
||||
Set pulse mode
|
||||
|
||||
Args:
|
||||
mode: PulseMode enum value
|
||||
|
||||
Returns:
|
||||
bool: True if command successful
|
||||
"""
|
||||
print(f"DEBUG: Setting pulse mode to {mode.name}")
|
||||
|
||||
cmd = self.protocol.cmd_set_pulse_mode(mode)
|
||||
query_cmd = self.protocol.cmd_query_pulse_mode()
|
||||
expected = str(mode.value)
|
||||
|
||||
if self._set_and_verify(cmd, query_cmd, expected):
|
||||
self._pulse_mode = mode
|
||||
return True
|
||||
return False
|
||||
|
||||
def get_pulse_mode(self) -> PulseMode:
|
||||
"""Get current pulse mode"""
|
||||
return self._pulse_mode
|
||||
|
||||
# ==================== Frequency Control ====================
|
||||
|
||||
def set_frequency_hz(self, freq_hz: float) -> bool:
|
||||
"""
|
||||
Set laser frequency in Hz
|
||||
|
||||
Args:
|
||||
freq_hz: Frequency in Hz (16.7 kHz to 125 kHz)
|
||||
|
||||
Returns:
|
||||
bool: True if command successful
|
||||
"""
|
||||
print(f"DEBUG: Setting laser frequency to {freq_hz} Hz")
|
||||
|
||||
try:
|
||||
cmd = self.protocol.cmd_set_frequency_hz(freq_hz)
|
||||
period_ns = self.protocol.frequency_to_period_ns(freq_hz)
|
||||
query_cmd = self.protocol.cmd_query_frequency()
|
||||
expected = str(period_ns)
|
||||
|
||||
if self._set_and_verify(cmd, query_cmd, expected):
|
||||
self._frequency_hz = freq_hz
|
||||
return True
|
||||
except ValueError as e:
|
||||
print(f"ERROR: {e}")
|
||||
return False
|
||||
|
||||
def get_frequency_hz(self) -> float:
|
||||
"""Get current frequency in Hz"""
|
||||
return self._frequency_hz
|
||||
|
||||
def query_frequency_hz(self) -> Optional[float]:
|
||||
"""Query frequency from hardware (returns Hz)"""
|
||||
cmd = self.protocol.cmd_query_frequency()
|
||||
response = self._query(cmd)
|
||||
if response:
|
||||
try:
|
||||
period_ns = int(response)
|
||||
freq_hz = self.protocol.period_ns_to_frequency(period_ns)
|
||||
self._frequency_hz = freq_hz
|
||||
return freq_hz
|
||||
except (ValueError, ZeroDivisionError) as e:
|
||||
print(f"ERROR: Failed to parse frequency: {e}")
|
||||
return None
|
||||
|
||||
# ==================== Current Control ====================
|
||||
|
||||
def set_current_ma(self, current_ma: float) -> bool:
|
||||
"""
|
||||
Set laser diode current in mA
|
||||
|
||||
Args:
|
||||
current_ma: Current in mA (0-7000)
|
||||
|
||||
Returns:
|
||||
bool: True if command successful
|
||||
"""
|
||||
print(f"DEBUG: Setting laser current to {current_ma} mA")
|
||||
|
||||
try:
|
||||
cmd = self.protocol.cmd_set_current_ma(current_ma)
|
||||
query_cmd = self.protocol.cmd_query_current()
|
||||
expected = str(int(current_ma))
|
||||
|
||||
if self._set_and_verify(cmd, query_cmd, expected):
|
||||
self._current_ma = current_ma
|
||||
return True
|
||||
except ValueError as e:
|
||||
print(f"ERROR: {e}")
|
||||
return False
|
||||
|
||||
def get_current_ma(self) -> float:
|
||||
"""Get current setting in mA"""
|
||||
return self._current_ma
|
||||
|
||||
# ==================== Temperature Monitoring ====================
|
||||
|
||||
def query_pump_temperature_c(self) -> Optional[float]:
|
||||
"""Query pump diode temperature in °C"""
|
||||
cmd = self.protocol.cmd_query_pump_temp()
|
||||
response = self._query(cmd)
|
||||
if response:
|
||||
try:
|
||||
temp_mc = int(response)
|
||||
temp_c = self.protocol.millicelsius_to_celsius(temp_mc)
|
||||
self._pump_temp_c = temp_c
|
||||
return temp_c
|
||||
except ValueError as e:
|
||||
print(f"ERROR: Failed to parse temperature: {e}")
|
||||
return None
|
||||
|
||||
def query_resonator_temperature_c(self) -> Optional[float]:
|
||||
"""Query resonator/SHG temperature in °C"""
|
||||
cmd = self.protocol.cmd_query_resonator_temp()
|
||||
response = self._query(cmd)
|
||||
if response:
|
||||
try:
|
||||
temp_mc = int(response)
|
||||
temp_c = self.protocol.millicelsius_to_celsius(temp_mc)
|
||||
self._resonator_temp_c = temp_c
|
||||
return temp_c
|
||||
except ValueError as e:
|
||||
print(f"ERROR: Failed to parse temperature: {e}")
|
||||
return None
|
||||
|
||||
def query_qswitch_temperature_c(self) -> Optional[float]:
|
||||
"""Query q-switch temperature in °C"""
|
||||
cmd = self.protocol.cmd_query_qswitch_temp()
|
||||
response = self._query(cmd)
|
||||
if response:
|
||||
try:
|
||||
temp_mc = int(response)
|
||||
temp_c = self.protocol.millicelsius_to_celsius(temp_mc)
|
||||
self._qswitch_temp_c = temp_c
|
||||
return temp_c
|
||||
except ValueError as e:
|
||||
print(f"ERROR: Failed to parse temperature: {e}")
|
||||
return None
|
||||
|
||||
def query_power_stage_temperature_c(self) -> Optional[float]:
|
||||
"""Query controller power stage temperature in °C"""
|
||||
cmd = self.protocol.cmd_query_power_stage_temp()
|
||||
response = self._query(cmd)
|
||||
if response:
|
||||
try:
|
||||
temp_mc = int(response)
|
||||
temp_c = self.protocol.millicelsius_to_celsius(temp_mc)
|
||||
self._power_stage_temp_c = temp_c
|
||||
return temp_c
|
||||
except ValueError as e:
|
||||
print(f"ERROR: Failed to parse temperature: {e}")
|
||||
return None
|
||||
|
||||
def query_all_temperatures(self) -> Dict[str, float]:
|
||||
"""
|
||||
Query all temperatures
|
||||
|
||||
Returns:
|
||||
dict: Temperature readings in °C
|
||||
"""
|
||||
temps = {}
|
||||
temps['pump'] = self.query_pump_temperature_c()
|
||||
time.sleep(0.5)
|
||||
temps['resonator'] = self.query_resonator_temperature_c()
|
||||
time.sleep(0.5)
|
||||
temps['qswitch'] = self.query_qswitch_temperature_c()
|
||||
time.sleep(0.5)
|
||||
temps['power_stage'] = self.query_power_stage_temperature_c()
|
||||
return temps
|
||||
|
||||
# ==================== Status and Monitoring ====================
|
||||
|
||||
def query_status(self) -> HeliosStatus:
|
||||
"""Query status register"""
|
||||
cmd = self.protocol.cmd_query_status()
|
||||
response = self._query(cmd)
|
||||
if response:
|
||||
try:
|
||||
status_value = int(response)
|
||||
self._status = HeliosStatus(status_value)
|
||||
return self._status
|
||||
except ValueError as e:
|
||||
print(f"ERROR: Failed to parse status: {e}")
|
||||
return self._status
|
||||
|
||||
def clear_status(self) -> bool:
|
||||
"""Clear status register"""
|
||||
cmd = self.protocol.cmd_clear_status()
|
||||
return self._send_command(cmd)
|
||||
|
||||
def clear_errors(self) -> bool:
|
||||
"""Clear controller errors"""
|
||||
cmd = self.protocol.cmd_clear_errors()
|
||||
return self._send_command(cmd)
|
||||
|
||||
def query_power_monitor_mw(self) -> Optional[float]:
|
||||
"""Query laser power monitor in mW"""
|
||||
cmd = self.protocol.cmd_query_power_monitor()
|
||||
response = self._query(cmd)
|
||||
if response:
|
||||
try:
|
||||
power_mw = float(response)
|
||||
self._power_mw = power_mw
|
||||
return power_mw
|
||||
except ValueError as e:
|
||||
print(f"ERROR: Failed to parse power: {e}")
|
||||
return None
|
||||
|
||||
def query_operation_hours(self) -> Optional[float]:
|
||||
"""Query laser diode operation time in hours"""
|
||||
cmd = self.protocol.cmd_query_operation_time()
|
||||
response = self._query(cmd)
|
||||
if response:
|
||||
try:
|
||||
hours = float(response)
|
||||
self._operation_hours = hours
|
||||
return hours
|
||||
except ValueError as e:
|
||||
print(f"ERROR: Failed to parse operation time: {e}")
|
||||
return None
|
||||
|
||||
# ==================== Serial Numbers ====================
|
||||
|
||||
def query_controller_serial(self) -> str:
|
||||
"""Query controller serial number"""
|
||||
cmd = self.protocol.cmd_query_controller_serial()
|
||||
response = self._query(cmd)
|
||||
if response:
|
||||
self._controller_serial = response
|
||||
return response
|
||||
return ""
|
||||
|
||||
def query_head_serial(self) -> str:
|
||||
"""Query laser head serial number"""
|
||||
cmd = self.protocol.cmd_query_head_serial()
|
||||
response = self._query(cmd)
|
||||
if response:
|
||||
self._head_serial = response
|
||||
return response
|
||||
return ""
|
||||
|
||||
def get_controller_serial(self) -> str:
|
||||
"""Get cached controller serial number"""
|
||||
return self._controller_serial
|
||||
|
||||
def get_head_serial(self) -> str:
|
||||
"""Get cached head serial number"""
|
||||
return self._head_serial
|
||||
|
||||
# ==================== Factory Reset ====================
|
||||
|
||||
def restore_factory_settings(self) -> bool:
|
||||
"""
|
||||
Restore factory settings
|
||||
|
||||
WARNING: This will reset all parameters to factory defaults.
|
||||
Laser must be disabled before calling this method.
|
||||
After calling, wait 2 seconds before power cycling.
|
||||
|
||||
Returns:
|
||||
bool: True if command sent successfully
|
||||
"""
|
||||
if self._laser_enabled:
|
||||
print("ERROR: Laser must be disabled before factory reset")
|
||||
return False
|
||||
|
||||
print("WARNING: Restoring factory settings")
|
||||
cmd = self.protocol.cmd_restore_factory()
|
||||
if self._send_command(cmd):
|
||||
print("INFO: Factory settings restored. Wait 2s before power cycle.")
|
||||
time.sleep(2)
|
||||
return True
|
||||
return False
|
||||
|
||||
# ==================== State Management ====================
|
||||
|
||||
def _update_cached_state(self):
|
||||
"""Update all cached state from hardware"""
|
||||
self.query_laser_enable()
|
||||
time.sleep(0.5)
|
||||
self.query_frequency_hz()
|
||||
time.sleep(0.5)
|
||||
# Current is write-only in some modes, skip query
|
||||
self.query_status()
|
||||
|
||||
def get_status(self) -> Dict:
|
||||
"""
|
||||
Get complete laser status
|
||||
|
||||
Returns:
|
||||
dict: Comprehensive status dictionary
|
||||
"""
|
||||
return {
|
||||
'connected': self.is_connected(),
|
||||
'laser_enabled': self._laser_enabled,
|
||||
'pulse_mode': self._pulse_mode.name,
|
||||
'frequency_hz': self._frequency_hz,
|
||||
'current_ma': self._current_ma,
|
||||
'power_mw': self._power_mw,
|
||||
'operation_hours': self._operation_hours,
|
||||
'temperatures': {
|
||||
'pump_c': self._pump_temp_c,
|
||||
'resonator_c': self._resonator_temp_c,
|
||||
'qswitch_c': self._qswitch_temp_c,
|
||||
'power_stage_c': self._power_stage_temp_c
|
||||
},
|
||||
'status_value': self._status.value,
|
||||
'has_errors': self._status.has_errors(),
|
||||
'controller_serial': self._controller_serial,
|
||||
'head_serial': self._head_serial
|
||||
}
|
||||
|
||||
def update_status(self):
|
||||
"""Update status information from hardware"""
|
||||
if not self.is_connected():
|
||||
return
|
||||
|
||||
self.query_status()
|
||||
time.sleep(0.5)
|
||||
self.query_power_monitor_mw()
|
||||
time.sleep(0.5)
|
||||
self.query_all_temperatures()
|
||||
@@ -0,0 +1,302 @@
|
||||
"""
|
||||
Helios Laser Protocol Handler
|
||||
ASCII-based RS-232 communication protocol for Helios laser systems
|
||||
|
||||
Copyright (C) 2025 Thomas Ales
|
||||
Licensed under GNU General Public License v2.0
|
||||
"""
|
||||
|
||||
from enum import IntEnum
|
||||
from typing import Union, Optional
|
||||
|
||||
|
||||
class PulseMode(IntEnum):
|
||||
"""Helios pulse mode settings"""
|
||||
SINGLE_PULSE = 1
|
||||
GATING = 4
|
||||
CONTINUOUS_PULSING = 14
|
||||
|
||||
|
||||
class HeliosCommand:
|
||||
"""
|
||||
Helios laser command constants and builders
|
||||
|
||||
All commands are ASCII strings terminated with carriage return <CR>
|
||||
Format: COMMAND value<CR> for setting
|
||||
COMMAND<CR> for querying
|
||||
"""
|
||||
|
||||
# Command constants
|
||||
LDO = "LDO" # Laser enabled (0/1)
|
||||
LDG = "LDG" # Pulse mode (1/4/14)
|
||||
LDF = "LDF" # Period between pulses (ns)
|
||||
LRE = "LRE" # Laser remote enable (0/1) - Single electronic only
|
||||
LDS = "LDS" # Laser diode pulse current (mA)
|
||||
LTA = "LTA" # Actual pump diode temperature (m°C)
|
||||
LMA = "LMA" # Actual resonator/SHG temperature (mA)
|
||||
EOA = "EOA" # Actual q-switch temperature (m°C)
|
||||
ELT = "ELT" # Pump diode temp control deviation (m°C)
|
||||
ELM = "ELM" # Resonator/SHG temp control deviation (m°C)
|
||||
EEO = "EEO" # Q-switch temp control deviation (m°C)
|
||||
LTT = "LTT" # Controller power stage temperature (m°C)
|
||||
LER = "LER" # Status register (read)
|
||||
LCE = "LCE" # Clear status register
|
||||
CCE = "CCE" # Clear controller errors
|
||||
CSR = "CSR" # Controller serial number
|
||||
HSR = "HSR" # Laser head serial number
|
||||
HTR = "HTR" # Laser diode operation time (hours)
|
||||
HPR = "HPR" # Restore factory settings
|
||||
HMP = "HMP" # Laser power monitor (mW)
|
||||
|
||||
@staticmethod
|
||||
def build_command(command: str, value: Optional[Union[int, float]] = None) -> bytes:
|
||||
"""
|
||||
Build a Helios command string
|
||||
|
||||
Args:
|
||||
command: Command string (e.g., "LDO", "LDF")
|
||||
value: Optional value to set (None for query)
|
||||
|
||||
Returns:
|
||||
bytes: Command ready to send over serial
|
||||
|
||||
Example:
|
||||
build_command("LDO", 1) -> b"LDO 1\r"
|
||||
build_command("LDO") -> b"LDO\r"
|
||||
"""
|
||||
if value is not None:
|
||||
cmd_str = f"{command} {value}\r"
|
||||
else:
|
||||
cmd_str = f"{command}\r"
|
||||
return cmd_str.encode('ascii')
|
||||
|
||||
@staticmethod
|
||||
def parse_response(response: bytes) -> str:
|
||||
"""
|
||||
Parse response from Helios laser
|
||||
|
||||
Args:
|
||||
response: Raw bytes from serial port
|
||||
|
||||
Returns:
|
||||
str: Parsed response string (stripped of CR/LF)
|
||||
"""
|
||||
return response.decode('ascii').strip()
|
||||
|
||||
|
||||
class HeliosStatus:
|
||||
"""
|
||||
Helios status register decoder
|
||||
|
||||
Status is sum of multiple bit flags:
|
||||
Example: 1*2^0 + 0*2^1 + 1*2^2 = 5
|
||||
"""
|
||||
|
||||
# Status bit definitions (from LER/LCE/CCE commands)
|
||||
# These are example flags - actual flags depend on controller model
|
||||
# Refer to "Troubleshooting" section in manual for complete list
|
||||
|
||||
def __init__(self, status_value: int):
|
||||
"""
|
||||
Initialize status decoder
|
||||
|
||||
Args:
|
||||
status_value: Numeric status value from controller
|
||||
"""
|
||||
self.value = status_value
|
||||
self.flags = self._decode_flags(status_value)
|
||||
|
||||
def _decode_flags(self, value: int) -> list:
|
||||
"""Decode status value into list of active bit positions"""
|
||||
flags = []
|
||||
bit_pos = 0
|
||||
while value > 0:
|
||||
if value & 1:
|
||||
flags.append(bit_pos)
|
||||
value >>= 1
|
||||
bit_pos += 1
|
||||
return flags
|
||||
|
||||
def has_errors(self) -> bool:
|
||||
"""Check if any error flags are set"""
|
||||
return self.value > 0
|
||||
|
||||
def __str__(self) -> str:
|
||||
return f"Status: {self.value} (flags: {self.flags})"
|
||||
|
||||
|
||||
class HeliosProtocol:
|
||||
"""
|
||||
High-level Helios protocol handler
|
||||
Provides parameter validation and unit conversions
|
||||
"""
|
||||
|
||||
# Parameter ranges (from protocol document)
|
||||
RANGE_LDO = (0, 1)
|
||||
RANGE_LDG = (0, 14)
|
||||
RANGE_LDF = (8000, 60000) # ns
|
||||
RANGE_LRE = (0, 1)
|
||||
RANGE_LDS = (0, 7000) # mA
|
||||
RANGE_LTA = (5000, 50000) # m°C
|
||||
RANGE_LMA = (0, 4000) # mA (seems like error in doc, should be m°C)
|
||||
RANGE_EOA = (5000, 50000) # m°C
|
||||
RANGE_ELT = (-32768, 32767) # m°C
|
||||
RANGE_ELM = (-32768, 32767) # m°C
|
||||
RANGE_EEO = (-32768, 32767) # m°C
|
||||
RANGE_LTT = (5000, 65535) # m°C
|
||||
RANGE_HTR = (0, 65535) # hours
|
||||
RANGE_HMP = (0, 5000) # mW
|
||||
|
||||
def __init__(self):
|
||||
"""Initialize protocol handler"""
|
||||
pass
|
||||
|
||||
# Temperature conversions (m°C <-> °C)
|
||||
|
||||
@staticmethod
|
||||
def celsius_to_millicelsius(temp_c: float) -> int:
|
||||
"""Convert temperature from °C to m°C (milli-Celsius)"""
|
||||
return int(temp_c * 1000)
|
||||
|
||||
@staticmethod
|
||||
def millicelsius_to_celsius(temp_mc: int) -> float:
|
||||
"""Convert temperature from m°C to °C"""
|
||||
return temp_mc / 1000.0
|
||||
|
||||
# Frequency conversions (Hz <-> ns period)
|
||||
|
||||
@staticmethod
|
||||
def frequency_to_period_ns(freq_hz: float) -> int:
|
||||
"""
|
||||
Convert frequency in Hz to period in nanoseconds
|
||||
|
||||
Args:
|
||||
freq_hz: Frequency in Hz
|
||||
|
||||
Returns:
|
||||
int: Period in nanoseconds
|
||||
|
||||
Example:
|
||||
50000 Hz -> 20000 ns (50 kHz)
|
||||
"""
|
||||
if freq_hz <= 0:
|
||||
raise ValueError("Frequency must be positive")
|
||||
period_ns = int(1e9 / freq_hz)
|
||||
return period_ns
|
||||
|
||||
@staticmethod
|
||||
def period_ns_to_frequency(period_ns: int) -> float:
|
||||
"""
|
||||
Convert period in nanoseconds to frequency in Hz
|
||||
|
||||
Args:
|
||||
period_ns: Period in nanoseconds
|
||||
|
||||
Returns:
|
||||
float: Frequency in Hz
|
||||
"""
|
||||
if period_ns <= 0:
|
||||
raise ValueError("Period must be positive")
|
||||
freq_hz = 1e9 / period_ns
|
||||
return freq_hz
|
||||
|
||||
# Command builders with validation
|
||||
|
||||
def cmd_set_laser_enable(self, enabled: bool) -> bytes:
|
||||
"""Build command to enable/disable laser"""
|
||||
value = 1 if enabled else 0
|
||||
return HeliosCommand.build_command(HeliosCommand.LDO, value)
|
||||
|
||||
def cmd_query_laser_enable(self) -> bytes:
|
||||
"""Build query for laser enable state"""
|
||||
return HeliosCommand.build_command(HeliosCommand.LDO)
|
||||
|
||||
def cmd_set_pulse_mode(self, mode: PulseMode) -> bytes:
|
||||
"""Build command to set pulse mode"""
|
||||
if mode not in [PulseMode.SINGLE_PULSE, PulseMode.GATING,
|
||||
PulseMode.CONTINUOUS_PULSING]:
|
||||
raise ValueError(f"Invalid pulse mode: {mode}")
|
||||
return HeliosCommand.build_command(HeliosCommand.LDG, mode)
|
||||
|
||||
def cmd_query_pulse_mode(self) -> bytes:
|
||||
"""Build query for pulse mode"""
|
||||
return HeliosCommand.build_command(HeliosCommand.LDG)
|
||||
|
||||
def cmd_set_frequency_hz(self, freq_hz: float) -> bytes:
|
||||
"""
|
||||
Build command to set laser frequency (Hz)
|
||||
Converts to period in ns internally
|
||||
"""
|
||||
period_ns = self.frequency_to_period_ns(freq_hz)
|
||||
if not (self.RANGE_LDF[0] <= period_ns <= self.RANGE_LDF[1]):
|
||||
raise ValueError(f"Frequency results in period {period_ns}ns, "
|
||||
f"valid range: {self.RANGE_LDF[0]}-{self.RANGE_LDF[1]}ns")
|
||||
return HeliosCommand.build_command(HeliosCommand.LDF, period_ns)
|
||||
|
||||
def cmd_query_frequency(self) -> bytes:
|
||||
"""Build query for laser frequency (returns period in ns)"""
|
||||
return HeliosCommand.build_command(HeliosCommand.LDF)
|
||||
|
||||
def cmd_set_current_ma(self, current_ma: float) -> bytes:
|
||||
"""Build command to set laser diode current (mA)"""
|
||||
if not (self.RANGE_LDS[0] <= current_ma <= self.RANGE_LDS[1]):
|
||||
raise ValueError(f"Current {current_ma}mA outside valid range: "
|
||||
f"{self.RANGE_LDS[0]}-{self.RANGE_LDS[1]}mA")
|
||||
return HeliosCommand.build_command(HeliosCommand.LDS, int(current_ma))
|
||||
|
||||
def cmd_query_current(self) -> bytes:
|
||||
"""Build query for laser diode current"""
|
||||
return HeliosCommand.build_command(HeliosCommand.LDS)
|
||||
|
||||
def cmd_query_pump_temp(self) -> bytes:
|
||||
"""Build query for actual pump diode temperature"""
|
||||
return HeliosCommand.build_command(HeliosCommand.LTA)
|
||||
|
||||
def cmd_query_resonator_temp(self) -> bytes:
|
||||
"""Build query for actual resonator/SHG temperature"""
|
||||
return HeliosCommand.build_command(HeliosCommand.LMA)
|
||||
|
||||
def cmd_query_qswitch_temp(self) -> bytes:
|
||||
"""Build query for actual q-switch temperature"""
|
||||
return HeliosCommand.build_command(HeliosCommand.EOA)
|
||||
|
||||
def cmd_query_power_stage_temp(self) -> bytes:
|
||||
"""Build query for controller power stage temperature"""
|
||||
return HeliosCommand.build_command(HeliosCommand.LTT)
|
||||
|
||||
def cmd_query_status(self) -> bytes:
|
||||
"""Build query for status register"""
|
||||
return HeliosCommand.build_command(HeliosCommand.LER)
|
||||
|
||||
def cmd_clear_status(self) -> bytes:
|
||||
"""Build command to clear status register"""
|
||||
return HeliosCommand.build_command(HeliosCommand.LCE, 0)
|
||||
|
||||
def cmd_clear_errors(self) -> bytes:
|
||||
"""Build command to clear controller errors"""
|
||||
return HeliosCommand.build_command(HeliosCommand.CCE, 0)
|
||||
|
||||
def cmd_query_controller_serial(self) -> bytes:
|
||||
"""Build query for controller serial number"""
|
||||
return HeliosCommand.build_command(HeliosCommand.CSR)
|
||||
|
||||
def cmd_query_head_serial(self) -> bytes:
|
||||
"""Build query for laser head serial number"""
|
||||
return HeliosCommand.build_command(HeliosCommand.HSR)
|
||||
|
||||
def cmd_query_operation_time(self) -> bytes:
|
||||
"""Build query for laser diode operation time (hours)"""
|
||||
return HeliosCommand.build_command(HeliosCommand.HTR)
|
||||
|
||||
def cmd_query_power_monitor(self) -> bytes:
|
||||
"""Build query for laser power monitor (mW)"""
|
||||
return HeliosCommand.build_command(HeliosCommand.HMP)
|
||||
|
||||
def cmd_restore_factory(self) -> bytes:
|
||||
"""
|
||||
Build command to restore factory settings
|
||||
|
||||
WARNING: Laser must be disabled (LDO 0) before this command
|
||||
After sending, wait 2 seconds before rebooting/power cycling
|
||||
"""
|
||||
return HeliosCommand.build_command(HeliosCommand.HPR)
|
||||
@@ -0,0 +1,387 @@
|
||||
"""
|
||||
Microscope Controller
|
||||
Handles communication with Genesis and Helios microscope/laser systems
|
||||
|
||||
Genesis: Laser scanning microscope system (stub)
|
||||
Helios: Laser control system with frequency and current control (full implementation)
|
||||
Both systems communicate via USB/Serial interfaces
|
||||
|
||||
Copyright (C) 2025 Thomas Ales
|
||||
Licensed under GNU General Public License v2.0
|
||||
"""
|
||||
|
||||
import time
|
||||
from typing import Dict, Optional
|
||||
from hardware.helios_driver import HeliosDriver, PulseMode
|
||||
|
||||
|
||||
class MicroscopeController:
|
||||
"""
|
||||
Controller for Genesis and Helios microscope systems
|
||||
|
||||
Provides methods for:
|
||||
- System connection and initialization
|
||||
- Interlock status monitoring
|
||||
- Parameter configuration
|
||||
- Safety checks
|
||||
"""
|
||||
|
||||
def __init__(self):
|
||||
"""Initialize microscope controller"""
|
||||
# Genesis state (stub)
|
||||
self._genesis_connected = False
|
||||
self._genesis_ready = False
|
||||
self._genesis_interlocked = False
|
||||
self._genesis_power_mw = 0.0
|
||||
|
||||
# Helios driver (full implementation)
|
||||
self._helios_driver = HeliosDriver()
|
||||
self._helios_port = None
|
||||
|
||||
# System interlock (master safety)
|
||||
self._system_interlocked = False
|
||||
|
||||
print("INFO: Microscope controller initialized")
|
||||
print(" - Genesis: Stub implementation")
|
||||
print(" - Helios: Full RS-232 driver")
|
||||
|
||||
# ==================== Genesis Methods ====================
|
||||
|
||||
def connect_genesis(self) -> bool:
|
||||
"""
|
||||
Connect to Genesis laser scanning microscope
|
||||
|
||||
Returns:
|
||||
bool: True if connection successful
|
||||
"""
|
||||
print("DEBUG: Connecting to Genesis microscope")
|
||||
|
||||
# Stub implementation
|
||||
time.sleep(0.1)
|
||||
|
||||
self._genesis_connected = True
|
||||
self._genesis_ready = True
|
||||
self._genesis_interlocked = True # Assume interlocks OK
|
||||
|
||||
print("INFO: Genesis microscope connected")
|
||||
return True
|
||||
|
||||
def disconnect_genesis(self) -> bool:
|
||||
"""
|
||||
Disconnect from Genesis microscope
|
||||
|
||||
Returns:
|
||||
bool: True if disconnection successful
|
||||
"""
|
||||
print("DEBUG: Disconnecting from Genesis microscope")
|
||||
|
||||
self._genesis_connected = False
|
||||
self._genesis_ready = False
|
||||
|
||||
print("INFO: Genesis microscope disconnected")
|
||||
return True
|
||||
|
||||
def apply_genesis_settings(self, settings: Dict) -> bool:
|
||||
"""
|
||||
Apply Genesis configuration settings
|
||||
|
||||
Args:
|
||||
settings: Dictionary containing Genesis parameters
|
||||
|
||||
Returns:
|
||||
bool: True if settings applied successfully
|
||||
"""
|
||||
print("DEBUG: Applying Genesis settings")
|
||||
print(f" Power: {settings.get('power_mw', 0)} mW")
|
||||
|
||||
if not self._genesis_connected:
|
||||
print("ERROR: Genesis not connected")
|
||||
return False
|
||||
|
||||
self._genesis_power_mw = settings.get('power_mw', 0.0)
|
||||
|
||||
# Stub - would send commands to actual hardware
|
||||
time.sleep(0.05)
|
||||
|
||||
print("INFO: Genesis settings applied")
|
||||
return True
|
||||
|
||||
def get_genesis_status(self) -> Dict[str, any]:
|
||||
"""
|
||||
Get Genesis microscope status
|
||||
|
||||
Returns:
|
||||
dict: Genesis status information
|
||||
"""
|
||||
return {
|
||||
'connected': self._genesis_connected,
|
||||
'ready': self._genesis_ready,
|
||||
'interlocked': self._genesis_interlocked,
|
||||
'power_mw': self._genesis_power_mw
|
||||
}
|
||||
|
||||
# ==================== Helios Methods ====================
|
||||
|
||||
def connect_helios(self, port: str) -> bool:
|
||||
"""
|
||||
Connect to Helios laser system
|
||||
|
||||
Args:
|
||||
port: COM port for Helios device
|
||||
|
||||
Returns:
|
||||
bool: True if connection successful
|
||||
"""
|
||||
print(f"DEBUG: Connecting to Helios on {port}")
|
||||
|
||||
if not self._helios_driver.connect(port):
|
||||
return False
|
||||
|
||||
self._helios_port = port
|
||||
print(f"INFO: Helios connected on {port}")
|
||||
print(f" Controller S/N: {self._helios_driver.get_controller_serial()}")
|
||||
print(f" Head S/N: {self._helios_driver.get_head_serial()}")
|
||||
return True
|
||||
|
||||
def disconnect_helios(self) -> bool:
|
||||
"""
|
||||
Disconnect from Helios laser system
|
||||
|
||||
Returns:
|
||||
bool: True if disconnection successful
|
||||
"""
|
||||
print("DEBUG: Disconnecting from Helios")
|
||||
return self._helios_driver.disconnect()
|
||||
|
||||
def is_helios_connected(self) -> bool:
|
||||
"""Check if Helios is connected"""
|
||||
return self._helios_driver.is_connected()
|
||||
|
||||
def apply_helios_settings(self, settings: Dict) -> bool:
|
||||
"""
|
||||
Apply Helios configuration settings
|
||||
|
||||
This method connects and configures the Helios laser with the
|
||||
specified parameters from the settings dialog.
|
||||
|
||||
Args:
|
||||
settings: Dictionary containing:
|
||||
- com_port: COM port name
|
||||
- frequency_hz: Laser frequency in Hz
|
||||
- current_ma: Laser diode current in mA
|
||||
|
||||
Returns:
|
||||
bool: True if settings applied successfully
|
||||
"""
|
||||
if settings is None:
|
||||
print("ERROR: No settings provided")
|
||||
return False
|
||||
|
||||
print("DEBUG: Applying Helios settings")
|
||||
print(f" Port: {settings.get('com_port', 'N/A')}")
|
||||
print(f" Frequency: {settings.get('frequency_hz', 0)} Hz")
|
||||
print(f" Current: {settings.get('current_ma', 0)} mA")
|
||||
|
||||
# Connect if not already connected or port changed
|
||||
port = settings.get('com_port')
|
||||
if not port:
|
||||
print("ERROR: No COM port specified")
|
||||
return False
|
||||
|
||||
if not self.is_helios_connected() or port != self._helios_port:
|
||||
if self.is_helios_connected():
|
||||
self.disconnect_helios()
|
||||
if not self.connect_helios(port):
|
||||
return False
|
||||
|
||||
# Set frequency
|
||||
freq_hz = settings.get('frequency_hz', 0.0)
|
||||
if freq_hz > 0:
|
||||
if not self._helios_driver.set_frequency_hz(freq_hz):
|
||||
print("ERROR: Failed to set frequency")
|
||||
return False
|
||||
time.sleep(0.05)
|
||||
|
||||
# Set current
|
||||
current_ma = settings.get('current_ma', 0.0)
|
||||
if current_ma > 0:
|
||||
if not self._helios_driver.set_current_ma(current_ma):
|
||||
print("ERROR: Failed to set current")
|
||||
return False
|
||||
time.sleep(0.05)
|
||||
|
||||
# Set to continuous pulsing mode by default
|
||||
if not self._helios_driver.set_pulse_mode(PulseMode.CONTINUOUS_PULSING):
|
||||
print("WARNING: Failed to set pulse mode")
|
||||
|
||||
print("INFO: Helios settings applied successfully")
|
||||
return True
|
||||
|
||||
def helios_enable_laser(self, enabled: bool) -> bool:
|
||||
"""
|
||||
Enable or disable Helios laser
|
||||
|
||||
Args:
|
||||
enabled: True to enable, False to disable
|
||||
|
||||
Returns:
|
||||
bool: True if command successful
|
||||
"""
|
||||
if not self.is_helios_connected():
|
||||
print("ERROR: Helios not connected")
|
||||
return False
|
||||
|
||||
return self._helios_driver.set_laser_enable(enabled)
|
||||
|
||||
def is_helios_laser_enabled(self) -> bool:
|
||||
"""Check if Helios laser is currently enabled"""
|
||||
if not self.is_helios_connected():
|
||||
return False
|
||||
return self._helios_driver.is_laser_enabled()
|
||||
|
||||
def get_helios_status(self) -> Dict[str, any]:
|
||||
"""
|
||||
Get Helios laser system status
|
||||
|
||||
Returns:
|
||||
dict: Helios status information
|
||||
"""
|
||||
if not self.is_helios_connected():
|
||||
return {
|
||||
'connected': False,
|
||||
'ready': False,
|
||||
'interlocked': False,
|
||||
'port': None,
|
||||
'frequency_hz': 0.0,
|
||||
'current_ma': 0.0,
|
||||
'laser_enabled': False,
|
||||
'power_mw': 0.0
|
||||
}
|
||||
|
||||
# Get comprehensive status from driver
|
||||
status = self._helios_driver.get_status()
|
||||
|
||||
return {
|
||||
'connected': status['connected'],
|
||||
'ready': not status['has_errors'],
|
||||
'interlocked': not status['has_errors'], # Use error state as interlock
|
||||
'port': self._helios_port,
|
||||
'frequency_hz': status['frequency_hz'],
|
||||
'current_ma': status['current_ma'],
|
||||
'laser_enabled': status['laser_enabled'],
|
||||
'power_mw': status['power_mw'],
|
||||
'operation_hours': status['operation_hours'],
|
||||
'controller_serial': status['controller_serial'],
|
||||
'head_serial': status['head_serial']
|
||||
}
|
||||
|
||||
def helios_update_status(self):
|
||||
"""Update Helios status from hardware"""
|
||||
if self.is_helios_connected():
|
||||
self._helios_driver.update_status()
|
||||
|
||||
# ==================== Combined Status Methods ====================
|
||||
|
||||
def get_status(self) -> Dict[str, any]:
|
||||
"""
|
||||
Get complete microscope system status
|
||||
|
||||
Returns:
|
||||
dict: Combined status for Genesis, Helios, and interlocks
|
||||
"""
|
||||
# Get Helios status from driver
|
||||
helios_status = self.get_helios_status()
|
||||
helios_ready = helios_status.get('ready', False)
|
||||
helios_interlocked = helios_status.get('interlocked', False)
|
||||
|
||||
return {
|
||||
# System-wide
|
||||
'interlocked': self._system_interlocked or (
|
||||
self._genesis_interlocked and helios_interlocked
|
||||
),
|
||||
|
||||
# Genesis
|
||||
'genesis_ready': self._genesis_ready,
|
||||
'genesis_interlocked': self._genesis_interlocked,
|
||||
|
||||
# Helios
|
||||
'helios_ready': helios_ready,
|
||||
'helios_interlocked': helios_interlocked
|
||||
}
|
||||
|
||||
def check_interlocks(self) -> bool:
|
||||
"""
|
||||
Check all safety interlocks
|
||||
|
||||
Returns:
|
||||
bool: True if all interlocks are satisfied
|
||||
"""
|
||||
# Check Genesis interlocks
|
||||
if self._genesis_connected and not self._genesis_interlocked:
|
||||
print("WARNING: Genesis interlock not satisfied")
|
||||
return False
|
||||
|
||||
# Check Helios interlocks
|
||||
if self.is_helios_connected():
|
||||
helios_status = self.get_helios_status()
|
||||
if not helios_status.get('interlocked', False):
|
||||
print("WARNING: Helios interlock not satisfied")
|
||||
return False
|
||||
|
||||
return True
|
||||
|
||||
# ==================== Scan Preparation ====================
|
||||
|
||||
def prepare_for_scan(self, params: Dict) -> bool:
|
||||
"""
|
||||
Prepare microscope systems for scanning
|
||||
|
||||
Args:
|
||||
params: Scan parameters dictionary
|
||||
|
||||
Returns:
|
||||
bool: True if preparation successful
|
||||
"""
|
||||
print("DEBUG: Preparing microscope systems for scan")
|
||||
|
||||
# Check interlocks
|
||||
if not self.check_interlocks():
|
||||
print("ERROR: Interlock check failed")
|
||||
return False
|
||||
|
||||
# Verify systems are ready
|
||||
if self._genesis_connected and not self._genesis_ready:
|
||||
print("ERROR: Genesis not ready")
|
||||
return False
|
||||
|
||||
if self.is_helios_connected():
|
||||
helios_status = self.get_helios_status()
|
||||
if not helios_status.get('ready', False):
|
||||
print("ERROR: Helios not ready")
|
||||
return False
|
||||
|
||||
# Configure for scan
|
||||
# Stub implementation
|
||||
time.sleep(0.1)
|
||||
|
||||
print("INFO: Microscope systems ready for scan")
|
||||
return True
|
||||
|
||||
def emergency_stop(self) -> bool:
|
||||
"""
|
||||
Emergency stop all microscope operations
|
||||
|
||||
Returns:
|
||||
bool: True if stop successful
|
||||
"""
|
||||
print("WARNING: Emergency stop triggered")
|
||||
|
||||
# Disable all systems
|
||||
if self._genesis_connected:
|
||||
self._genesis_ready = False
|
||||
|
||||
if self.is_helios_connected():
|
||||
# Disable Helios laser immediately
|
||||
self._helios_driver.set_laser_enable(False)
|
||||
|
||||
return True
|
||||
@@ -0,0 +1,244 @@
|
||||
"""
|
||||
ThorLabs Stage Settings Management
|
||||
Handles saving and loading of stage configuration parameters
|
||||
|
||||
Copyright (C) 2025 Thomas Ales
|
||||
Licensed under GNU General Public License v2.0
|
||||
"""
|
||||
|
||||
import json
|
||||
import os
|
||||
from typing import Dict, Optional
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
class StageSettings:
|
||||
"""
|
||||
Manages stage configuration settings including velocity, acceleration,
|
||||
and trigger I/O configuration
|
||||
"""
|
||||
|
||||
DEFAULT_SETTINGS = {
|
||||
'velocity': {
|
||||
'x_axis': 1.0, # mm/s
|
||||
'y_axis': 1.0, # mm/s
|
||||
'z_axis': 1.0, # mm/s
|
||||
},
|
||||
'acceleration': {
|
||||
'x_axis': 5.0, # mm/s²
|
||||
'y_axis': 5.0, # mm/s²
|
||||
'z_axis': 5.0, # mm/s²
|
||||
},
|
||||
'trigger': {
|
||||
'x_axis': {
|
||||
'mode': 0x00, # Disabled
|
||||
'polarity': 0x01, # Active high
|
||||
'start_pos_fwd': 0.0,
|
||||
'start_pos_rev': 0.0,
|
||||
'interval_fwd': 0.0,
|
||||
'interval_rev': 0.0
|
||||
},
|
||||
'y_axis': {
|
||||
'mode': 0x00, # Disabled
|
||||
'polarity': 0x01, # Active high
|
||||
'start_pos_fwd': 0.0,
|
||||
'start_pos_rev': 0.0,
|
||||
'interval_fwd': 0.0,
|
||||
'interval_rev': 0.0
|
||||
},
|
||||
'z_axis': {
|
||||
'mode': 0x00, # Disabled
|
||||
'polarity': 0x01, # Active high
|
||||
'start_pos_fwd': 0.0,
|
||||
'start_pos_rev': 0.0,
|
||||
'interval_fwd': 0.0,
|
||||
'interval_rev': 0.0
|
||||
}
|
||||
},
|
||||
'digital_io': {
|
||||
'output_bits': 0x00
|
||||
}
|
||||
}
|
||||
|
||||
def __init__(self, settings_file: Optional[str] = None):
|
||||
"""
|
||||
Initialize stage settings manager
|
||||
|
||||
Args:
|
||||
settings_file: Path to settings file (default: ~/.nuescan/stage_settings.json)
|
||||
"""
|
||||
if settings_file is None:
|
||||
# Default to user home directory
|
||||
home = Path.home()
|
||||
settings_dir = home / '.nuescan'
|
||||
settings_dir.mkdir(exist_ok=True)
|
||||
settings_file = str(settings_dir / 'stage_settings.json')
|
||||
|
||||
self.settings_file = settings_file
|
||||
self.settings = self.DEFAULT_SETTINGS.copy()
|
||||
|
||||
# Load existing settings if available
|
||||
self.load()
|
||||
|
||||
def load(self) -> bool:
|
||||
"""
|
||||
Load settings from file
|
||||
|
||||
Returns:
|
||||
bool: True if settings loaded successfully, False otherwise
|
||||
"""
|
||||
if not os.path.exists(self.settings_file):
|
||||
print(f"INFO: Settings file not found, using defaults: {self.settings_file}")
|
||||
return False
|
||||
|
||||
try:
|
||||
with open(self.settings_file, 'r') as f:
|
||||
loaded_settings = json.load(f)
|
||||
|
||||
# Merge with defaults to ensure all keys exist
|
||||
self._merge_settings(loaded_settings)
|
||||
|
||||
print(f"INFO: Loaded stage settings from {self.settings_file}")
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
print(f"ERROR: Failed to load settings from {self.settings_file}: {e}")
|
||||
return False
|
||||
|
||||
def save(self) -> bool:
|
||||
"""
|
||||
Save settings to file
|
||||
|
||||
Returns:
|
||||
bool: True if settings saved successfully, False otherwise
|
||||
"""
|
||||
try:
|
||||
# Ensure directory exists
|
||||
os.makedirs(os.path.dirname(self.settings_file), exist_ok=True)
|
||||
|
||||
with open(self.settings_file, 'w') as f:
|
||||
json.dump(self.settings, f, indent=4)
|
||||
|
||||
print(f"INFO: Saved stage settings to {self.settings_file}")
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
print(f"ERROR: Failed to save settings to {self.settings_file}: {e}")
|
||||
return False
|
||||
|
||||
def _merge_settings(self, loaded_settings: Dict):
|
||||
"""Merge loaded settings with defaults"""
|
||||
# Velocity
|
||||
if 'velocity' in loaded_settings:
|
||||
self.settings['velocity'].update(loaded_settings['velocity'])
|
||||
|
||||
# Acceleration
|
||||
if 'acceleration' in loaded_settings:
|
||||
self.settings['acceleration'].update(loaded_settings['acceleration'])
|
||||
|
||||
# Trigger
|
||||
if 'trigger' in loaded_settings:
|
||||
for axis in ['x_axis', 'y_axis', 'z_axis']:
|
||||
if axis in loaded_settings['trigger']:
|
||||
self.settings['trigger'][axis].update(loaded_settings['trigger'][axis])
|
||||
|
||||
# Digital I/O
|
||||
if 'digital_io' in loaded_settings:
|
||||
self.settings['digital_io'].update(loaded_settings['digital_io'])
|
||||
|
||||
# ==================== Velocity Settings ====================
|
||||
|
||||
def get_velocity(self, axis: str) -> float:
|
||||
"""Get velocity for specific axis (x_axis, y_axis, z_axis)"""
|
||||
return self.settings['velocity'].get(axis, 1.0)
|
||||
|
||||
def set_velocity(self, axis: str, velocity: float):
|
||||
"""Set velocity for specific axis"""
|
||||
if axis in ['x_axis', 'y_axis', 'z_axis']:
|
||||
self.settings['velocity'][axis] = velocity
|
||||
|
||||
def get_all_velocities(self) -> Dict[str, float]:
|
||||
"""Get all axis velocities"""
|
||||
return self.settings['velocity'].copy()
|
||||
|
||||
def set_all_velocities(self, x: float, y: float, z: float):
|
||||
"""Set all axis velocities"""
|
||||
self.settings['velocity']['x_axis'] = x
|
||||
self.settings['velocity']['y_axis'] = y
|
||||
self.settings['velocity']['z_axis'] = z
|
||||
|
||||
# ==================== Acceleration Settings ====================
|
||||
|
||||
def get_acceleration(self, axis: str) -> float:
|
||||
"""Get acceleration for specific axis"""
|
||||
return self.settings['acceleration'].get(axis, 5.0)
|
||||
|
||||
def set_acceleration(self, axis: str, acceleration: float):
|
||||
"""Set acceleration for specific axis"""
|
||||
if axis in ['x_axis', 'y_axis', 'z_axis']:
|
||||
self.settings['acceleration'][axis] = acceleration
|
||||
|
||||
def get_all_accelerations(self) -> Dict[str, float]:
|
||||
"""Get all axis accelerations"""
|
||||
return self.settings['acceleration'].copy()
|
||||
|
||||
def set_all_accelerations(self, x: float, y: float, z: float):
|
||||
"""Set all axis accelerations"""
|
||||
self.settings['acceleration']['x_axis'] = x
|
||||
self.settings['acceleration']['y_axis'] = y
|
||||
self.settings['acceleration']['z_axis'] = z
|
||||
|
||||
# ==================== Trigger Settings ====================
|
||||
|
||||
def get_trigger_config(self, axis: str) -> Dict:
|
||||
"""Get trigger configuration for specific axis"""
|
||||
return self.settings['trigger'].get(axis, {}).copy()
|
||||
|
||||
def set_trigger_config(self, axis: str, mode: int, polarity: int = 0x01,
|
||||
start_pos_fwd: float = 0.0, start_pos_rev: float = 0.0,
|
||||
interval_fwd: float = 0.0, interval_rev: float = 0.0):
|
||||
"""Set trigger configuration for specific axis"""
|
||||
if axis in ['x_axis', 'y_axis', 'z_axis']:
|
||||
self.settings['trigger'][axis] = {
|
||||
'mode': mode,
|
||||
'polarity': polarity,
|
||||
'start_pos_fwd': start_pos_fwd,
|
||||
'start_pos_rev': start_pos_rev,
|
||||
'interval_fwd': interval_fwd,
|
||||
'interval_rev': interval_rev
|
||||
}
|
||||
|
||||
def get_trigger_mode(self, axis: str) -> int:
|
||||
"""Get trigger mode for specific axis"""
|
||||
return self.settings['trigger'].get(axis, {}).get('mode', 0x00)
|
||||
|
||||
def set_trigger_mode(self, axis: str, mode: int):
|
||||
"""Set trigger mode for specific axis"""
|
||||
if axis in ['x_axis', 'y_axis', 'z_axis']:
|
||||
if axis not in self.settings['trigger']:
|
||||
self.settings['trigger'][axis] = self.DEFAULT_SETTINGS['trigger'][axis].copy()
|
||||
self.settings['trigger'][axis]['mode'] = mode
|
||||
|
||||
# ==================== Digital I/O Settings ====================
|
||||
|
||||
def get_digital_outputs(self) -> int:
|
||||
"""Get digital output bits"""
|
||||
return self.settings['digital_io'].get('output_bits', 0x00)
|
||||
|
||||
def set_digital_outputs(self, output_bits: int):
|
||||
"""Set digital output bits"""
|
||||
self.settings['digital_io']['output_bits'] = output_bits
|
||||
|
||||
# ==================== Utility Methods ====================
|
||||
|
||||
def reset_to_defaults(self):
|
||||
"""Reset all settings to defaults"""
|
||||
self.settings = self.DEFAULT_SETTINGS.copy()
|
||||
|
||||
def get_all_settings(self) -> Dict:
|
||||
"""Get copy of all settings"""
|
||||
return json.loads(json.dumps(self.settings)) # Deep copy via JSON
|
||||
|
||||
def update_from_dict(self, settings_dict: Dict):
|
||||
"""Update settings from dictionary"""
|
||||
self._merge_settings(settings_dict)
|
||||
@@ -0,0 +1,259 @@
|
||||
"""
|
||||
T3R-SL Device Controller
|
||||
Handles communication with T3R-SL device via USB/Serial
|
||||
|
||||
The T3R-SL is a specialized instrument control device that provides
|
||||
timing, triggering, and coordination for the SRAS scanning system.
|
||||
"""
|
||||
|
||||
import time
|
||||
from typing import Dict, List, Optional
|
||||
|
||||
|
||||
class T3RDevice:
|
||||
"""
|
||||
Controller for T3R-SL device
|
||||
|
||||
Provides methods for:
|
||||
- Connecting/disconnecting
|
||||
- Device initialization and homing
|
||||
- Status monitoring
|
||||
- Trigger and timing control
|
||||
"""
|
||||
|
||||
def __init__(self):
|
||||
"""Initialize T3R device controller"""
|
||||
self._connected = False
|
||||
self._port = None
|
||||
self._homed = False
|
||||
self._ready = False
|
||||
|
||||
# Device state
|
||||
self._initialized = False
|
||||
self._error_state = False
|
||||
|
||||
print("INFO: T3R-SL Device controller initialized (stub)")
|
||||
|
||||
def get_available_ports(self) -> List[str]:
|
||||
"""
|
||||
Get list of available COM ports
|
||||
|
||||
Returns:
|
||||
list: Available port names
|
||||
"""
|
||||
# Stub implementation - return dummy ports
|
||||
# In production, would scan for actual serial ports
|
||||
return [
|
||||
"COM1", "COM2", "COM3", "COM4", "COM5",
|
||||
"/dev/ttyUSB0", "/dev/ttyUSB1", "/dev/ttyUSB2"
|
||||
]
|
||||
|
||||
def connect(self, port: str) -> bool:
|
||||
"""
|
||||
Connect to T3R device on specified port
|
||||
|
||||
Args:
|
||||
port: COM port name
|
||||
|
||||
Returns:
|
||||
bool: True if connection successful
|
||||
"""
|
||||
print(f"DEBUG: Connecting to T3R device on {port}")
|
||||
|
||||
# Stub implementation
|
||||
time.sleep(0.1)
|
||||
|
||||
self._port = port
|
||||
self._connected = True
|
||||
self._ready = False # Need to initialize after connect
|
||||
|
||||
print(f"INFO: Connected to T3R device on {port}")
|
||||
|
||||
# Auto-initialize
|
||||
return self._initialize()
|
||||
|
||||
def disconnect(self) -> bool:
|
||||
"""
|
||||
Disconnect from T3R device
|
||||
|
||||
Returns:
|
||||
bool: True if disconnection successful
|
||||
"""
|
||||
print("DEBUG: Disconnecting from T3R device")
|
||||
|
||||
self._connected = False
|
||||
self._homed = False
|
||||
self._ready = False
|
||||
self._initialized = False
|
||||
|
||||
print("INFO: Disconnected from T3R device")
|
||||
return True
|
||||
|
||||
def is_connected(self) -> bool:
|
||||
"""Check if device is connected"""
|
||||
return self._connected
|
||||
|
||||
def _initialize(self) -> bool:
|
||||
"""
|
||||
Initialize T3R device after connection
|
||||
|
||||
Returns:
|
||||
bool: True if initialization successful
|
||||
"""
|
||||
print("DEBUG: Initializing T3R device")
|
||||
|
||||
if not self._connected:
|
||||
print("ERROR: Cannot initialize - device not connected")
|
||||
return False
|
||||
|
||||
# Stub - simulate initialization
|
||||
time.sleep(0.2)
|
||||
|
||||
self._initialized = True
|
||||
self._error_state = False
|
||||
|
||||
# Perform homing
|
||||
return self.home()
|
||||
|
||||
def home(self) -> bool:
|
||||
"""
|
||||
Home/zero T3R device
|
||||
|
||||
Returns:
|
||||
bool: True if homing successful
|
||||
"""
|
||||
print("DEBUG: Homing T3R device")
|
||||
|
||||
if not self._connected or not self._initialized:
|
||||
print("ERROR: Cannot home - device not initialized")
|
||||
return False
|
||||
|
||||
# Stub - simulate homing
|
||||
time.sleep(0.3)
|
||||
|
||||
self._homed = True
|
||||
self._ready = True
|
||||
|
||||
print("INFO: T3R device homed successfully")
|
||||
return True
|
||||
|
||||
def get_status(self) -> Dict[str, bool]:
|
||||
"""
|
||||
Get current device status
|
||||
|
||||
Returns:
|
||||
dict: Status information
|
||||
"""
|
||||
return {
|
||||
'connected': self._connected,
|
||||
'initialized': self._initialized,
|
||||
'homed': self._homed,
|
||||
'ready': self._ready,
|
||||
'error': self._error_state
|
||||
}
|
||||
|
||||
def prepare_for_scan(self, params: Dict) -> bool:
|
||||
"""
|
||||
Prepare T3R device for scanning operation
|
||||
|
||||
Args:
|
||||
params: Scan parameters dictionary
|
||||
|
||||
Returns:
|
||||
bool: True if preparation successful
|
||||
"""
|
||||
print("DEBUG: Preparing T3R device for scan")
|
||||
print(f" Number of scans: {params.get('num_scans', 1)}")
|
||||
print(f" Trigger voltage: {params.get('trigger_voltage', 0)}V")
|
||||
|
||||
if not self._ready:
|
||||
print("ERROR: T3R device not ready for scanning")
|
||||
return False
|
||||
|
||||
# Configure device for scan parameters
|
||||
# Stub implementation
|
||||
time.sleep(0.1)
|
||||
|
||||
print("INFO: T3R device ready for scanning")
|
||||
return True
|
||||
|
||||
def trigger_acquisition(self) -> bool:
|
||||
"""
|
||||
Trigger a data acquisition event
|
||||
|
||||
Returns:
|
||||
bool: True if trigger successful
|
||||
"""
|
||||
if not self._ready:
|
||||
print("ERROR: Cannot trigger - device not ready")
|
||||
return False
|
||||
|
||||
print("DEBUG: Triggering acquisition")
|
||||
# Stub - would send trigger command
|
||||
return True
|
||||
|
||||
def read_position(self) -> Optional[float]:
|
||||
"""
|
||||
Read current position from T3R device
|
||||
|
||||
Returns:
|
||||
float: Current position value, or None if error
|
||||
"""
|
||||
if not self._ready:
|
||||
return None
|
||||
|
||||
# Stub - return dummy position
|
||||
return 0.0
|
||||
|
||||
def set_timing_parameters(self, acquisition_time: float,
|
||||
delay_time: float) -> bool:
|
||||
"""
|
||||
Set timing parameters for acquisition
|
||||
|
||||
Args:
|
||||
acquisition_time: Acquisition window time in seconds
|
||||
delay_time: Delay before acquisition in seconds
|
||||
|
||||
Returns:
|
||||
bool: True if parameters set successfully
|
||||
"""
|
||||
print(f"DEBUG: Setting timing - acq: {acquisition_time}s, delay: {delay_time}s")
|
||||
|
||||
if not self._connected:
|
||||
print("ERROR: Device not connected")
|
||||
return False
|
||||
|
||||
# Stub implementation
|
||||
return True
|
||||
|
||||
def get_error_status(self) -> Dict[str, any]:
|
||||
"""
|
||||
Get detailed error status
|
||||
|
||||
Returns:
|
||||
dict: Error status information
|
||||
"""
|
||||
return {
|
||||
'has_error': self._error_state,
|
||||
'error_code': 0,
|
||||
'error_message': 'No error'
|
||||
}
|
||||
|
||||
def reset(self) -> bool:
|
||||
"""
|
||||
Reset T3R device to initial state
|
||||
|
||||
Returns:
|
||||
bool: True if reset successful
|
||||
"""
|
||||
print("DEBUG: Resetting T3R device")
|
||||
|
||||
if not self._connected:
|
||||
return False
|
||||
|
||||
self._error_state = False
|
||||
self._homed = False
|
||||
self._ready = False
|
||||
|
||||
# Re-initialize
|
||||
return self._initialize()
|
||||
@@ -0,0 +1,651 @@
|
||||
"""
|
||||
ThorLabs MLS Stage Controller
|
||||
Handles communication with ThorLabs MLS 3-axis positioning stage via BBD203 motor controller
|
||||
|
||||
The ThorLabs MLS stage provides precision X/Y/Z positioning for scanning operations.
|
||||
This implementation uses the BBD203 3-channel motor controller with the APT protocol.
|
||||
|
||||
Channel Mapping:
|
||||
- Channel 1: X-axis
|
||||
- Channel 2: Y-axis
|
||||
- Channel 3: Z-axis (optional)
|
||||
|
||||
Copyright (C) 2025 Thomas Ales
|
||||
Licensed under GNU General Public License v2.0
|
||||
"""
|
||||
|
||||
import time
|
||||
from typing import Dict, Optional
|
||||
|
||||
from hardware.bbd203_driver import BBD203Driver
|
||||
from hardware.bbd203_protocol import TriggerMode
|
||||
from hardware.stage_settings import StageSettings
|
||||
|
||||
|
||||
class ThorLabsStage:
|
||||
"""
|
||||
Controller for ThorLabs MLS positioning stage using BBD203 motor controller
|
||||
|
||||
Provides methods for:
|
||||
- Connecting/disconnecting from stage
|
||||
- Homing axes
|
||||
- Position control and readout
|
||||
- Status monitoring
|
||||
"""
|
||||
|
||||
# Channel mapping
|
||||
X_AXIS = 1
|
||||
Y_AXIS = 2
|
||||
Z_AXIS = 3
|
||||
|
||||
def __init__(self, encoder_counts_per_mm: int = 20000, settings_file: Optional[str] = None):
|
||||
"""
|
||||
Initialize ThorLabs stage controller
|
||||
|
||||
Args:
|
||||
encoder_counts_per_mm: Encoder resolution (default: 20000 for MLS203)
|
||||
settings_file: Path to settings file (default: ~/.nuescan/stage_settings.json)
|
||||
"""
|
||||
self._driver = BBD203Driver(encoder_counts_per_mm)
|
||||
self._port = None
|
||||
|
||||
# Settings manager
|
||||
self.settings = StageSettings(settings_file)
|
||||
|
||||
# Scanning state
|
||||
self._scanning = False
|
||||
|
||||
# Default velocity and acceleration (can be overridden by settings)
|
||||
self._default_velocity = 1.0 # mm/s
|
||||
self._default_accel = 5.0 # mm/s²
|
||||
|
||||
print("INFO: ThorLabs Stage controller initialized (BBD203 driver)")
|
||||
|
||||
def connect(self, serial_number: str, baudrate: int = 115200) -> bool:
|
||||
"""
|
||||
Connect to ThorLabs stage via BBD203 controller using serial number
|
||||
|
||||
The serial number is printed on the BBD203 controller (e.g., '83123456').
|
||||
The driver will automatically find the USB device and connect.
|
||||
|
||||
Args:
|
||||
serial_number: BBD203 device serial number
|
||||
baudrate: Baud rate (default: 115200)
|
||||
|
||||
Returns:
|
||||
bool: True if connection successful
|
||||
"""
|
||||
print(f"DEBUG: Connecting to BBD203/MLS Stage with serial number {serial_number}")
|
||||
|
||||
# Connect by serial number - driver will auto-find the port
|
||||
if not self._driver.connect_by_serial(serial_number, baudrate):
|
||||
return False
|
||||
|
||||
self._port = serial_number # Store serial for reference
|
||||
|
||||
# Enable all channels
|
||||
time.sleep(0.2)
|
||||
self._driver.enable_channel(self.X_AXIS, True)
|
||||
time.sleep(0.1)
|
||||
self._driver.enable_channel(self.Y_AXIS, True)
|
||||
time.sleep(0.1)
|
||||
self._driver.enable_channel(self.Z_AXIS, True)
|
||||
time.sleep(0.1)
|
||||
|
||||
# Apply startup settings from configuration
|
||||
self.apply_startup_settings()
|
||||
|
||||
print("INFO: Stage connected and channels enabled")
|
||||
return True
|
||||
|
||||
def disconnect(self) -> bool:
|
||||
"""
|
||||
Disconnect from ThorLabs stage
|
||||
|
||||
Returns:
|
||||
bool: True if disconnection successful
|
||||
"""
|
||||
self._scanning = False
|
||||
return self._driver.disconnect()
|
||||
|
||||
def is_connected(self) -> bool:
|
||||
"""Check if stage is connected"""
|
||||
return self._driver.is_connected()
|
||||
|
||||
# ==================== Homing ====================
|
||||
|
||||
def home_all_axes(self, wait: bool = True, timeout: float = 60.0) -> bool:
|
||||
"""
|
||||
Home all axes (X, Y, Z)
|
||||
|
||||
Args:
|
||||
wait: If True, block until homing complete
|
||||
timeout: Timeout in seconds
|
||||
|
||||
Returns:
|
||||
bool: True if homing successful
|
||||
"""
|
||||
print("DEBUG: Homing all axes")
|
||||
|
||||
if not self.is_connected():
|
||||
print("ERROR: Cannot home - stage not connected")
|
||||
return False
|
||||
|
||||
return self._driver.home_all_channels(wait=wait, timeout=timeout)
|
||||
|
||||
def home_axis(self, axis: str, wait: bool = True, timeout: float = 30.0) -> bool:
|
||||
"""
|
||||
Home a specific axis
|
||||
|
||||
Args:
|
||||
axis: Axis to home ('X', 'Y', or 'Z')
|
||||
wait: If True, block until homing complete
|
||||
timeout: Timeout in seconds
|
||||
|
||||
Returns:
|
||||
bool: True if homing successful
|
||||
"""
|
||||
axis = axis.upper()
|
||||
if axis not in ['X', 'Y', 'Z']:
|
||||
print(f"ERROR: Invalid axis: {axis}")
|
||||
return False
|
||||
|
||||
channel = {'X': self.X_AXIS, 'Y': self.Y_AXIS, 'Z': self.Z_AXIS}[axis]
|
||||
|
||||
print(f"DEBUG: Homing {axis} axis (channel {channel})")
|
||||
|
||||
return self._driver.home_channel(channel, wait=wait, timeout=timeout)
|
||||
|
||||
# ==================== Motion Control ====================
|
||||
|
||||
def move_absolute(self, x: Optional[float] = None,
|
||||
y: Optional[float] = None,
|
||||
z: Optional[float] = None,
|
||||
wait: bool = False) -> bool:
|
||||
"""
|
||||
Move to absolute position
|
||||
|
||||
Args:
|
||||
x: X position in mm (None to leave unchanged)
|
||||
y: Y position in mm (None to leave unchanged)
|
||||
z: Z position in mm (None to leave unchanged)
|
||||
wait: If True, block until move complete
|
||||
|
||||
Returns:
|
||||
bool: True if move successful
|
||||
"""
|
||||
if not self.is_connected():
|
||||
print("ERROR: Stage not connected")
|
||||
return False
|
||||
|
||||
success = True
|
||||
|
||||
# Move each axis that was specified
|
||||
if x is not None:
|
||||
if not self._driver.move_absolute(self.X_AXIS, x, wait=wait):
|
||||
success = False
|
||||
|
||||
if y is not None:
|
||||
if not self._driver.move_absolute(self.Y_AXIS, y, wait=wait):
|
||||
success = False
|
||||
|
||||
if z is not None:
|
||||
if not self._driver.move_absolute(self.Z_AXIS, z, wait=wait):
|
||||
success = False
|
||||
|
||||
return success
|
||||
|
||||
def move_relative(self, dx: float = 0.0, dy: float = 0.0, dz: float = 0.0,
|
||||
wait: bool = False) -> bool:
|
||||
"""
|
||||
Move relative to current position
|
||||
|
||||
Args:
|
||||
dx: X displacement in mm
|
||||
dy: Y displacement in mm
|
||||
dz: Z displacement in mm
|
||||
wait: If True, block until move complete
|
||||
|
||||
Returns:
|
||||
bool: True if move successful
|
||||
"""
|
||||
if not self.is_connected():
|
||||
print("ERROR: Stage not connected")
|
||||
return False
|
||||
|
||||
success = True
|
||||
|
||||
if dx != 0.0:
|
||||
if not self._driver.move_relative(self.X_AXIS, dx, wait=wait):
|
||||
success = False
|
||||
|
||||
if dy != 0.0:
|
||||
if not self._driver.move_relative(self.Y_AXIS, dy, wait=wait):
|
||||
success = False
|
||||
|
||||
if dz != 0.0:
|
||||
if not self._driver.move_relative(self.Z_AXIS, dz, wait=wait):
|
||||
success = False
|
||||
|
||||
return success
|
||||
|
||||
def stop_all(self, immediate: bool = True) -> bool:
|
||||
"""
|
||||
Stop all motion
|
||||
|
||||
Args:
|
||||
immediate: If True, stop immediately; if False, decelerate
|
||||
|
||||
Returns:
|
||||
bool: True if stop successful
|
||||
"""
|
||||
return self._driver.stop(0, immediate) # Channel 0 = all channels
|
||||
|
||||
# ==================== Position and Status ====================
|
||||
|
||||
def get_position(self) -> Dict[str, float]:
|
||||
"""
|
||||
Get current position
|
||||
|
||||
Returns:
|
||||
dict: Current X, Y, Z positions in mm
|
||||
"""
|
||||
return {
|
||||
'x': self._driver.get_position(self.X_AXIS) or 0.0,
|
||||
'y': self._driver.get_position(self.Y_AXIS) or 0.0,
|
||||
'z': self._driver.get_position(self.Z_AXIS) or 0.0
|
||||
}
|
||||
|
||||
def get_status(self) -> Dict[str, bool]:
|
||||
"""
|
||||
Get current stage status
|
||||
|
||||
Returns:
|
||||
dict: Status information compatible with main_window expectations
|
||||
"""
|
||||
x_status = self._driver.get_channel_status(self.X_AXIS) or {}
|
||||
y_status = self._driver.get_channel_status(self.Y_AXIS) or {}
|
||||
z_status = self._driver.get_channel_status(self.Z_AXIS) or {}
|
||||
|
||||
# Determine if any axis is moving
|
||||
moving = (x_status.get('moving', False) or
|
||||
y_status.get('moving', False) or
|
||||
z_status.get('moving', False))
|
||||
|
||||
# Determine if stage is ready (all enabled axes are homed)
|
||||
x_ready = x_status.get('enabled', False) and x_status.get('homed', False)
|
||||
y_ready = y_status.get('enabled', False) and y_status.get('homed', False)
|
||||
ready = x_ready and y_ready # Z is optional
|
||||
|
||||
return {
|
||||
'connected': self.is_connected(),
|
||||
'x_homed': x_status.get('homed', False),
|
||||
'y_homed': y_status.get('homed', False),
|
||||
'z_homed': z_status.get('homed', False),
|
||||
'ready': ready,
|
||||
'scanning': self._scanning,
|
||||
'moving': moving
|
||||
}
|
||||
|
||||
# ==================== Velocity Control ====================
|
||||
|
||||
def set_velocity(self, velocity_mm_s: float, accel_mm_s2: float,
|
||||
axis: Optional[str] = None) -> bool:
|
||||
"""
|
||||
Set velocity and acceleration parameters
|
||||
|
||||
Args:
|
||||
velocity_mm_s: Maximum velocity in mm/s
|
||||
accel_mm_s2: Acceleration in mm/s²
|
||||
axis: Specific axis ('X', 'Y', 'Z'), or None for all axes
|
||||
|
||||
Returns:
|
||||
bool: True if parameters set successfully
|
||||
"""
|
||||
if axis:
|
||||
axis = axis.upper()
|
||||
if axis not in ['X', 'Y', 'Z']:
|
||||
print(f"ERROR: Invalid axis: {axis}")
|
||||
return False
|
||||
|
||||
channel = {'X': self.X_AXIS, 'Y': self.Y_AXIS, 'Z': self.Z_AXIS}[axis]
|
||||
return self._driver.set_velocity_params(channel, velocity_mm_s, accel_mm_s2)
|
||||
else:
|
||||
# Set for all axes
|
||||
success = True
|
||||
for channel in [self.X_AXIS, self.Y_AXIS, self.Z_AXIS]:
|
||||
if not self._driver.set_velocity_params(channel, velocity_mm_s, accel_mm_s2):
|
||||
success = False
|
||||
time.sleep(0.05)
|
||||
return success
|
||||
|
||||
# ==================== Scan Support ====================
|
||||
|
||||
def prepare_for_scan(self, params: Dict) -> bool:
|
||||
"""
|
||||
Prepare stage for scanning operation
|
||||
|
||||
Args:
|
||||
params: Scan parameters dictionary
|
||||
|
||||
Returns:
|
||||
bool: True if preparation successful
|
||||
"""
|
||||
print("DEBUG: Preparing stage for scan")
|
||||
|
||||
status = self.get_status()
|
||||
if not status['ready']:
|
||||
print("ERROR: Stage not ready for scanning")
|
||||
return False
|
||||
|
||||
# Move to start position
|
||||
x_start = params.get('x_start', 0)
|
||||
y_start = params.get('y_start', 0)
|
||||
|
||||
print(f" Moving to scan start: X={x_start} mm, Y={y_start} mm")
|
||||
|
||||
if not self.move_absolute(x=x_start, y=y_start, wait=True):
|
||||
print("ERROR: Failed to move to start position")
|
||||
return False
|
||||
|
||||
self._scanning = True
|
||||
print("INFO: Stage ready for scanning")
|
||||
return True
|
||||
|
||||
def stop_scan(self) -> bool:
|
||||
"""
|
||||
Stop current scan operation
|
||||
|
||||
Returns:
|
||||
bool: True if stop successful
|
||||
"""
|
||||
print("DEBUG: Stopping scan")
|
||||
self._scanning = False
|
||||
return self.stop_all(immediate=True)
|
||||
|
||||
# ==================== Utility Methods ====================
|
||||
|
||||
def identify(self, channel: Optional[int] = None) -> bool:
|
||||
"""
|
||||
Flash front panel LEDs to identify controller
|
||||
|
||||
Args:
|
||||
channel: Specific channel (1, 2, 3), or None for all
|
||||
|
||||
Returns:
|
||||
bool: True if command sent successfully
|
||||
"""
|
||||
if channel is None:
|
||||
# Identify all channels
|
||||
for ch in [1, 2, 3]:
|
||||
self._driver.identify(ch)
|
||||
time.sleep(0.1)
|
||||
return True
|
||||
else:
|
||||
return self._driver.identify(channel)
|
||||
|
||||
@staticmethod
|
||||
def list_available_ports():
|
||||
"""
|
||||
List available serial ports (deprecated - use list_devices instead)
|
||||
|
||||
Returns:
|
||||
list: Available port names
|
||||
"""
|
||||
return BBD203Driver.list_available_ports()
|
||||
|
||||
@staticmethod
|
||||
def list_devices():
|
||||
"""
|
||||
List all connected ThorLabs BBD203 devices
|
||||
|
||||
Returns:
|
||||
list: List of dicts with device info including 'serial' and 'port'
|
||||
"""
|
||||
return BBD203Driver.list_thorlabs_devices()
|
||||
|
||||
# ==================== Settings Management ====================
|
||||
|
||||
def apply_startup_settings(self) -> bool:
|
||||
"""
|
||||
Apply saved settings to the stage on startup
|
||||
|
||||
This includes:
|
||||
- Velocity parameters for all axes
|
||||
- Acceleration parameters for all axes
|
||||
- Trigger configuration for all axes
|
||||
|
||||
Returns:
|
||||
bool: True if all settings applied successfully
|
||||
"""
|
||||
print("INFO: Applying startup settings to stage")
|
||||
|
||||
success = True
|
||||
|
||||
# Apply velocity and acceleration settings
|
||||
velocities = self.settings.get_all_velocities()
|
||||
accelerations = self.settings.get_all_accelerations()
|
||||
|
||||
print(f" Velocity settings: X={velocities['x_axis']} mm/s, "
|
||||
f"Y={velocities['y_axis']} mm/s, Z={velocities['z_axis']} mm/s")
|
||||
print(f" Acceleration settings: X={accelerations['x_axis']} mm/s², "
|
||||
f"Y={accelerations['y_axis']} mm/s², Z={accelerations['z_axis']} mm/s²")
|
||||
|
||||
# Set velocity/acceleration for each axis
|
||||
if not self._driver.set_velocity_params(
|
||||
self.X_AXIS, velocities['x_axis'], accelerations['x_axis']
|
||||
):
|
||||
success = False
|
||||
time.sleep(0.05)
|
||||
|
||||
if not self._driver.set_velocity_params(
|
||||
self.Y_AXIS, velocities['y_axis'], accelerations['y_axis']
|
||||
):
|
||||
success = False
|
||||
time.sleep(0.05)
|
||||
|
||||
if not self._driver.set_velocity_params(
|
||||
self.Z_AXIS, velocities['z_axis'], accelerations['z_axis']
|
||||
):
|
||||
success = False
|
||||
time.sleep(0.05)
|
||||
|
||||
# Apply trigger configuration for each axis
|
||||
for axis_name, channel in [('x_axis', self.X_AXIS),
|
||||
('y_axis', self.Y_AXIS),
|
||||
('z_axis', self.Z_AXIS)]:
|
||||
trigger_config = self.settings.get_trigger_config(axis_name)
|
||||
|
||||
if not self._driver.set_trigger_mode(
|
||||
channel,
|
||||
trigger_config['mode'],
|
||||
trigger_config['polarity'],
|
||||
trigger_config['start_pos_fwd'],
|
||||
trigger_config['start_pos_rev'],
|
||||
trigger_config['interval_fwd'],
|
||||
trigger_config['interval_rev']
|
||||
):
|
||||
success = False
|
||||
time.sleep(0.05)
|
||||
|
||||
if success:
|
||||
print("INFO: All startup settings applied successfully")
|
||||
else:
|
||||
print("WARNING: Some startup settings failed to apply")
|
||||
|
||||
return success
|
||||
|
||||
def save_current_settings(self) -> bool:
|
||||
"""
|
||||
Save current settings to file
|
||||
|
||||
Returns:
|
||||
bool: True if saved successfully
|
||||
"""
|
||||
return self.settings.save()
|
||||
|
||||
def reload_settings(self) -> bool:
|
||||
"""
|
||||
Reload settings from file
|
||||
|
||||
Returns:
|
||||
bool: True if reloaded successfully
|
||||
"""
|
||||
return self.settings.load()
|
||||
|
||||
def configure_velocity(self, x: Optional[float] = None,
|
||||
y: Optional[float] = None,
|
||||
z: Optional[float] = None,
|
||||
save: bool = True) -> bool:
|
||||
"""
|
||||
Configure velocity for one or more axes
|
||||
|
||||
Args:
|
||||
x: X-axis velocity in mm/s (None to keep current)
|
||||
y: Y-axis velocity in mm/s (None to keep current)
|
||||
z: Z-axis velocity in mm/s (None to keep current)
|
||||
save: Save settings to file after updating
|
||||
|
||||
Returns:
|
||||
bool: True if configuration successful
|
||||
"""
|
||||
success = True
|
||||
|
||||
if x is not None:
|
||||
self.settings.set_velocity('x_axis', x)
|
||||
accel = self.settings.get_acceleration('x_axis')
|
||||
if self.is_connected():
|
||||
success &= self._driver.set_velocity_params(self.X_AXIS, x, accel)
|
||||
time.sleep(0.05)
|
||||
|
||||
if y is not None:
|
||||
self.settings.set_velocity('y_axis', y)
|
||||
accel = self.settings.get_acceleration('y_axis')
|
||||
if self.is_connected():
|
||||
success &= self._driver.set_velocity_params(self.Y_AXIS, y, accel)
|
||||
time.sleep(0.05)
|
||||
|
||||
if z is not None:
|
||||
self.settings.set_velocity('z_axis', z)
|
||||
accel = self.settings.get_acceleration('z_axis')
|
||||
if self.is_connected():
|
||||
success &= self._driver.set_velocity_params(self.Z_AXIS, z, accel)
|
||||
time.sleep(0.05)
|
||||
|
||||
if save:
|
||||
self.settings.save()
|
||||
|
||||
return success
|
||||
|
||||
def configure_acceleration(self, x: Optional[float] = None,
|
||||
y: Optional[float] = None,
|
||||
z: Optional[float] = None,
|
||||
save: bool = True) -> bool:
|
||||
"""
|
||||
Configure acceleration for one or more axes
|
||||
|
||||
Args:
|
||||
x: X-axis acceleration in mm/s² (None to keep current)
|
||||
y: Y-axis acceleration in mm/s² (None to keep current)
|
||||
z: Z-axis acceleration in mm/s² (None to keep current)
|
||||
save: Save settings to file after updating
|
||||
|
||||
Returns:
|
||||
bool: True if configuration successful
|
||||
"""
|
||||
success = True
|
||||
|
||||
if x is not None:
|
||||
self.settings.set_acceleration('x_axis', x)
|
||||
vel = self.settings.get_velocity('x_axis')
|
||||
if self.is_connected():
|
||||
success &= self._driver.set_velocity_params(self.X_AXIS, vel, x)
|
||||
time.sleep(0.05)
|
||||
|
||||
if y is not None:
|
||||
self.settings.set_acceleration('y_axis', y)
|
||||
vel = self.settings.get_velocity('y_axis')
|
||||
if self.is_connected():
|
||||
success &= self._driver.set_velocity_params(self.Y_AXIS, vel, y)
|
||||
time.sleep(0.05)
|
||||
|
||||
if z is not None:
|
||||
self.settings.set_acceleration('z_axis', z)
|
||||
vel = self.settings.get_velocity('z_axis')
|
||||
if self.is_connected():
|
||||
success &= self._driver.set_velocity_params(self.Z_AXIS, vel, z)
|
||||
time.sleep(0.05)
|
||||
|
||||
if save:
|
||||
self.settings.save()
|
||||
|
||||
return success
|
||||
|
||||
def configure_trigger(self, axis: str, mode: int,
|
||||
polarity: int = 0x01,
|
||||
start_pos_fwd: float = 0.0,
|
||||
start_pos_rev: float = 0.0,
|
||||
interval_fwd: float = 0.0,
|
||||
interval_rev: float = 0.0,
|
||||
save: bool = True) -> bool:
|
||||
"""
|
||||
Configure trigger for specific axis
|
||||
|
||||
Args:
|
||||
axis: Axis name ('X', 'Y', or 'Z')
|
||||
mode: Trigger mode (TriggerMode enum value)
|
||||
polarity: Trigger polarity (0x01 = active high, 0x02 = active low)
|
||||
start_pos_fwd: Start position for forward trigger (mm)
|
||||
start_pos_rev: Start position for reverse trigger (mm)
|
||||
interval_fwd: Interval for forward trigger (mm)
|
||||
interval_rev: Interval for reverse trigger (mm)
|
||||
save: Save settings to file after updating
|
||||
|
||||
Returns:
|
||||
bool: True if configuration successful
|
||||
"""
|
||||
axis = axis.upper()
|
||||
if axis not in ['X', 'Y', 'Z']:
|
||||
print(f"ERROR: Invalid axis: {axis}")
|
||||
return False
|
||||
|
||||
axis_name = f"{axis.lower()}_axis"
|
||||
channel = {'X': self.X_AXIS, 'Y': self.Y_AXIS, 'Z': self.Z_AXIS}[axis]
|
||||
|
||||
# Update settings
|
||||
self.settings.set_trigger_config(
|
||||
axis_name, mode, polarity,
|
||||
start_pos_fwd, start_pos_rev,
|
||||
interval_fwd, interval_rev
|
||||
)
|
||||
|
||||
# Apply to hardware if connected
|
||||
success = True
|
||||
if self.is_connected():
|
||||
success = self._driver.set_trigger_mode(
|
||||
channel, mode, polarity,
|
||||
start_pos_fwd, start_pos_rev,
|
||||
interval_fwd, interval_rev
|
||||
)
|
||||
|
||||
if save:
|
||||
self.settings.save()
|
||||
|
||||
return success
|
||||
|
||||
def get_detailed_status(self) -> Dict:
|
||||
"""
|
||||
Get detailed status of all channels
|
||||
|
||||
Returns:
|
||||
dict: Detailed status information
|
||||
"""
|
||||
return {
|
||||
'connected': self.is_connected(),
|
||||
'scanning': self._scanning,
|
||||
'x_axis': self._driver.get_channel_status(self.X_AXIS),
|
||||
'y_axis': self._driver.get_channel_status(self.Y_AXIS),
|
||||
'z_axis': self._driver.get_channel_status(self.Z_AXIS),
|
||||
'position': self.get_position(),
|
||||
'settings': self.settings.get_all_settings()
|
||||
}
|
||||
Reference in New Issue
Block a user