4、pyserial核心API：Serial类、open/close、read/write、in_waiting、flush、超时设置等常用方法

在Python硬件控制领域，pyserial是串口通信的事实标准库。它屏蔽了操作系统底层的差异，为开发者提供了一套统一、简洁的接口。本章将深入剖析pyserial的核心——Serial类，从实例化、连接管理到数据读写与缓冲区控制，逐一拆解其常用API的底层逻辑与实战用法。理解这些API不仅仅是记住函数名，更重要的是理解串口通信的异步特性、缓冲区机制以及超时对程序健壮性的决定性影响。我们将通过一个完整的工业扫码枪数据采集案例，把这些API串联起来，展示一个稳定、高效的串口应用应具备的代码骨架。

串口通信的本质是字节流的异步传输。与I2C或SPI这类由主机提供时钟的同步通信不同，UART（通用异步收发器）在物理层没有时钟线，收发双方必须预先约定波特率、数据位、停止位和校验位。这意味着，接收方无法预知数据何时到达，只能被动地等待并读取缓冲区。pyserial的所有API设计，都是围绕这一异步特性展开的。理解这一点，是掌握read、in_waiting和timeout等核心概念的前提。

4.1 Serial类：串口资源的抽象与实例化

Serial类是pyserial对串口资源的完整抽象。它封装了从打开端口、配置参数到读写数据、管理缓冲区的全部操作。实例化Serial对象时，最常用的参数包括：

• port：串口设备名。在Windows下通常为COM3、COM10等；在Linux下为/dev/ttyUSB0、/dev/ttyS0等；macOS下为/dev/cu.usbserial-xxx。
• baudrate：波特率，即每秒传输的比特数。常见值有9600、115200、460800等。必须与对端设备严格一致，否则会产生乱码。
• bytesize：数据位，通常为8（EIGHTBITS），也有5、6、7。
• parity：校验位，可选PARITY_NONE（无校验）、PARITY_EVEN（偶校验）、PARITY_ODD（奇校验）。
• stopbits：停止位，通常为1（STOPBITS_ONE），也有1.5和2。
• timeout：读操作超时时间（秒）。这是本章的重点之一，稍后详细展开。
• write_timeout：写操作超时时间（秒）。

实例化有两种方式：一种是在构造时直接传入所有参数，端口会立即打开；另一种是先创建对象，再调用open()方法。推荐使用第一种方式，因为它更简洁，且能在第一时间暴露端口不存在或权限不足等问题。

import serial

# 方式一：实例化时直接打开端口
ser = serial.Serial(
    port='/dev/ttyUSB0',
    baudrate=115200,
    bytesize=serial.EIGHTBITS,
    parity=serial.PARITY_NONE,
    stopbits=serial.STOPBITS_ONE,
    timeout=1.0,          # 读超时1秒
    write_timeout=2.0     # 写超时2秒
)
print(f"串口已打开: {ser.name}")

# 方式二：先创建后打开
ser2 = serial.Serial()
ser2.port = 'COM5'
ser2.baudrate = 9600
ser2.timeout = 0.5
ser2.open()
ser2.close()

4.2 连接管理：open、close与is_open

串口是一种独占性资源，一个端口同一时刻只能被一个进程打开。因此，规范的连接管理是避免资源泄漏和端口冲突的关键。

• is_open属性：返回一个布尔值，表示串口当前是否处于打开状态。在调用读写方法前，建议先检查此属性。
• open()方法：打开串口。如果端口不存在、被占用或权限不足，会抛出异常。
• close()方法：关闭串口，释放资源。关闭后，所有未完成的读写操作将被终止。

在实际工程中，强烈建议使用with语句或try/finally块来确保串口被正确关闭。即使程序中途发生异常，也能保证资源释放。

import serial

ser = None
try:
    ser = serial.Serial('COM3', 115200, timeout=1)
    # 执行读写操作...
    data = ser.read(10)
finally:
    if ser and ser.is_open:
        ser.close()
        print("串口已安全关闭")

4.3 数据写入：write与flush

write(data)方法用于向串口发送数据。它接受一个bytes类型对象（在Python 3中，字符串必须编码为bytes）。该方法返回实际写入的字节数。需要注意的是，write()只是将数据拷贝到操作系统的发送缓冲区，并不保证数据已经物理地发送到线路上。

flush()方法用于阻塞等待，直到发送缓冲区中的所有数据都被发送完毕。这在需要确保命令完全发出后再进行下一步操作（如等待设备响应）时非常关键。

import serial
import time

ser = serial.Serial('/dev/ttyUSB0', 9600, timeout=1)

# 发送AT指令（以回车换行结尾）
cmd = b'AT+RESET\r\n'
written = ser.write(cmd)
print(f"已写入 {written} 字节")

# 关键：等待数据真正发送完成
ser.flush()
print("数据已全部发送到线路")

# 发送十六进制数据
hex_data = bytes.fromhex('01 03 00 00 00 01 84 0A')
ser.write(hex_data)
ser.flush()

4.4 数据读取：read、readline与in_waiting

读取是串口编程中最容易出错的部分，因为数据到达的时间是不确定的。pyserial提供了多种读取方式，以适应不同的应用场景。

in_waiting的典型用法：在实时性要求较高的场景中，我们通常先检查in_waiting，如果大于0，则一次性读取所有可用数据，避免多次调用read()带来的系统调用开销。

import serial

ser = serial.Serial('COM8', 115200, timeout=0.1)

# 非阻塞轮询读取
while True:
    if ser.in_waiting > 0:
        # 读取当前缓冲区中的所有数据
        data = ser.read(ser.in_waiting)
        print(f"收到 {len(data)} 字节: {data.hex()}")
    # 做其他事情，避免忙等待
    time.sleep(0.01)

4.5 超时设置：阻塞、非阻塞与超时返回

timeout参数是串口编程的灵魂。它决定了read()系列方法在数据不足时的行为。理解三种模式的差异至关重要：

• timeout=None：阻塞模式。读取操作会一直等待，直到读到足够的字节数（对于read(size)）或遇到行结束符（对于readline()）。如果设备不响应，程序将永久卡死。这种模式只适用于确定会返回数据的场景。
• timeout=0：非阻塞模式。读取操作立即返回，不管有没有数据。如果没有数据，read()返回空bytes。这种模式需要配合in_waiting轮询使用，否则会空转消耗CPU。
• timeout=N（N>0）：超时模式。读取操作最多等待N秒。如果在N秒内没有读到足够数据，则返回当前已读取的部分。这是最常用的模式，既能保证响应性，又不会无限阻塞。

对于write()操作，write_timeout的作用类似。如果发送缓冲区在指定时间内始终没有空间，会抛出SerialTimeoutException。

import serial

# 阻塞模式：等待完整的一行，最多等5秒
ser = serial.Serial('/dev/ttyS1', 9600, timeout=5)
line = ser.readline()
print(f"收到一行: {line}")

# 非阻塞模式：立即返回，配合轮询
ser.timeout = 0
data = ser.read(10)  # 立即返回，可能为空
if data:
    print(f"非阻塞读取: {data}")

# 超时模式：最多等2秒，读取最多64字节
ser.timeout = 2
chunk = ser.read(64)
print(f"超时读取，实际获得 {len(chunk)} 字节")

4.6 实战案例：工业扫码枪数据采集与解析

下面我们通过一个完整的案例，将上述API整合起来。假设我们需要从一台通过RS232接口连接的条码扫描枪读取数据。扫码枪的通信参数为：波特率9600，8数据位，无校验，1停止位。扫码枪在成功扫码后，会主动发送一串ASCII字符，并以回车换行（\r\n）结尾。

需求与约束：

• 需要持续监听扫码枪，实时获取扫码结果。
• 扫码枪可能长时间不工作（没有扫码），程序不能因此崩溃或卡死。
• 如果扫码枪断线或数据异常，需要有明确的错误提示。

代码实现：

import serial
import time
import logging

# 配置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

class BarcodeScanner:
    def __init__(self, port, baudrate=9600, timeout=1.0):
        self.port = port
        self.baudrate = baudrate
        self.timeout = timeout
        self.ser = None
        
    def connect(self):
        """建立串口连接"""
        try:
            self.ser = serial.Serial(
                port=self.port,
                baudrate=self.baudrate,
                bytesize=serial.EIGHTBITS,
                parity=serial.PARITY_NONE,
                stopbits=serial.STOPBITS_ONE,
                timeout=self.timeout
            )
            logging.info(f"扫码枪连接成功: {self.port}")
            return True
        except serial.SerialException as e:
            logging.error(f"连接失败: {e}")
            return False
    
    def read_barcode(self):
        """读取一条完整的条码数据，返回字符串"""
        if not self.ser or not self.ser.is_open:
            logging.warning("串口未打开，尝试重连...")
            if not self.connect():
                return None
        
        try:
            # 使用read_until读取直到遇到换行符，超时由self.timeout控制
            raw_data = self.ser.read_until(expected=b'\n')
            if raw_data.endswith(b'\n'):
                # 去除末尾的\r\n
                barcode = raw_data.decode('ascii', errors='ignore').strip()
                if barcode:
                    logging.info(f"扫码成功: {barcode}")
                    return barcode
                else:
                    logging.debug("收到空行，忽略")
                    return None
            else:
                # 超时未读到完整行
                logging.debug("读取超时，无数据")
                return None
        except serial.SerialException as e:
            logging.error(f"读取异常: {e}")
            self.disconnect()
            return None
    
    def disconnect(self):
        """关闭连接"""
        if self.ser and self.ser.is_open:
            self.ser.close()
            logging.info("扫码枪连接已关闭")
    
    def run(self, duration=60):
        """持续监听duration秒"""
        if not self.connect():
            return
        
        start_time = time.time()
        while time.time() - start_time < duration:
            barcode = self.read_barcode()
            if barcode:
                # 在这里处理条码，例如查询数据库、控制PLC等
                print(f"处理条码: {barcode}")
            # 避免过于频繁的轮询
            time.sleep(0.05)
        
        self.disconnect()

if __name__ == '__main__':
    scanner = BarcodeScanner(port='/dev/ttyUSB0', baudrate=9600, timeout=1.0)
    scanner.run(duration=30)

案例复盘与关键点分析：

这个案例展示了如何将Serial类、read_until、timeout、异常处理等知识点融合成一个健壮的工业级应用。在实际项目中，你可能还需要处理更复杂的协议（如Modbus RTU的CRC校验），但核心的串口读写逻辑是相通的。

4.7 常见误区与总结复盘

在串口编程中，以下几个误区是初学者最容易踩的坑：

1. 误区一：read(size) 必须读到size个字节才返回。实际上，只要设置了timeout，read(size)在超时后就会返回已读取的部分数据。因此，调用read(100)后，返回的数据长度可能小于100。
2. 误区二：write() 后立即调用 read() 一定能读到响应。设备处理指令需要时间，而且数据在传输线上有延迟。必须使用flush()确保数据发出，并设置合理的timeout等待响应。
3. 误区三：忽略 in_waiting 的轮询开销。在循环中频繁调用ser.in_waiting和ser.read(1)会消耗大量CPU。更好的做法是批量读取：data = ser.read(ser.in_waiting)。
4. 误区四：不处理异常。串口通信受环境影响大，设备可能随时断开。必须捕获SerialException、ValueError（参数错误）和OSError，并设计重连机制。
5. 误区五：使用字符串而非字节。pyserial在Python 3中只接受bytes类型。发送字符串前必须encode()，接收后必须decode()。直接使用str会导致TypeError。