# LaiYu液体处理设备硬件连接配置指南
## 📋 文档概述
本指南提供LaiYu液体处理设备的完整硬件连接配置方案,包括快速入门、详细配置、连接验证和故障排除。适用于设备初次安装、配置变更和问题诊断。
---
## 🚀 快速入门指南
### 基本配置步骤
1. **确认硬件连接**
- 将RS485转USB设备连接到计算机
- 确保XYZ控制器和移液器通过RS485总线连接
- 检查设备供电状态
2. **获取串口信息**
```bash
# macOS/Linux
ls /dev/cu.* | grep usbserial
# 常见输出: /dev/cu.usbserial-3130
```
3. **基本配置参数**
```python
# 推荐的默认配置
config = LaiYuLiquidConfig(
port="/dev/cu.usbserial-3130", # 🔧 替换为实际串口号
address=4, # 移液器地址(固定)
baudrate=115200, # 推荐波特率
timeout=5.0 # 通信超时
)
```
4. **快速连接测试**
```python
device = LaiYuLiquid(config)
success = await device.setup()
print(f"连接状态: {'成功' if success else '失败'}")
```
---
## 🏗️ 硬件架构详解
### 系统组成
LaiYu液体处理设备采用RS485总线架构,包含以下核心组件:
| 组件 | 通信协议 | 设备地址 | 默认波特率 | 功能描述 |
|------|----------|----------|------------|----------|
| **XYZ三轴控制器** | RS485 (Modbus) | X轴=1, Y轴=2, Z轴=3 | 115200 | 三维运动控制 |
| **SOPA移液器** | RS485 | 4 (推荐) | 115200 | 液体吸取分配 |
| **RS485转USB** | USB/串口 | - | 115200 | 通信接口转换 |
### 地址分配策略
```
RS485总线地址分配:
├── 地址 1: X轴步进电机 (自动分配)
├── 地址 2: Y轴步进电机 (自动分配)
├── 地址 3: Z轴步进电机 (自动分配)
├── 地址 4: SOPA移液器 (推荐配置)
└── 禁用地址: 47('/'), 69('E'), 91('[')
```
### 通信参数规范
| 参数 | XYZ控制器 | SOPA移液器 | 说明 |
|------|-----------|------------|------|
| **数据位** | 8 | 8 | 固定值 |
| **停止位** | 1 | 1 | 固定值 |
| **校验位** | 无 | 无 | 固定值 |
| **流控制** | 无 | 无 | 固定值 |
---
## ⚙️ 配置参数详解
### 1. 核心配置类
#### LaiYuLiquidConfig 参数说明
```python
@dataclass
class LaiYuLiquidConfig:
# === 通信参数 ===
port: str = "/dev/cu.usbserial-3130" # 串口设备路径
address: int = 4 # 移液器地址(推荐值)
baudrate: int = 115200 # 通信波特率(推荐值)
timeout: float = 5.0 # 通信超时时间(秒)
# === 工作台物理尺寸 ===
deck_width: float = 340.0 # 工作台宽度 (mm)
deck_height: float = 250.0 # 工作台高度 (mm)
deck_depth: float = 160.0 # 工作台深度 (mm)
# === 运动控制参数 ===
max_speed: float = 100.0 # 最大移动速度 (mm/s)
acceleration: float = 50.0 # 加速度 (mm/s²)
safe_height: float = 50.0 # 安全移动高度 (mm)
# === 移液参数 ===
max_volume: float = 1000.0 # 最大移液体积 (μL)
min_volume: float = 0.1 # 最小移液体积 (μL)
liquid_detection: bool = True # 启用液面检测
# === 枪头操作参数 ===
tip_pickup_speed: int = 30 # 取枪头速度 (rpm)
tip_pickup_acceleration: int = 500 # 取枪头加速度 (rpm/s)
tip_pickup_depth: float = 10.0 # 枪头插入深度 (mm)
tip_drop_height: float = 10.0 # 丢弃枪头高度 (mm)
```
### 2. 配置文件位置
#### A. 代码配置(推荐)
```python
# 在Python代码中直接配置
from unilabos.devices.laiyu_liquid import LaiYuLiquidConfig
config = LaiYuLiquidConfig(
port="/dev/cu.usbserial-3130", # 🔧 修改为实际串口
address=4, # 🔧 移液器地址
baudrate=115200, # 🔧 通信波特率
timeout=5.0 # 🔧 超时时间
)
```
#### B. JSON配置文件
```json
{
"laiyu_liquid_config": {
"port": "/dev/cu.usbserial-3130",
"address": 4,
"baudrate": 115200,
"timeout": 5.0,
"deck_width": 340.0,
"deck_height": 250.0,
"deck_depth": 160.0,
"max_speed": 100.0,
"acceleration": 50.0,
"safe_height": 50.0
}
}
```
#### C. 实验协议配置
```json
// test/experiments/laiyu_liquid.json
{
"device_config": {
"type": "laiyu_liquid",
"config": {
"port": "/dev/cu.usbserial-3130",
"address": 4,
"baudrate": 115200
}
}
}
```
### 2. 串口设备识别
#### 自动识别方法(推荐)
```python
import serial.tools.list_ports
def find_laiyu_device():
"""自动查找LaiYu设备串口"""
ports = serial.tools.list_ports.comports()
for port in ports:
# 根据设备描述或VID/PID识别
if 'usbserial' in port.device.lower():
print(f"找到可能的设备: {port.device}")
print(f"描述: {port.description}")
print(f"硬件ID: {port.hwid}")
return port.device
return None
# 使用示例
device_port = find_laiyu_device()
if device_port:
print(f"检测到设备端口: {device_port}")
else:
print("未检测到设备")
```
#### 手动识别方法
| 操作系统 | 命令 | 设备路径格式 |
|---------|------|-------------|
| **macOS** | `ls /dev/cu.*` | `/dev/cu.usbserial-XXXX` |
| **Linux** | `ls /dev/ttyUSB*` | `/dev/ttyUSB0` |
| **Windows** | 设备管理器 | `COM3`, `COM4` 等 |
#### macOS 详细识别
```bash
# 1. 列出所有USB串口设备
ls /dev/cu.usbserial-*
# 2. 查看USB设备详细信息
system_profiler SPUSBDataType | grep -A 10 "Serial"
# 3. 实时监控设备插拔
ls /dev/cu.* && echo "--- 请插入设备 ---" && sleep 3 && ls /dev/cu.*
```
#### Linux 详细识别
```bash
# 1. 列出串口设备
ls /dev/ttyUSB* /dev/ttyACM*
# 2. 查看设备信息
dmesg | grep -i "usb.*serial"
lsusb | grep -i "serial\|converter"
# 3. 查看设备属性
udevadm info --name=/dev/ttyUSB0 --attribute-walk
```
#### Windows 详细识别
```powershell
# PowerShell命令
Get-WmiObject -Class Win32_SerialPort | Select-Object Name, DeviceID, Description
# 或在设备管理器中查看"端口(COM和LPT)"
```
### 3. 控制器特定配置
#### XYZ步进电机控制器
- **地址范围**: 1-3 (X轴=1, Y轴=2, Z轴=3)
- **通信协议**: Modbus RTU
- **波特率**: 9600 或 115200
- **数据位**: 8
- **停止位**: 1
- **校验位**: None
#### XYZ控制器配置 (`controllers/xyz_controller.py`)
XYZ控制器负责三轴运动控制,提供精确的位置控制和运动规划功能。
**主要功能:**
- 三轴独立控制(X、Y、Z轴)
- 位置精度控制
- 运动速度调节
- 安全限位检测
**配置参数:**
```python
xyz_config = {
"port": "/dev/ttyUSB0", # 串口设备
"baudrate": 115200, # 波特率
"timeout": 1.0, # 通信超时
"max_speed": { # 最大速度限制
"x": 1000, # X轴最大速度
"y": 1000, # Y轴最大速度
"z": 500 # Z轴最大速度
},
"acceleration": 500, # 加速度
"home_position": [0, 0, 0] # 原点位置
}
```
```python
def __init__(self, port: str, baudrate: int = 115200,
machine_config: Optional[MachineConfig] = None,
config_file: str = "machine_config.json",
auto_connect: bool = True):
"""
Args:
port: 串口端口 (如: "/dev/cu.usbserial-3130")
baudrate: 波特率 (默认: 115200)
machine_config: 机械配置参数
config_file: 配置文件路径
auto_connect: 是否自动连接
"""
```
#### SOPA移液器
- **地址**: 通常为 4 或更高
- **通信协议**: 自定义协议
- **波特率**: 115200 (推荐)
- **响应时间**: < 100ms
#### 移液器控制器配置 (`controllers/pipette_controller.py`)
移液器控制器负责精确的液体吸取和分配操作,支持多种移液模式和参数配置。
**主要功能:**
- 精确体积控制
- 液面检测
- 枪头管理
- 速度调节
**配置参数:**
```python
@dataclass
class SOPAConfig:
# 通信参数
port: str = "/dev/ttyUSB0" # 🔧 修改串口号
baudrate: int = 115200 # 🔧 修改波特率
address: int = 1 # 🔧 修改设备地址 (1-254)
timeout: float = 5.0 # 🔧 修改超时时间
comm_type: CommunicationType = CommunicationType.TERMINAL_DEBUG
```
## 🔍 连接验证与测试
### 1. 编程方式验证连接
#### 创建测试脚本
```python
#!/usr/bin/env python3
"""
LaiYu液体处理设备连接测试脚本
"""
import sys
import os
sys.path.append('/Users/dp/Documents/DPT/HuaiRou/Uni-Lab-OS')
from unilabos.devices.laiyu_liquid.core.LaiYu_Liquid import (
LaiYuLiquid, LaiYuLiquidConfig
)
def test_connection():
"""测试设备连接"""
# 🔧 修改这里的配置参数
config = LaiYuLiquidConfig(
port="/dev/cu.usbserial-3130", # 修改为你的串口号
address=1, # 修改为你的设备地址
baudrate=9600, # 修改为你的波特率
timeout=5.0
)
print("🔌 正在测试LaiYu液体处理设备连接...")
print(f"串口: {config.port}")
print(f"波特率: {config.baudrate}")
print(f"设备地址: {config.address}")
print("-" * 50)
try:
# 创建设备实例
device = LaiYuLiquid(config)
# 尝试连接和初始化
print("📡 正在连接设备...")
success = await device.setup()
if success:
print("✅ 设备连接成功!")
print(f"连接状态: {device.is_connected}")
print(f"初始化状态: {device.is_initialized}")
print(f"当前位置: {device.current_position}")
# 获取设备状态
status = device.get_status()
print("\n📊 设备状态:")
for key, value in status.items():
print(f" {key}: {value}")
else:
print("❌ 设备连接失败!")
print("请检查:")
print(" 1. 串口号是否正确")
print(" 2. 设备是否已连接并通电")
print(" 3. 波特率和设备地址是否匹配")
print(" 4. 串口是否被其他程序占用")
except Exception as e:
print(f"❌ 连接测试出错: {e}")
print("\n🔧 故障排除建议:")
print(" 1. 检查串口设备是否存在:")
print(" macOS: ls /dev/cu.*")
print(" Linux: ls /dev/ttyUSB* /dev/ttyACM*")
print(" 2. 检查设备权限:")
print(" sudo chmod 666 /dev/cu.usbserial-*")
print(" 3. 检查设备是否被占用:")
print(" lsof | grep /dev/cu.usbserial")
finally:
# 清理连接
if 'device' in locals():
await device.stop()
if __name__ == "__main__":
import asyncio
asyncio.run(test_connection())
```
### 2. 命令行验证工具
#### 串口通信测试
```bash
# 安装串口调试工具
pip install pyserial
# 使用Python测试串口
python -c "
import serial
try:
ser = serial.Serial('/dev/cu.usbserial-3130', 9600, timeout=1)
print('串口连接成功:', ser.is_open)
ser.close()
except Exception as e:
print('串口连接失败:', e)
"
```
#### 设备权限检查
```bash
# macOS/Linux 检查串口权限
ls -la /dev/cu.usbserial-*
# 如果权限不足,修改权限
sudo chmod 666 /dev/cu.usbserial-*
# 检查串口是否被占用
lsof | grep /dev/cu.usbserial
```
### 3. 连接状态指示器
设备提供多种方式检查连接状态:
#### A. 属性检查
```python
device = LaiYuLiquid(config)
# 检查连接状态
print(f"设备已连接: {device.is_connected}")
print(f"设备已初始化: {device.is_initialized}")
print(f"枪头已安装: {device.tip_attached}")
print(f"当前位置: {device.current_position}")
print(f"当前体积: {device.current_volume}")
```
#### B. 状态字典
```python
status = device.get_status()
print("完整设备状态:", status)
# 输出示例:
# {
# 'connected': True,
# 'initialized': True,
# 'position': (0.0, 0.0, 50.0),
# 'tip_attached': False,
# 'current_volume': 0.0,
# 'last_error': None
# }
```
## 🛠️ 故障排除指南
### 1. 连接问题诊断
#### 🔍 问题诊断流程
```python
def diagnose_connection_issues():
"""连接问题诊断工具"""
import serial.tools.list_ports
import serial
print("🔍 开始连接问题诊断...")
# 1. 检查串口设备
ports = list(serial.tools.list_ports.comports())
if not ports:
print("❌ 未检测到任何串口设备")
print("💡 解决方案:")
print(" - 检查USB连接线")
print(" - 确认设备电源")
print(" - 安装设备驱动")
return
print(f"✅ 检测到 {len(ports)} 个串口设备")
for port in ports:
print(f" 📍 {port.device}: {port.description}")
# 2. 测试串口访问权限
for port in ports:
try:
with serial.Serial(port.device, 9600, timeout=1):
print(f"✅ {port.device}: 访问权限正常")
except PermissionError:
print(f"❌ {port.device}: 权限不足")
print("💡 解决方案: sudo chmod 666 " + port.device)
except Exception as e:
print(f"⚠️ {port.device}: {e}")
# 运行诊断
diagnose_connection_issues()
```
#### 🚫 常见连接错误
| 错误类型 | 症状 | 解决方案 |
|---------|------|----------|
| **设备未找到** | `FileNotFoundError: No such file or directory` | 1. 检查USB连接
2. 确认设备驱动
3. 重新插拔设备 |
| **权限不足** | `PermissionError: Permission denied` | 1. `sudo chmod 666 /dev/ttyUSB0`
2. 添加用户到dialout组
3. 使用sudo运行 |
| **设备占用** | `SerialException: Device or resource busy` | 1. 关闭其他程序
2. `lsof /dev/ttyUSB0`查找占用
3. 重启系统 |
| **驱动问题** | 设备管理器显示未知设备 | 1. 安装CH340/CP210x驱动
2. 更新系统驱动
3. 使用原装USB线 |
### 2. 通信问题解决
#### 📡 通信参数调试
```python
def test_communication_parameters():
"""测试不同通信参数"""
import serial
port = "/dev/cu.usbserial-3130" # 修改为实际端口
baudrates = [9600, 19200, 38400, 57600, 115200]
for baudrate in baudrates:
print(f"🔄 测试波特率: {baudrate}")
try:
with serial.Serial(port, baudrate, timeout=2) as ser:
# 发送测试命令
test_cmd = b'\x01\x03\x00\x00\x00\x01\x84\x0A'
ser.write(test_cmd)
response = ser.read(100)
if response:
print(f" ✅ 成功: 收到 {len(response)} 字节")
print(f" 📦 数据: {response.hex()}")
return baudrate
else:
print(f" ❌ 无响应")
except Exception as e:
print(f" ❌ 错误: {e}")
return None
```
#### ⚡ 通信故障排除
| 问题类型 | 症状 | 诊断方法 | 解决方案 |
|---------|------|----------|----------|
| **通信超时** | `TimeoutError` | 检查波特率和设备地址 | 1. 调整超时时间
2. 验证波特率
3. 检查设备地址 |
| **数据校验错误** | `CRCError` | 检查数据完整性 | 1. 更换USB线
2. 降低波特率
3. 检查电磁干扰 |
| **协议错误** | 响应格式异常 | 验证命令格式 | 1. 检查协议版本
2. 确认设备类型
3. 更新固件 |
| **间歇性故障** | 时好时坏 | 监控连接稳定性 | 1. 检查连接线
2. 稳定电源
3. 减少干扰源 |
### 3. 设备功能问题
#### 🎯 设备状态检查
```python
def check_device_health():
"""设备健康状态检查"""
from unilabos.devices.laiyu_liquid import LaiYuLiquidConfig, LaiYuLiquidBackend
config = LaiYuLiquidConfig(
port="/dev/cu.usbserial-3130",
address=4,
baudrate=115200,
timeout=5.0
)
try:
backend = LaiYuLiquidBackend(config)
backend.connect()
# 检查项目
checks = {
"设备连接": lambda: backend.is_connected(),
"XYZ轴状态": lambda: backend.xyz_controller.get_all_positions(),
"移液器状态": lambda: backend.pipette_controller.get_status(),
"设备温度": lambda: backend.get_temperature(),
"错误状态": lambda: backend.get_error_status(),
}
print("🏥 设备健康检查报告")
print("=" * 40)
for check_name, check_func in checks.items():
try:
result = check_func()
print(f"✅ {check_name}: 正常")
if result:
print(f" 📊 数据: {result}")
except Exception as e:
print(f"❌ {check_name}: 异常 - {e}")
backend.disconnect()
except Exception as e:
print(f"❌ 无法连接设备: {e}")
```
### 4. 高级故障排除
#### 🔧 日志分析工具
```python
import logging
def setup_debug_logging():
"""设置调试日志"""
logging.basicConfig(
level=logging.DEBUG,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('laiyu_debug.log'),
logging.StreamHandler()
]
)
# 启用串口通信日志
serial_logger = logging.getLogger('serial')
serial_logger.setLevel(logging.DEBUG)
print("🔍 调试日志已启用,日志文件: laiyu_debug.log")
```
#### 📊 性能监控
```python
def monitor_performance():
"""性能监控工具"""
import time
import psutil
print("📊 开始性能监控...")
start_time = time.time()
start_cpu = psutil.cpu_percent()
start_memory = psutil.virtual_memory().percent
# 执行设备操作
# ... 你的设备操作代码 ...
end_time = time.time()
end_cpu = psutil.cpu_percent()
end_memory = psutil.virtual_memory().percent
print(f"⏱️ 执行时间: {end_time - start_time:.2f} 秒")
print(f"💻 CPU使用: {end_cpu - start_cpu:.1f}%")
print(f"🧠 内存使用: {end_memory - start_memory:.1f}%")
```
## 📝 配置文件模板
### 1. 基础配置模板
#### 标准配置(推荐)
```python
from unilabos.devices.laiyu_liquid import LaiYuLiquidConfig, LaiYuLiquidBackend, LaiYuLiquid
# 创建标准配置
config = LaiYuLiquidConfig(
# === 通信参数 ===
port="/dev/cu.usbserial-3130", # 🔧 修改为实际串口
address=4, # 移液器地址(推荐)
baudrate=115200, # 通信波特率(推荐)
timeout=5.0, # 通信超时时间
# === 工作台尺寸 ===
deck_width=340.0, # 工作台宽度 (mm)
deck_height=250.0, # 工作台高度 (mm)
deck_depth=160.0, # 工作台深度 (mm)
# === 运动控制参数 ===
max_speed=100.0, # 最大移动速度 (mm/s)
acceleration=50.0, # 加速度 (mm/s²)
safe_height=50.0, # 安全移动高度 (mm)
# === 移液参数 ===
max_volume=1000.0, # 最大移液体积 (μL)
min_volume=0.1, # 最小移液体积 (μL)
liquid_detection=True, # 启用液面检测
# === 枪头操作参数 ===
tip_pickup_speed=30, # 取枪头速度 (rpm)
tip_pickup_acceleration=500, # 取枪头加速度 (rpm/s)
tip_pickup_depth=10.0, # 枪头插入深度 (mm)
tip_drop_height=10.0, # 丢弃枪头高度 (mm)
)
# 创建设备实例
backend = LaiYuLiquidBackend(config)
device = LaiYuLiquid(backend)
```
### 2. 高级配置模板
#### 多设备配置
```python
# 配置多个LaiYu设备
configs = {
"device_1": LaiYuLiquidConfig(
port="/dev/cu.usbserial-3130",
address=4,
baudrate=115200,
deck_width=340.0,
deck_height=250.0,
deck_depth=160.0
),
"device_2": LaiYuLiquidConfig(
port="/dev/cu.usbserial-3131",
address=4,
baudrate=115200,
deck_width=340.0,
deck_height=250.0,
deck_depth=160.0
)
}
# 创建设备实例
devices = {}
for name, config in configs.items():
backend = LaiYuLiquidBackend(config)
devices[name] = LaiYuLiquid(backend)
```
#### 自定义参数配置
```python
# 高精度移液配置
precision_config = LaiYuLiquidConfig(
port="/dev/cu.usbserial-3130",
address=4,
baudrate=115200,
timeout=10.0, # 增加超时时间
# 精密运动控制
max_speed=50.0, # 降低速度提高精度
acceleration=25.0, # 降低加速度
safe_height=30.0, # 降低安全高度
# 精密移液参数
max_volume=200.0, # 小体积移液
min_volume=0.5, # 提高最小体积
liquid_detection=True,
# 精密枪头操作
tip_pickup_speed=15, # 降低取枪头速度
tip_pickup_acceleration=250, # 降低加速度
tip_pickup_depth=8.0, # 减少插入深度
tip_drop_height=5.0, # 降低丢弃高度
)
```
### 3. 实验协议配置
#### JSON配置文件模板
```json
{
"experiment_name": "LaiYu液体处理实验",
"version": "1.0",
"devices": {
"laiyu_liquid": {
"type": "LaiYu_Liquid",
"config": {
"port": "/dev/cu.usbserial-3130",
"address": 4,
"baudrate": 115200,
"timeout": 5.0,
"deck_width": 340.0,
"deck_height": 250.0,
"deck_depth": 160.0,
"max_speed": 100.0,
"acceleration": 50.0,
"safe_height": 50.0,
"max_volume": 1000.0,
"min_volume": 0.1,
"liquid_detection": true
}
}
},
"deck_layout": {
"tip_rack": {
"type": "tip_rack_96",
"position": [10, 10, 0],
"tips": "1000μL"
},
"source_plate": {
"type": "plate_96",
"position": [100, 10, 0],
"contents": "样品"
},
"dest_plate": {
"type": "plate_96",
"position": [200, 10, 0],
"contents": "目标"
}
}
}
```
### 4. 完整配置示例
```json
{
"laiyu_liquid_config": {
"communication": {
"xyz_controller": {
"port": "/dev/cu.usbserial-3130",
"baudrate": 115200,
"timeout": 5.0
},
"pipette_controller": {
"port": "/dev/cu.usbserial-3131",
"baudrate": 115200,
"address": 4,
"timeout": 5.0
}
},
"mechanical": {
"deck_width": 340.0,
"deck_height": 250.0,
"deck_depth": 160.0,
"safe_height": 50.0
},
"motion": {
"max_speed": 100.0,
"acceleration": 50.0,
"tip_pickup_speed": 30,
"tip_pickup_acceleration": 500
},
"safety": {
"position_validation": true,
"emergency_stop_enabled": true,
"deck_width": 300.0,
"deck_height": 200.0,
"deck_depth": 100.0,
"safe_height": 50.0
}
}
}
```
### 5. 完整使用示例
#### 基础移液操作
```python
async def basic_pipetting_example():
"""基础移液操作示例"""
# 1. 设备初始化
config = LaiYuLiquidConfig(
port="/dev/cu.usbserial-3130",
address=4,
baudrate=115200
)
backend = LaiYuLiquidBackend(config)
device = LaiYuLiquid(backend)
try:
# 2. 设备设置
await device.setup()
print("✅ 设备初始化完成")
# 3. 回到原点
await device.home_all_axes()
print("✅ 轴归零完成")
# 4. 取枪头
tip_position = (50, 50, 10) # 枪头架位置
await device.pick_up_tip(tip_position)
print("✅ 取枪头完成")
# 5. 移液操作
source_pos = (100, 100, 15) # 源位置
dest_pos = (200, 200, 15) # 目标位置
volume = 100.0 # 移液体积 (μL)
await device.aspirate(volume, source_pos)
print(f"✅ 吸取 {volume}μL 完成")
await device.dispense(volume, dest_pos)
print(f"✅ 分配 {volume}μL 完成")
# 6. 丢弃枪头
trash_position = (300, 300, 20)
await device.drop_tip(trash_position)
print("✅ 丢弃枪头完成")
except Exception as e:
print(f"❌ 操作失败: {e}")
finally:
# 7. 清理资源
await device.cleanup()
print("✅ 设备清理完成")
# 运行示例
import asyncio
asyncio.run(basic_pipetting_example())
```
#### 批量处理示例
```python
async def batch_processing_example():
"""批量处理示例"""
config = LaiYuLiquidConfig(
port="/dev/cu.usbserial-3130",
address=4,
baudrate=115200
)
backend = LaiYuLiquidBackend(config)
device = LaiYuLiquid(backend)
try:
await device.setup()
await device.home_all_axes()
# 定义位置
tip_rack = [(50 + i*9, 50, 10) for i in range(12)] # 12个枪头位置
source_wells = [(100 + i*9, 100, 15) for i in range(12)] # 12个源孔
dest_wells = [(200 + i*9, 200, 15) for i in range(12)] # 12个目标孔
# 批量移液
for i in range(12):
print(f"🔄 处理第 {i+1} 个样品...")
# 取枪头
await device.pick_up_tip(tip_rack[i])
# 移液
await device.aspirate(50.0, source_wells[i])
await device.dispense(50.0, dest_wells[i])
# 丢弃枪头
await device.drop_tip((300, 300, 20))
print(f"✅ 第 {i+1} 个样品处理完成")
print("🎉 批量处理完成!")
except Exception as e:
print(f"❌ 批量处理失败: {e}")
finally:
await device.cleanup()
# 运行批量处理
asyncio.run(batch_processing_example())
```
## 🔧 调试与日志管理
### 1. 调试模式配置
#### 启用全局调试
```python
import logging
from unilabos.devices.laiyu_liquid import LaiYuLiquidConfig, LaiYuLiquidBackend
# 配置全局日志
logging.basicConfig(
level=logging.DEBUG,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('laiyu_debug.log'),
logging.StreamHandler()
]
)
# 创建调试配置
debug_config = LaiYuLiquidConfig(
port="/dev/cu.usbserial-3130",
address=4,
baudrate=115200,
timeout=10.0, # 增加超时时间便于调试
debug_mode=True # 启用调试模式
)
```
#### 分级日志配置
```python
def setup_logging(log_level="INFO"):
"""设置分级日志"""
# 日志级别映射
levels = {
"DEBUG": logging.DEBUG,
"INFO": logging.INFO,
"WARNING": logging.WARNING,
"ERROR": logging.ERROR
}
# 创建日志记录器
logger = logging.getLogger('LaiYu_Liquid')
logger.setLevel(levels.get(log_level, logging.INFO))
# 文件处理器
file_handler = logging.FileHandler('laiyu_operations.log')
file_formatter = logging.Formatter(
'%(asctime)s - %(name)s - %(levelname)s - %(funcName)s:%(lineno)d - %(message)s'
)
file_handler.setFormatter(file_formatter)
# 控制台处理器
console_handler = logging.StreamHandler()
console_formatter = logging.Formatter('%(levelname)s - %(message)s')
console_handler.setFormatter(console_formatter)
logger.addHandler(file_handler)
logger.addHandler(console_handler)
return logger
# 使用示例
logger = setup_logging("DEBUG")
logger.info("开始LaiYu设备操作")
```
### 2. 通信监控
#### 串口通信日志
```python
def enable_serial_logging():
"""启用串口通信日志"""
import serial
# 创建串口日志记录器
serial_logger = logging.getLogger('serial.communication')
serial_logger.setLevel(logging.DEBUG)
# 创建专用的串口日志文件
serial_handler = logging.FileHandler('laiyu_serial.log')
serial_formatter = logging.Formatter(
'%(asctime)s - SERIAL - %(message)s'
)
serial_handler.setFormatter(serial_formatter)
serial_logger.addHandler(serial_handler)
print("📡 串口通信日志已启用: laiyu_serial.log")
return serial_logger
```
#### 实时通信监控
```python
class CommunicationMonitor:
"""通信监控器"""
def __init__(self):
self.sent_count = 0
self.received_count = 0
self.error_count = 0
self.start_time = time.time()
def log_sent(self, data):
"""记录发送数据"""
self.sent_count += 1
logging.debug(f"📤 发送 #{self.sent_count}: {data.hex()}")
def log_received(self, data):
"""记录接收数据"""
self.received_count += 1
logging.debug(f"📥 接收 #{self.received_count}: {data.hex()}")
def log_error(self, error):
"""记录错误"""
self.error_count += 1
logging.error(f"❌ 通信错误 #{self.error_count}: {error}")
def get_statistics(self):
"""获取统计信息"""
duration = time.time() - self.start_time
return {
"运行时间": f"{duration:.2f}秒",
"发送次数": self.sent_count,
"接收次数": self.received_count,
"错误次数": self.error_count,
"成功率": f"{((self.sent_count - self.error_count) / max(self.sent_count, 1) * 100):.1f}%"
}
```
### 3. 性能监控
#### 操作性能分析
```python
import time
import functools
def performance_monitor(operation_name):
"""性能监控装饰器"""
def decorator(func):
@functools.wraps(func)
async def wrapper(*args, **kwargs):
start_time = time.time()
start_memory = psutil.Process().memory_info().rss / 1024 / 1024 # MB
try:
result = await func(*args, **kwargs)
end_time = time.time()
end_memory = psutil.Process().memory_info().rss / 1024 / 1024 # MB
duration = end_time - start_time
memory_delta = end_memory - start_memory
logging.info(f"⏱️ {operation_name}: {duration:.3f}s, 内存变化: {memory_delta:+.1f}MB")
return result
except Exception as e:
end_time = time.time()
duration = end_time - start_time
logging.error(f"❌ {operation_name} 失败 ({duration:.3f}s): {e}")
raise
return wrapper
return decorator
# 使用示例
@performance_monitor("移液操作")
async def monitored_pipetting():
await device.aspirate(100.0, (100, 100, 15))
await device.dispense(100.0, (200, 200, 15))
```
#### 系统资源监控
```python
import psutil
import threading
import time
class SystemMonitor:
"""系统资源监控器"""
def __init__(self, interval=1.0):
self.interval = interval
self.monitoring = False
self.data = []
def start_monitoring(self):
"""开始监控"""
self.monitoring = True
self.monitor_thread = threading.Thread(target=self._monitor_loop)
self.monitor_thread.daemon = True
self.monitor_thread.start()
print("📊 系统监控已启动")
def stop_monitoring(self):
"""停止监控"""
self.monitoring = False
if hasattr(self, 'monitor_thread'):
self.monitor_thread.join()
print("📊 系统监控已停止")
def _monitor_loop(self):
"""监控循环"""
while self.monitoring:
cpu_percent = psutil.cpu_percent()
memory = psutil.virtual_memory()
self.data.append({
'timestamp': time.time(),
'cpu_percent': cpu_percent,
'memory_percent': memory.percent,
'memory_used_mb': memory.used / 1024 / 1024
})
time.sleep(self.interval)
def get_report(self):
"""生成监控报告"""
if not self.data:
return "无监控数据"
avg_cpu = sum(d['cpu_percent'] for d in self.data) / len(self.data)
avg_memory = sum(d['memory_percent'] for d in self.data) / len(self.data)
max_memory = max(d['memory_used_mb'] for d in self.data)
return f"""
📊 系统资源监控报告
==================
监控时长: {len(self.data) * self.interval:.1f}秒
平均CPU使用率: {avg_cpu:.1f}%
平均内存使用率: {avg_memory:.1f}%
峰值内存使用: {max_memory:.1f}MB
"""
# 使用示例
monitor = SystemMonitor()
monitor.start_monitoring()
# 执行设备操作
# ... 你的代码 ...
monitor.stop_monitoring()
print(monitor.get_report())
```
### 4. 错误追踪
#### 异常处理和记录
```python
import traceback
class ErrorTracker:
"""错误追踪器"""
def __init__(self):
self.errors = []
def log_error(self, operation, error, context=None):
"""记录错误"""
error_info = {
'timestamp': time.time(),
'operation': operation,
'error_type': type(error).__name__,
'error_message': str(error),
'traceback': traceback.format_exc(),
'context': context or {}
}
self.errors.append(error_info)
# 记录到日志
logging.error(f"❌ {operation} 失败: {error}")
logging.debug(f"错误详情: {error_info}")
def get_error_summary(self):
"""获取错误摘要"""
if not self.errors:
return "✅ 无错误记录"
error_types = {}
for error in self.errors:
error_type = error['error_type']
error_types[error_type] = error_types.get(error_type, 0) + 1
summary = f"❌ 共记录 {len(self.errors)} 个错误:\n"
for error_type, count in error_types.items():
summary += f" - {error_type}: {count} 次\n"
return summary
# 全局错误追踪器
error_tracker = ErrorTracker()
# 使用示例
try:
await device.move_to(x=1000, y=1000, z=100) # 可能超出范围
except Exception as e:
error_tracker.log_error("移动操作", e, {"target": (1000, 1000, 100)})
```
---
## 📚 总结
本文档提供了LaiYu液体处理设备的完整硬件连接配置指南,涵盖了从基础设置到高级故障排除的所有方面。
### 🎯 关键要点
1. **标准配置**: 使用 `port="/dev/cu.usbserial-3130"`, `address=4`, `baudrate=115200`
2. **设备架构**: XYZ轴控制器(地址1-3) + SOPA移液器(地址4)
3. **连接验证**: 使用提供的测试脚本验证硬件连接
4. **故障排除**: 参考故障排除指南解决常见问题
5. **性能监控**: 启用日志和监控确保稳定运行
### 🔗 相关文档
- [LaiYu控制架构详解](./UniLab_LaiYu_控制架构详解.md)
- [XYZ集成功能说明](./XYZ_集成功能说明.md)
- [设备API文档](./readme.md)
### 📞 技术支持
如遇到问题,请:
1. 检查硬件连接和配置
2. 查看调试日志
3. 参考故障排除指南
4. 联系技术支持团队
---
*最后更新: 2024年1月*