Python调用RS232串口通信：完整实现指南与实战技巧

作者：蛮不讲李2025.09.25 17:12浏览量：18

简介：本文详细讲解Python如何通过串口（RS232）实现设备通信，涵盖库选择、代码实现、错误处理及优化建议，适合工业控制、仪器仪表开发者。

Python调用RS232串口通信：完整实现指南与实战技巧

一、RS232接口基础与Python适用场景

RS232（推荐标准232）是工业领域广泛使用的串行通信协议，采用差分信号传输，具有抗干扰能力强、传输距离远（通常15米内）的特点。典型应用场景包括：

  • 工业PLC设备控制（如西门子S7-200）
  • 医疗仪器数据采集（如心电图机）
  • 实验室设备通信（如电子天平、光谱仪）
  • 嵌入式系统调试（如STM32开发板）

Python通过pyserial库可高效实现RS232通信，其优势在于：

  1. 跨平台支持（Windows/Linux/macOS）
  2. 简洁的API设计
  3. 丰富的错误处理机制
  4. 与NumPy/Pandas等科学计算库无缝集成

二、环境准备与依赖安装

2.1 系统级配置

Windows用户需确认：

  • 设备管理器中COM端口已正确识别
  • 无其他程序占用串口（如厂商提供的调试工具）

Linux用户需检查：

  1. ls /dev/ttyS*  # 查看物理串口
  2. ls /dev/ttyUSB*  # 查看USB转串口设备
  3. sudo usermod -aG dialout $USER  # 添加用户到拨号组（无需重启）

2.2 Python库安装

推荐使用最新稳定版pyserial：

  1. pip install pyserial --upgrade
  2. # 验证安装
  3. python -c "import serial; print(serial.__version__)"

三、核心通信实现

3.1 基础连接建立

  1. import serial
  3. def create_serial_connection(port, baudrate=9600, timeout=1):
  4.     """
  5.     创建串口连接
  6.     :param port: 端口名（如'COM3'或'/dev/ttyUSB0'）
  7.     :param baudrate: 波特率（常用9600/19200/115200）
  8.     :param timeout: 读超时（秒）
  9.     :return: Serial对象
  10.     """
  11.     try:
  12.         ser = serial.Serial(
  13.             port=port,
  14.             baudrate=baudrate,
  15.             bytesize=serial.EIGHTBITS,
  16.             parity=serial.PARITY_NONE,
  17.             stopbits=serial.STOPBITS_ONE,
  18.             timeout=timeout
  19.         )
  20.         print(f"成功连接至 {port}，参数：{ser.get_settings()}")
  21.         return ser
  22.     except serial.SerialException as e:
  23.         print(f"连接失败：{str(e)}")
  24.         return None
  26. # 使用示例
  27. conn = create_serial_connection('COM3', 115200)

3.2 数据收发实现

发送数据（字节流）

  1. def send_data(ser, data):
  2.     """
  3.     发送字节数据
  4.     :param ser: Serial对象
  5.     :param data: bytes或bytearray类型
  6.     """
  7.     if ser and ser.is_open:
  8.         ser.write(data)
  9.         print(f"已发送：{data.hex()}")
  10.     else:
  11.         print("串口未连接")
  13. # 示例：发送Modbus RTU请求
  14. modbus_request = bytes.fromhex('01 03 00 00 00 02 C4 0B')
  15. send_data(conn, modbus_request)

接收数据（带校验）

  1. def receive_data(ser, expected_length=None):
  2.     """
  3.     接收数据（支持长度校验）
  4.     :param ser: Serial对象
  5.     :param expected_length: 期望接收的字节数
  6.     :return: 接收到的字节数据
  7.     """
  8.     if not ser or not ser.is_open:
  9.         return None
  11.     received = bytearray()
  12.     start_time = time.time()
  14.     while True:
  15.         if ser.in_waiting > 0:
  16.             byte = ser.read(1)
  17.             received.extend(byte)
  18.             # 简单校验：收到停止位（示例）
  19.             if len(received) >= expected_length if expected_length else False:
  20.                 break
  21.         else:
  22.             if time.time() - start_time > ser.timeout:
  23.                 print(f"接收超时，已收{len(received)}字节")
  24.                 break
  25.             time.sleep(0.01)  # 避免CPU占用过高
  27.     print(f"接收到：{received.hex()}")
  28.     return received

3.3 完整通信流程示例

  1. import time
  3. def modbus_rtu_communication(port, slave_id, register_addr, count):
  4.     """
  5.     Modbus RTU读取保持寄存器示例
  6.     :param port: 串口名称
  7.     :param slave_id: 从站地址
  8.     :param register_addr: 寄存器起始地址
  9.     :param count: 读取寄存器数量
  10.     """
  11.     # 1. 构建请求帧
  12.     # 请求格式：从站地址(1) + 功能码(1) + 起始地址(2) + 寄存器数(2) + CRC(2)
  13.     request = bytearray([
  14.         slave_id,
  15.         0x03,  # 读取保持寄存器
  16.         (register_addr >> 8) & 0xFF,
  17.         register_addr & 0xFF,
  18.         (count >> 8) & 0xFF,
  19.         count & 0xFF
  20.     ])
  22.     # 2. 计算CRC校验
  23.     crc = calculate_crc(request)
  24.     request.extend(crc)
  26.     # 3. 通信循环
  27.     with create_serial_connection(port, 19200) as ser:
  28.         if ser:
  29.             # 发送请求
  30.             send_data(ser, request)
  32.             # 接收响应（Modbus RTU响应最小长度为5字节）
  33.             response = receive_data(ser, 5)
  35.             if response and len(response) >= 5:
  36.                 # 验证CRC
  37.                 received_crc = response[-2:]
  38.                 calc_crc = calculate_crc(response[:-2])
  39.                 if received_crc == calc_crc:
  40.                     # 解析数据（跳过从站地址和功能码）
  41.                     byte_count = response[2]
  42.                     register_values = []
  43.                     for i in range(byte_count // 2):
  44.                         start_idx = 3 + i * 2
  45.                         value = (response[start_idx] << 8) | response[start_idx + 1]
  46.                         register_values.append(value)
  47.                     print(f"读取到寄存器值：{register_values}")
  48.                 else:
  49.                     print("CRC校验失败")
  51. def calculate_crc(data):
  52.     """Modbus CRC16计算"""
  53.     crc = 0xFFFF
  54.     for byte in data:
  55.         crc ^= byte
  56.         for _ in range(8):
  57.             if crc & 0x0001:
  58.                 crc >>= 1
  59.                 crc ^= 0xA001
  60.             else:
  61.                 crc >>= 1
  62.     return bytes([crc & 0xFF, (crc >> 8) & 0xFF])

四、高级应用与优化

4.1 多线程通信架构

  1. import threading
  2. import queue
  4. class SerialCommunicator:
  5.     def __init__(self, port):
  6.         self.ser = create_serial_connection(port)
  7.         self.send_queue = queue.Queue()
  8.         self.receive_thread = threading.Thread(target=self._receive_loop)
  9.         self.receive_thread.daemon = True
  10.         self.receive_thread.start()
  12.     def send_async(self, data):
  13.         self.send_queue.put(data)
  15.     def _receive_loop(self):
  16.         while self.ser and self.ser.is_open:
  17.             if self.ser.in_waiting > 0:
  18.                 data = self.ser.read(self.ser.in_waiting)
  19.                 # 处理接收到的数据（可添加解析逻辑）
  20.                 print(f"异步接收：{data.hex()}")
  21.             time.sleep(0.01)
  23.     def close(self):
  24.         if self.ser:
  25.             self.ser.close()

4.2 性能优化技巧

  1. 缓冲区管理：

    1. ser = serial.Serial(...)
    2. ser.set_buffer_size(rx_size=4096, tx_size=4096)  # 增大缓冲区

  2. 流控制配置：

    1. ser = serial.Serial(
    2.     ...,
    3.     xonxoff=True,  # 软件流控
    4.     rtscts=True    # 硬件流控（需设备支持）
    5. )

  3. 非阻塞读取：

    1. def non_blocking_read(ser):
    2.     if ser.in_waiting > 0:
    3.         return ser.read(ser.in_waiting)
    4.     return None

五、常见问题解决方案

5.1 连接失败排查

  1. 端口占用检查：

    • Windows：使用netstat -ano | findstr "COM3"
    • Linux：lsof | grep /dev/ttyUSB0

  2. 权限问题：

    1. sudo chmod 666 /dev/ttyUSB0  # 临时解决方案

5.2 数据乱码处理

  1. 编码转换：

    1. # 接收ASCII数据
    2. ascii_data = received_bytes.decode('ascii', errors='ignore')
    4. # 接收HEX字符串
    5. hex_str = received_bytes.hex().upper()

  2. 帧同步策略：

    • 添加起始/结束标记（如0xAA 0x55）
    • 实现超时重传机制

六、工业级实现建议

  1. 看门狗机制：

    1. class Watchdog:
    2.     def __init__(self, timeout=5):
    3.         self.timeout = timeout
    4.         self.last_activity = time.time()
    6.     def touch(self):
    7.         self.last_activity = time.time()
    9.     def is_alive(self):
    10.         return (time.time() - self.last_activity) < self.timeout

  2. 日志记录系统：

    1. import logging
    2. logging.basicConfig(
    3.     filename='serial_com.log',
    4.     level=logging.DEBUG,
    5.     format='%(asctime)s - %(levelname)s - %(message)s'
    6. )

  3. 配置管理：

    1. import configparser
    2. config = configparser.ConfigParser()
    3. config.read('serial_config.ini')
    4. port = config.get('SERIAL', 'port')
    5. baudrate = config.getint('SERIAL', 'baudrate')

七、总结与扩展

Python通过pyserial库实现RS232通信具有高效、灵活的特点。实际开发中需注意：

  1. 严格遵循设备通信协议（如Modbus、DNP3）
  2. 实现完善的错误处理和重试机制
  3. 考虑使用异步框架（如asyncio）提升并发性能

扩展方向：

  • 结合WebSocket实现远程串口访问
  • 开发可视化调试工具（基于PyQt/Tkinter）
  • 集成到IoT平台（如MQTT桥接）

完整代码示例已包含基础通信、协议实现和错误处理，开发者可根据实际需求调整参数和逻辑。

最热文章