Python+pyserial实现模拟串口通信的示例代码详解

在嵌入式系统开发、工业自动化和物联网设备调试中，串口通信是设备间数据交互的基础技术。Python通过pyserial库提供了强大的串口通信支持，本文ZHANID工具网将结合实际案例，详细解析如何使用pyserial实现模拟串口通信，涵盖环境配置、核心代码、异常处理及高级功能。

一、环境准备与基础概念

（一）安装pyserial库

通过pip安装pyserial库，支持跨平台（Windows/Linux/macOS）：

pip install pyserial

（二）虚拟串口工具

使用com0com或Virtual Serial Port Driver创建虚拟串口对（如COM3和COM4），实现数据自环测试。

（三）核心参数解析

二、基础通信示例

（一）单次收发示例

import serial

# 配置串口参数
ser = serial.Serial(
    port='COM3',
    baudrate=9600,
    timeout=1
)

if ser.is_open:
    print(f"串口 {ser.port} 已打开")
    
    # 发送数据（需编码为字节）
    ser.write("Hello, Serial!".encode('utf-8'))
    
    # 接收数据（读取10字节）
    data = ser.read(10)
    print(f"接收数据: {data.decode('utf-8')}")
    
    ser.close()
else:
    print("串口打开失败")

关键点：

• 发送数据需使用encode()转换为字节类型。

• 接收数据后需用decode()转换为字符串。

（二）实时交互示例

import serial
import time

ser = serial.Serial('COM3', 9600, timeout=1)

try:
    while True:
        # 发送用户输入的数据
        user_input = input("输入要发送的数据（输入exit退出）: ")
        if user_input.lower() == 'exit':
            break
        ser.write((user_input + '\n').encode('utf-8'))
        
        # 接收设备响应
        time.sleep(0.5)  # 等待设备响应
        if ser.in_waiting > 0:
            response = ser.readline().decode('utf-8').strip()
            print(f"设备响应: {response}")
finally:
    ser.close()
    print("串口已关闭")

关键点：

• 使用in_waiting检查缓冲区是否有数据。

• 通过try-finally确保串口资源释放。

三、高级功能实现

（一）多线程双向通信

import serial
import threading

def send_data(ser):
    while True:
        ser.write("Ping\n".encode('utf-8'))
        time.sleep(1)

def receive_data(ser):
    while True:
        if ser.in_waiting > 0:
            data = ser.readline().decode('utf-8').strip()
            print(f"接收: {data}")

ser = serial.Serial('COM3', 9600, timeout=1)

send_thread = threading.Thread(target=send_data, args=(ser,))
recv_thread = threading.Thread(target=receive_data, args=(ser,))

send_thread.start()
recv_thread.start()

send_thread.join()
recv_thread.join()
ser.close()

关键点：

• 使用多线程实现同时收发。

• 通过join()等待线程结束。

（二）AT指令交互示例

import serial
import re

def send_at_cmd(ser, cmd, wait_for_ok=True):
    ser.write((cmd + '\r\n').encode('utf-8'))
    if wait_for_ok:
        max_retries = 10
        retries = 0
        while retries < max_retries:
            line = ser.readline().decode('utf-8').strip()
            if re.search(r'OK', line):
                print("指令执行成功")
                return True
            retries += 1
        print("指令执行超时")
        return False

ser = serial.Serial('COM3', 9600, timeout=1)
send_at_cmd(ser, 'AT+CFUN=1')  # 开启模块功能
ser.close()

关键点：

• 通过正则表达式匹配响应。

• 实现超时重试机制。

四、常见问题与解决方案

（一）串口无法打开

• 原因：端口号错误、设备未连接、被其他程序占用。

• 解决方案：

  import serial.tools.list_ports
  ports = list(serial.tools.list_ports.comports())
  for port in ports:
      print(f"可用端口: {port.device}")

（二）数据乱码

• 原因：波特率、数据位、校验位等参数不匹配。

• 解决方案：确保发送端和接收端参数一致。

（三）接收数据不完整

• 原因：未等待足够时间或缓冲区不足。

• 解决方案：

  # 方法1：增加超时时间
  ser = serial.Serial('COM3', 9600, timeout=2)
  
  # 方法2：循环读取直到换行符
  def read_line(ser):
      line = b''
      while True:
          char = ser.read(1)
          if char == b'\n' or char == b'':
              break
          line += char
      return line.decode('utf-8').strip()

五、实战案例：传感器数据采集

import serial
import csv
import time

ser = serial.Serial('COM3', 9600, timeout=1)

with open('sensor_data.csv', 'w', newline='') as csvfile:
    writer = csv.writer(csvfile)
    writer.writerow(['Timestamp', 'Temperature', 'Humidity'])
    
    try:
        while True:
            # 发送读取指令
            ser.write("READ_DATA\n".encode('utf-8'))
            
            # 接收并解析数据（假设格式为"TEMP:25.5,HUM:60.0"）
            data = ser.readline().decode('utf-8').strip()
            if data:
                try:
                    temp, hum = data.split(',')
                    temp_val = float(temp.split(':')[1])
                    hum_val = float(hum.split(':')[1])
                    timestamp = time.strftime('%Y-%m-%d %H:%M:%S')
                    writer.writerow([timestamp, temp_val, hum_val])
                    print(f"{timestamp} - 温度: {temp_val}°C, 湿度: {hum_val}%")
                except ValueError:
                    print("数据解析错误")
    except KeyboardInterrupt:
        print("采集终止")
    finally:
        ser.close()

关键点：

• 将数据写入CSV文件。

• 添加异常处理应对数据格式错误。

六、总结与最佳实践

1. 资源管理：

• 使用with语句或try-finally确保串口关闭。

• 避免频繁打开/关闭串口，建议保持长连接。

2. 性能优化：

• 根据数据量调整缓冲区大小。

• 对高频数据采集使用多线程。

3. 调试工具：

• 使用miniterm工具快速测试：

  python -m serial.tools.miniterm COM3 9600

4. 扩展性：

• 结合pyqtgraph或matplotlib实现数据可视化。

• 通过Flask或FastAPI将串口数据发布为HTTP接口。

通过本文的示例代码与解析，开发者可快速掌握pyserial的核心功能，并根据实际需求扩展出更复杂的串口通信应用。在工业控制、智能家居等场景中，pyserial将成为连接物理世界与数字世界的桥梁。