Thomas Ales [M S E]
670 lines
20 KiB
Python
670 lines
20 KiB
Python
"""
|
|
Genesis SLM MX 532 Laser Core Hardware Control Module
|
|
======================================================
|
|
|
|
This module provides low-level hardware control for the Genesis SLM MX 532 laser
|
|
using NXP I2C-over-serial protocol. It contains reusable classes for serial
|
|
communication, I2C protocol handling, device control, and laser operations.
|
|
|
|
This module is GUI-independent and can be used by both command-line and GUI applications.
|
|
|
|
Protocol Overview:
|
|
-----------------
|
|
The laser uses NXP I2C-over-serial protocol with packet format:
|
|
[0x53] [I2C_ADDR] [LENGTH] [COMMAND_BYTES] [DATA_BYTES] [0x50]
|
|
|
|
For reads:
|
|
[0x53] [ADDR_WRITE] [CMD_LEN] [CMD] [0x53] [ADDR_READ] [DATA_LEN] [0x50]
|
|
|
|
I2C Devices:
|
|
-----------
|
|
- X9119 digital potentiometer at 0x52 - laser current control
|
|
- PCA9555 I/O expander at 0x4a - digital I/O
|
|
- AD5254 digital potentiometer at 0x58 - limits control
|
|
- ADS7828 ADC at 0x90 - sensor readings
|
|
- M24C64 EEPROM at 0xa4 - configuration storage
|
|
|
|
Author: Claude
|
|
Date: 2026-01-24
|
|
"""
|
|
|
|
import time
|
|
from datetime import datetime
|
|
from typing import Optional, List
|
|
from enum import IntEnum
|
|
|
|
import serial
|
|
from serial.tools import list_ports
|
|
|
|
|
|
# ============================================================================
|
|
# Constants - I2C Addresses
|
|
# ============================================================================
|
|
|
|
class I2CAddress(IntEnum):
|
|
"""I2C device addresses (7-bit addresses shifted left by 1)"""
|
|
# X9119 Digital Potentiometer (current control)
|
|
X9119_CURRENT_WRITE = 0x52
|
|
X9119_CURRENT_READ = 0x53
|
|
|
|
# PCA9555 I/O Expanders
|
|
PCA9555_DIO_WRITE = 0x4a # Main control I/O at 0x14a
|
|
PCA9555_DIO_READ = 0x4b
|
|
PCA9555_PSGLUE_WRITE = 0x48 # Power supply glue at 0x148
|
|
PCA9555_PSGLUE_READ = 0x49
|
|
PCA9555_HEAD_WRITE = 0x44 # Head DIO at 0x144
|
|
PCA9555_HEAD_READ = 0x45
|
|
PCA9555_LDD_WRITE = 0x40 # LDD control at 0x140
|
|
PCA9555_LDD_READ = 0x41
|
|
|
|
# AD5254 Digital Potentiometer (limits)
|
|
AD5254_WRITE = 0x58
|
|
|
|
# ADS7828 ADC (sensor readings)
|
|
ADS7828_WRITE = 0x90
|
|
ADS7828_READ = 0x91
|
|
|
|
# M24C64 EEPROM (configuration)
|
|
EEPROM_WRITE = 0xa4
|
|
EEPROM_READ = 0xa5
|
|
|
|
|
|
# ============================================================================
|
|
# Constants - PCA9555 Register Addresses
|
|
# ============================================================================
|
|
|
|
class PCA9555Register(IntEnum):
|
|
"""PCA9555 I/O expander register addresses"""
|
|
INPUT_PORT_0 = 0x00
|
|
INPUT_PORT_1 = 0x01
|
|
OUTPUT_PORT_0 = 0x02
|
|
OUTPUT_PORT_1 = 0x03
|
|
POLARITY_INV_0 = 0x04
|
|
POLARITY_INV_1 = 0x05
|
|
CONFIG_PORT_0 = 0x06
|
|
CONFIG_PORT_1 = 0x07
|
|
|
|
|
|
# ============================================================================
|
|
# Constants - Control Bitmasks
|
|
# ============================================================================
|
|
|
|
class ControlBitmask(IntEnum):
|
|
"""Bitmasks for PCA9555 port 0 control bits"""
|
|
SHUTTER = 0x01 # Bit 0
|
|
CURRENT_MODE = 0x04 # Bit 2
|
|
REMOTE_ENABLE = 0x08 # Bit 3
|
|
ANALOG_ENABLE = 0x10 # Bit 4
|
|
KEYSWITCH = 0x20 # Bit 5
|
|
INTERLOCK = 0x01 # Bit 0 (on different PCA9555)
|
|
LDD_ENABLE = 0x01 # Bit 0 (on LDD PCA9555)
|
|
|
|
|
|
# ============================================================================
|
|
# Serial Communication Layer
|
|
# ============================================================================
|
|
|
|
class SerialComm:
|
|
"""Handles low-level serial port communication"""
|
|
|
|
def __init__(self):
|
|
self.port: Optional[serial.Serial] = None
|
|
self.port_name: str = "/dev/ttyUSB0"
|
|
self.baudrate: int = 9600
|
|
self.timeout: float = 1.0
|
|
self.packet_log: List[str] = []
|
|
|
|
def connect(self, port_name: str, baudrate: int = 9600) -> bool:
|
|
"""
|
|
Connect to serial port
|
|
|
|
Args:
|
|
port_name: Serial port device name
|
|
baudrate: Baud rate (default 9600)
|
|
|
|
Returns:
|
|
True if successful, False otherwise
|
|
"""
|
|
try:
|
|
self.port_name = port_name
|
|
self.baudrate = baudrate
|
|
self.port = serial.Serial(
|
|
port=port_name,
|
|
baudrate=baudrate,
|
|
bytesize=serial.EIGHTBITS,
|
|
parity=serial.PARITY_NONE,
|
|
stopbits=serial.STOPBITS_ONE,
|
|
timeout=self.timeout
|
|
)
|
|
time.sleep(0.1) # Allow port to stabilize
|
|
return True
|
|
except Exception as e:
|
|
print(f"Serial connection error: {e}")
|
|
return False
|
|
|
|
def disconnect(self):
|
|
"""Disconnect from serial port"""
|
|
if self.port and self.port.is_open:
|
|
self.port.close()
|
|
self.port = None
|
|
|
|
def is_connected(self) -> bool:
|
|
"""Check if serial port is connected"""
|
|
return self.port is not None and self.port.is_open
|
|
|
|
def write(self, data: bytes) -> bool:
|
|
"""
|
|
Write data to serial port
|
|
|
|
Args:
|
|
data: Bytes to write
|
|
|
|
Returns:
|
|
True if successful, False otherwise
|
|
"""
|
|
if not self.is_connected():
|
|
return False
|
|
|
|
try:
|
|
self.port.write(data)
|
|
timestamp = datetime.now().strftime("%H:%M:%S.%f")[:-3]
|
|
hex_str = " ".join(f"{b:02x}" for b in data)
|
|
self.packet_log.append(f"[{timestamp}] TX: {hex_str}")
|
|
return True
|
|
except Exception as e:
|
|
print(f"Serial write error: {e}")
|
|
return False
|
|
|
|
def read(self, size: int) -> Optional[bytes]:
|
|
"""
|
|
Read data from serial port
|
|
|
|
Args:
|
|
size: Number of bytes to read
|
|
|
|
Returns:
|
|
Bytes read or None on error
|
|
"""
|
|
if not self.is_connected():
|
|
return None
|
|
|
|
try:
|
|
data = self.port.read(size)
|
|
if data:
|
|
timestamp = datetime.now().strftime("%H:%M:%S.%f")[:-3]
|
|
hex_str = " ".join(f"{b:02x}" for b in data)
|
|
self.packet_log.append(f"[{timestamp}] RX: {hex_str}")
|
|
return data
|
|
except Exception as e:
|
|
print(f"Serial read error: {e}")
|
|
return None
|
|
|
|
def flush(self):
|
|
"""Flush serial port buffers"""
|
|
if self.is_connected():
|
|
self.port.reset_input_buffer()
|
|
self.port.reset_output_buffer()
|
|
|
|
def get_packet_log(self, last_n: int = 100) -> List[str]:
|
|
"""Get last N entries from packet log"""
|
|
return self.packet_log[-last_n:]
|
|
|
|
def clear_packet_log(self):
|
|
"""Clear packet log"""
|
|
self.packet_log.clear()
|
|
|
|
|
|
# ============================================================================
|
|
# I2C Protocol Layer
|
|
# ============================================================================
|
|
|
|
class I2CProtocol:
|
|
"""Handles NXP I2C-over-serial protocol packet construction"""
|
|
|
|
NXP_START = 0x53
|
|
NXP_STOP = 0x50
|
|
|
|
def __init__(self, serial_comm: SerialComm):
|
|
self.serial = serial_comm
|
|
|
|
def write(self, i2c_addr_write: int, cmd: bytes, data: bytes = b'') -> bool:
|
|
"""
|
|
Build and send NXP I2C write packet
|
|
|
|
Packet format: [0x53] [I2C_ADDR] [LENGTH] [COMMAND] [DATA] [0x50]
|
|
|
|
Args:
|
|
i2c_addr_write: I2C write address
|
|
cmd: Command bytes
|
|
data: Data bytes (optional)
|
|
|
|
Returns:
|
|
True if successful, False otherwise
|
|
"""
|
|
if not isinstance(cmd, bytes):
|
|
cmd = bytes([cmd])
|
|
if not isinstance(data, bytes):
|
|
data = bytes(data)
|
|
|
|
length = len(cmd) + len(data)
|
|
packet = bytes([self.NXP_START, i2c_addr_write, length]) + cmd + data + bytes([self.NXP_STOP])
|
|
|
|
return self.serial.write(packet)
|
|
|
|
def read(self, i2c_addr_write: int, cmd: bytes, cmd_len: int, data_len: int) -> Optional[bytes]:
|
|
"""
|
|
Build and send NXP I2C read packet, return data
|
|
|
|
Packet format for read:
|
|
[0x53] [ADDR_WRITE] [CMD_LEN] [CMD] [0x53] [ADDR_READ] [DATA_LEN] [0x50]
|
|
|
|
Args:
|
|
i2c_addr_write: I2C write address
|
|
cmd: Command bytes
|
|
cmd_len: Length of command
|
|
data_len: Expected data length to read
|
|
|
|
Returns:
|
|
Data bytes read or None on error
|
|
"""
|
|
if not isinstance(cmd, bytes):
|
|
cmd = bytes([cmd])
|
|
|
|
i2c_addr_read = i2c_addr_write | 0x01 # Set read bit
|
|
|
|
# First part: write command
|
|
packet_write = bytes([self.NXP_START, i2c_addr_write, cmd_len]) + cmd
|
|
|
|
# Second part: read data
|
|
packet_read = bytes([self.NXP_START, i2c_addr_read, data_len, self.NXP_STOP])
|
|
|
|
packet = packet_write + packet_read
|
|
|
|
if not self.serial.write(packet):
|
|
return None
|
|
|
|
# Read response
|
|
# Expected response: [data_bytes]
|
|
time.sleep(0.05) # Give device time to respond
|
|
response = self.serial.read(data_len)
|
|
|
|
return response if response and len(response) == data_len else None
|
|
|
|
|
|
# ============================================================================
|
|
# I2C Device Functions
|
|
# ============================================================================
|
|
|
|
class I2CDevices:
|
|
"""High-level I2C device control functions"""
|
|
|
|
def __init__(self, protocol: I2CProtocol):
|
|
self.protocol = protocol
|
|
|
|
# ------------------------------------------------------------------------
|
|
# PCA9555 I/O Expander Functions
|
|
# ------------------------------------------------------------------------
|
|
|
|
def pca9555_read_port(self, addr_write: int, port: int) -> Optional[int]:
|
|
"""
|
|
Read PCA9555 port register
|
|
|
|
Args:
|
|
addr_write: I2C write address
|
|
port: Register address (0-7)
|
|
|
|
Returns:
|
|
Port value (0-255) or None on error
|
|
"""
|
|
data = self.protocol.read(addr_write, bytes([port]), 1, 1)
|
|
return data[0] if data else None
|
|
|
|
def pca9555_write_port(self, addr_write: int, port: int, value: int) -> bool:
|
|
"""
|
|
Write PCA9555 port register
|
|
|
|
Args:
|
|
addr_write: I2C write address
|
|
port: Register address (0-7)
|
|
value: Value to write (0-255)
|
|
|
|
Returns:
|
|
True if successful, False otherwise
|
|
"""
|
|
return self.protocol.write(addr_write, bytes([port]), bytes([value]))
|
|
|
|
def pca9555_set_bit(self, addr_write: int, port: int, bitmask: int, state: bool) -> bool:
|
|
"""
|
|
Set or clear specific bit on PCA9555
|
|
|
|
Args:
|
|
addr_write: I2C write address
|
|
port: Register address
|
|
bitmask: Bit mask (e.g., 0x01 for bit 0)
|
|
state: True to set bit, False to clear
|
|
|
|
Returns:
|
|
True if successful, False otherwise
|
|
"""
|
|
# Read current value
|
|
current = self.pca9555_read_port(addr_write, port)
|
|
if current is None:
|
|
return False
|
|
|
|
# Modify bit
|
|
if state:
|
|
new_value = current | bitmask
|
|
else:
|
|
new_value = current & ~bitmask
|
|
|
|
# Write back
|
|
return self.pca9555_write_port(addr_write, port, new_value)
|
|
|
|
# ------------------------------------------------------------------------
|
|
# X9119 Digital Potentiometer Functions
|
|
# ------------------------------------------------------------------------
|
|
|
|
def x9119_write_wiper(self, addr_write: int, value: int) -> bool:
|
|
"""
|
|
Set X9119 wiper position
|
|
|
|
Args:
|
|
addr_write: I2C write address
|
|
value: Wiper position (0-1023, 10-bit)
|
|
|
|
Returns:
|
|
True if successful, False otherwise
|
|
"""
|
|
# X9119 write wiper command: 0xa0 followed by 2 bytes (10-bit value)
|
|
# Value is split into two bytes: MSB contains upper 2 bits, LSB contains lower 8 bits
|
|
value = max(0, min(1023, value)) # Clamp to valid range
|
|
|
|
msb = (value >> 8) & 0x03 # Upper 2 bits
|
|
lsb = value & 0xFF # Lower 8 bits
|
|
|
|
cmd = bytes([0xa0])
|
|
data = bytes([msb, lsb])
|
|
|
|
return self.protocol.write(addr_write, cmd, data)
|
|
|
|
# ------------------------------------------------------------------------
|
|
# ADS7828 ADC Functions
|
|
# ------------------------------------------------------------------------
|
|
|
|
def ads7828_read(self, addr_write: int, channel: int) -> Optional[int]:
|
|
"""
|
|
Read ADS7828 ADC channel
|
|
|
|
Args:
|
|
addr_write: I2C write address
|
|
channel: Channel to read (0-7)
|
|
|
|
Returns:
|
|
12-bit ADC value (0-4095) or None on error
|
|
"""
|
|
# ADS7828 command byte format:
|
|
# Bit 7: SD (single-ended=1)
|
|
# Bits 6-4: Channel select
|
|
# Bit 3: PD1 (power-down mode)
|
|
# Bit 2: PD0 (power-down mode)
|
|
# Bits 1-0: Don't care
|
|
|
|
# For single-ended read with internal reference on
|
|
cmd_byte = 0x80 | ((channel & 0x07) << 4) | 0x0c
|
|
|
|
data = self.protocol.read(addr_write, bytes([cmd_byte]), 1, 2)
|
|
if not data or len(data) != 2:
|
|
return None
|
|
|
|
# Combine two bytes into 12-bit value
|
|
value = (data[0] << 8) | data[1]
|
|
value = (value >> 4) & 0x0FFF # Extract 12-bit value
|
|
|
|
return value
|
|
|
|
|
|
# ============================================================================
|
|
# Laser Control Layer
|
|
# ============================================================================
|
|
|
|
class LaserControl:
|
|
"""High-level laser control functions"""
|
|
|
|
def __init__(self, devices: I2CDevices):
|
|
self.devices = devices
|
|
self._safe_state_active = False
|
|
|
|
# ------------------------------------------------------------------------
|
|
# SET Commands
|
|
# ------------------------------------------------------------------------
|
|
|
|
def set_current(self, value: int) -> bool:
|
|
"""
|
|
Set laser current command (CCMD=)
|
|
|
|
Args:
|
|
value: Current value (0-1023)
|
|
|
|
Returns:
|
|
True if successful, False otherwise
|
|
"""
|
|
return self.devices.x9119_write_wiper(I2CAddress.X9119_CURRENT_WRITE, value)
|
|
|
|
def set_power_cmd(self, value: int) -> bool:
|
|
"""
|
|
Set power command (PCMD=)
|
|
|
|
Note: Uses same device as current control, different channel
|
|
|
|
Args:
|
|
value: Power value (0-1023)
|
|
|
|
Returns:
|
|
True if successful, False otherwise
|
|
"""
|
|
return self.devices.x9119_write_wiper(I2CAddress.X9119_CURRENT_WRITE, value)
|
|
|
|
def set_shutter(self, state: bool) -> bool:
|
|
"""
|
|
Set shutter state (SHCMD=)
|
|
|
|
Args:
|
|
state: True for open, False for closed
|
|
|
|
Returns:
|
|
True if successful, False otherwise
|
|
"""
|
|
return self.devices.pca9555_set_bit(
|
|
I2CAddress.PCA9555_DIO_WRITE,
|
|
PCA9555Register.OUTPUT_PORT_0,
|
|
ControlBitmask.SHUTTER,
|
|
state
|
|
)
|
|
|
|
def set_keyswitch(self, state: bool) -> bool:
|
|
"""
|
|
Set keyswitch state (KSWCMD=)
|
|
|
|
Args:
|
|
state: True for on, False for off
|
|
|
|
Returns:
|
|
True if successful, False otherwise
|
|
"""
|
|
return self.devices.pca9555_set_bit(
|
|
I2CAddress.PCA9555_DIO_WRITE,
|
|
PCA9555Register.OUTPUT_PORT_0,
|
|
ControlBitmask.KEYSWITCH,
|
|
state
|
|
)
|
|
|
|
def set_current_mode(self, state: bool) -> bool:
|
|
"""
|
|
Set current mode (CMODECMD=)
|
|
|
|
Args:
|
|
state: True for on, False for off
|
|
|
|
Returns:
|
|
True if successful, False otherwise
|
|
"""
|
|
return self.devices.pca9555_set_bit(
|
|
I2CAddress.PCA9555_DIO_WRITE,
|
|
PCA9555Register.OUTPUT_PORT_0,
|
|
ControlBitmask.CURRENT_MODE,
|
|
state
|
|
)
|
|
|
|
def set_analog_enable(self, state: bool) -> bool:
|
|
"""
|
|
Set analog input enable (ANACMD=)
|
|
|
|
Args:
|
|
state: True for enabled, False for disabled
|
|
|
|
Returns:
|
|
True if successful, False otherwise
|
|
"""
|
|
return self.devices.pca9555_set_bit(
|
|
I2CAddress.PCA9555_DIO_WRITE,
|
|
PCA9555Register.OUTPUT_PORT_0,
|
|
ControlBitmask.ANALOG_ENABLE,
|
|
state
|
|
)
|
|
|
|
def set_remote_enable(self, state: bool) -> bool:
|
|
"""
|
|
Set remote enable (REM=)
|
|
|
|
Args:
|
|
state: True for enabled, False for disabled
|
|
|
|
Returns:
|
|
True if successful, False otherwise
|
|
"""
|
|
return self.devices.pca9555_set_bit(
|
|
I2CAddress.PCA9555_DIO_WRITE,
|
|
PCA9555Register.OUTPUT_PORT_0,
|
|
ControlBitmask.REMOTE_ENABLE,
|
|
state
|
|
)
|
|
|
|
# ------------------------------------------------------------------------
|
|
# GET/Query Commands
|
|
# ------------------------------------------------------------------------
|
|
|
|
def get_current_actual(self) -> Optional[float]:
|
|
"""
|
|
Get actual current reading
|
|
|
|
Returns:
|
|
Current in arbitrary units (0-4095) or None on error
|
|
"""
|
|
# Read from ADS7828, channel based on command 0x84
|
|
# Command 0x84 suggests channel 0
|
|
value = self.devices.ads7828_read(I2CAddress.ADS7828_WRITE, 0)
|
|
return value if value is not None else None
|
|
|
|
def get_interlock_status(self) -> Optional[bool]:
|
|
"""
|
|
Get interlock status
|
|
|
|
Returns:
|
|
True if interlock OK, False if fault, None on error
|
|
"""
|
|
value = self.devices.pca9555_read_port(
|
|
I2CAddress.PCA9555_PSGLUE_WRITE,
|
|
PCA9555Register.INPUT_PORT_0
|
|
)
|
|
if value is None:
|
|
return None
|
|
|
|
# Bit 0 indicates interlock status
|
|
return bool(value & ControlBitmask.INTERLOCK)
|
|
|
|
def get_ldd_enable_status(self) -> Optional[bool]:
|
|
"""
|
|
Get laser diode driver enable status
|
|
|
|
Returns:
|
|
True if enabled, False if disabled, None on error
|
|
"""
|
|
value = self.devices.pca9555_read_port(
|
|
I2CAddress.PCA9555_LDD_WRITE,
|
|
PCA9555Register.INPUT_PORT_0
|
|
)
|
|
if value is None:
|
|
return None
|
|
|
|
return bool(value & ControlBitmask.LDD_ENABLE)
|
|
|
|
def get_psglue_in_status(self) -> Optional[int]:
|
|
"""
|
|
Get power supply glue input status
|
|
|
|
Returns:
|
|
Port value or None on error
|
|
"""
|
|
return self.devices.pca9555_read_port(
|
|
I2CAddress.PCA9555_PSGLUE_WRITE,
|
|
PCA9555Register.INPUT_PORT_0
|
|
)
|
|
|
|
def get_psglue_out_status(self) -> Optional[int]:
|
|
"""
|
|
Get power supply glue output status
|
|
|
|
Returns:
|
|
Port value or None on error
|
|
"""
|
|
return self.devices.pca9555_read_port(
|
|
I2CAddress.PCA9555_DIO_WRITE,
|
|
PCA9555Register.OUTPUT_PORT_0
|
|
)
|
|
|
|
def get_head_dio_status(self) -> Optional[int]:
|
|
"""
|
|
Get head DIO status
|
|
|
|
Returns:
|
|
Port value or None on error
|
|
"""
|
|
return self.devices.pca9555_read_port(
|
|
I2CAddress.PCA9555_HEAD_WRITE,
|
|
PCA9555Register.INPUT_PORT_0
|
|
)
|
|
|
|
# ------------------------------------------------------------------------
|
|
# Safety Functions
|
|
# ------------------------------------------------------------------------
|
|
|
|
def emergency_stop(self) -> bool:
|
|
"""
|
|
Emergency stop: close shutter, set current to 0, disable keyswitch
|
|
|
|
Returns:
|
|
True if all operations successful, False otherwise
|
|
"""
|
|
success = True
|
|
success &= self.set_shutter(False)
|
|
success &= self.set_current(0)
|
|
success &= self.set_keyswitch(False)
|
|
self._safe_state_active = True
|
|
return success
|
|
|
|
def enter_safe_state(self) -> bool:
|
|
"""
|
|
Enter safe state (similar to emergency stop but also disables remote)
|
|
|
|
Returns:
|
|
True if successful, False otherwise
|
|
"""
|
|
success = True
|
|
success &= self.set_shutter(False)
|
|
success &= self.set_current(0)
|
|
success &= self.set_power_cmd(0)
|
|
success &= self.set_keyswitch(False)
|
|
success &= self.set_remote_enable(False)
|
|
self._safe_state_active = True
|
|
return success
|