【python】导入serial模块，读写串口数据（嵌入式软件自动化测试）

'# 【python】导入serial模块，读写串口数据（嵌入式软件自动化测试）

一、背景与问题

在嵌入式软件开发和自动化测试领域，串口通信是设备间数据传输的常见方式。Python的serial模块为开发者提供了对串口通信的高级抽象，但其底层原理和使用细节往往被忽视。本文将深入剖析serial模块的工作原理，结合真实开发场景，探讨其在自动化测试中的应用边界和实践技巧。

二、基本原理

串口通信（Serial Communication）是通过异步串行通信协议（RS-232/RS-485等）进行数据传输的技术。核心参数包括：

波特率（Baud Rate）：单位时间内传输的比特数（如9600 bps）
数据位（Data Bits）：通常为8位
停止位（Stop Bits）：1或2位
校验位（Parity）：偶校验/奇校验/无校验

serial模块通过封装底层系统调用（Linux的termios、Windows的CreateFile等），将复杂的通信参数抽象为简单的API。其核心工作原理包括：

设备打开：获取串口文件描述符
参数配置：设置波特率、校验位等
数据传输：通过write/read进行数据交换
异常处理：处理通信中断、超时等异常

三、环境准备

安装依赖

pip install pyserial

系统检测

Linux系统需确认串口设备节点：

dmesg | grep tty

Windows系统需通过设备管理器查看COM端口。

软件配置

确保开发环境支持串口通信，如在Linux中需赋予设备读写权限：

sudo chmod 666 /dev/ttyUSB0

四、核心实现

1. 基础读写示例

import serial

def basic_serial_operations():
    # 创建串口对象
    ser = serial.Serial(
        port='/dev/ttyUSB0',   # 设备路径
        baudrate=9600,         # 波特率
        parity=serial.PARITY_NONE,  # 无校验
        stopbits=serial.STOPBITS_ONE,  # 1位停止位
        bytesize=serial.EIGHTBITS   # 8位数据位
    )
    
    try:
        # 发送数据
        ser.write(b'Hello, Serial!\n')
        
        # 接收数据
        response = ser.read(100)
        print(f"Received: {response.decode('utf-8')}")
    finally:
        # 关闭串口
        ser.close()

关键代码解析：

serial.Serial参数含义需与设备实际参数匹配
write方法接受bytes类型数据
read方法需要指定读取长度，避免阻塞

2. 带超时的读取

def timeout_read_example():
    ser = serial.Serial(
        port='/dev/ttyUSB0',
        baudrate=115200,
        timeout=1  # 设置读取超时时间（秒）
    )
    
    try:
        ser.write(b'Query\n')
        data = ser.read(1024)
        if not data:
            print("No response received")
        else:
            print(f"Response: {data.decode()}")
    finally:
        ser.close()

关键点：

timeout参数控制读取阻塞时间
空数据返回时需特殊处理
避免在主线程中长时间阻塞

3. 多线程处理

import threading

def read_thread(ser):
    while True:
        data = ser.read(1024)
        if data:
            print(f"Thread received: {data.decode()}")
        else:
            break

def multi_thread_example():
    ser = serial.Serial('/dev/ttyUSB0', 115200)
    
    # 启动读线程
    reader_thread = threading.Thread(target=read_thread, args=(ser,))
    reader_thread.start()
    
    # 主线程发送数据
    ser.write(b'Test\n')
    ser.close()

注意事项：

多线程需注意资源竞争
建议使用with语句管理资源
线程间通信需使用队列或锁

五、完整案例：温度传感器测试

场景描述

模拟测试一个温度传感器设备，通过串口发送读取指令，验证返回数据格式。

实现代码

import serial
import time
import unittest

class TemperatureSensorTest(unittest.TestCase):
    def setUp(self):
        self.ser = serial.Serial(
            port='/dev/ttyUSB0',
            baudrate=9600,
            timeout=1
        )
    
    def test_temperature_read(self):
        # 发送读取指令
        self.ser.write(b'READ_TEMP\n')
        
        # 等待响应
        time.sleep(0.5)
        
        # 读取响应
        response = self.ser.read(100)
        self.assertIn(b'OK', response)
        self.assertTrue(len(response) > 10)
        
        # 解析温度数据
        temp_str = response.decode().split(':')[1].strip()
        temp = float(temp_str)
        self.assertTrue(-50 <= temp <= 120)
    
    def tearDown(self):
        self.ser.close()

if __name__ == '__main__':
    unittest.main()

关键点：

测试用例包含完整生命周期管理
验证响应格式和数据有效性
模拟真实设备的响应时间

六、源码解析

pyserial的源码核心逻辑位于serial/serial.py，关键部分包括：

class Serial:
    def __init__(self, port, baudrate, ...):
        self.fd = open(port, 'r+b')  # 打开设备文件
        self._set_port_params()  # 设置通信参数
        
    def _set_port_params(self):
        # 设置波特率
        self._set_baudrate()
        
        # 设置校验位
        self._set_parity()
        
        # 设置停止位
        self._set_stopbits()
        
    def write(self, data):
        # 发送数据
        os.write(self.fd, data)
        
    def read(self, size):
        # 读取数据
        return os.read(self.fd, size)

关键机制：

使用os.open获取设备文件描述符
通过底层系统调用设置通信参数
使用try...finally确保资源释放

七、进阶使用

1. 协议解析

处理复杂协议时需添加校验机制：

def parse_data(data):
    # 假设数据格式为：[CMD][LEN][DATA][CRC]
    if len(data) < 5:
        return None
        
    cmd = data[0]
    length = int(data[1:3])
    payload = data[3:3+length]
    crc = data[3+length:]
    
    # 验证CRC
    if compute_crc(payload) != crc:
        return None
        
    return {
        'cmd': cmd,
        'payload': payload
    }

2. 基于事件的通信

使用select模块实现非阻塞通信：

import select

def nonblocking_read(ser):
    r, w, e = select.select([ser.fd], [], [], 0.1)
    if r:
        return ser.read(1024)
    return None

3. 数据缓存机制

处理突发数据时可添加缓冲区：

class Buffer:
    def __init__(self, size=1024):
        self.buf = bytearray()
        self.size = size
        
    def write(self, data):
        self.buf.extend(data)
        if len(self.buf) > self.size:
            self.buf = self.buf[-self.size:]
            
    def read(self, size):
        data = self.buf[:size]
        self.buf = self.buf[size:]
        return data

八、性能与工程实践

1. 性能优化策略

优化措施	说明
缓冲读取	使用`read(1024)`代替`read(1)`
避免频繁open	使用`with`管理资源
非阻塞模式	设置`timeout=0`
数据压缩	对连续数据进行压缩处理

2. 安全风险分析

数据完整性：未校验的通信可能导致数据篡改
拒绝服务：恶意数据可导致缓冲区溢出
权限控制：需限制对串口设备的访问权限

3. 异常处理

try:
    ser = serial.Serial('/dev/ttyUSB0', 9600)
    ser.write(b'CMD\n')
except serial.SerialException as e:
    print(f"Serial error: {e}")
finally:
    ser.close()

九、常见问题与踩坑

1. 常见错误

错误现象	原因	解决方案
无法打开设备	端口权限不足	使用`sudo`或修改权限
通信失败	波特率不匹配	与设备参数完全一致
数据丢失	未设置超时	合理配置`timeout`
线程冲突	资源未释放	使用`with`语句管理资源

2. 特殊场景处理

Windows系统：

import serial.tools.list_ports

ports = serial.tools.list_ports.comports()
for p in ports:
    print(p.device)

Linux系统：

import serial

ser = serial.Serial('/dev/ttyUSB0', 115200)
print(ser.name)

十、最佳实践

资源管理：使用with语句确保资源释放
参数校验：确保所有通信参数与设备一致
日志记录：添加详细日志便于调试
异常处理：覆盖所有可能的异常情况
版本兼容：注意不同系统下参数差异
测试覆盖：编写完整的测试用例
性能监控：监控通信吞吐量和延迟

十一、总结

serial模块为嵌入式开发和自动化测试提供了强大的串口通信支持，但其应用需要深入理解底层原理和实际场景。在开发过程中，需要重点关注参数配置、异常处理和资源管理，同时结合具体业务需求选择合适的通信策略。对于高并发或实时性要求的场景，可考虑使用更高级的通信框架（如pyserial-asyncio）。通过合理的实践和规范的开发流程，可以充分发挥串口通信的优势，提升嵌入式系统的可靠性和可维护性。