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

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

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

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

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

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

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

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

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

二、环境准备与依赖安装

2.1 系统级配置

Windows用户需确认：

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

Linux用户需检查：

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

2.2 Python库安装

推荐使用最新稳定版pyserial：

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

三、核心通信实现

3.1 基础连接建立

import serial
def create_serial_connection(port, baudrate=9600, timeout=1):
    """
    创建串口连接
    :param port: 端口名（如'COM3'或'/dev/ttyUSB0'）
    :param baudrate: 波特率（常用9600/19200/115200）
    :param timeout: 读超时（秒）
    :return: Serial对象
    """
    try:
        ser = serial.Serial(
            port=port,
            baudrate=baudrate,
            bytesize=serial.EIGHTBITS,
            parity=serial.PARITY_NONE,
            stopbits=serial.STOPBITS_ONE,
            timeout=timeout
        )
        print(f"成功连接至 {port}，参数：{ser.get_settings()}")
        return ser
    except serial.SerialException as e:
        print(f"连接失败：{str(e)}")
        return None
# 使用示例
conn = create_serial_connection('COM3', 115200)

3.2 数据收发实现

发送数据（字节流）

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

接收数据（带校验）

def receive_data(ser, expected_length=None):
    """
    接收数据（支持长度校验）
    :param ser: Serial对象
    :param expected_length: 期望接收的字节数
    :return: 接收到的字节数据
    """
    if not ser or not ser.is_open:
        return None
    received = bytearray()
    start_time = time.time()
    while True:
        if ser.in_waiting > 0:
            byte = ser.read(1)
            received.extend(byte)
            # 简单校验：收到停止位（示例）
            if len(received) >= expected_length if expected_length else False:
                break
        else:
            if time.time() - start_time > ser.timeout:
                print(f"接收超时，已收{len(received)}字节")
                break
            time.sleep(0.01)  # 避免CPU占用过高
    print(f"接收到：{received.hex()}")
    return received

3.3 完整通信流程示例

import time
def modbus_rtu_communication(port, slave_id, register_addr, count):
    """
    Modbus RTU读取保持寄存器示例
    :param port: 串口名称
    :param slave_id: 从站地址
    :param register_addr: 寄存器起始地址
    :param count: 读取寄存器数量
    """
    # 1. 构建请求帧
    # 请求格式：从站地址(1) + 功能码(1) + 起始地址(2) + 寄存器数(2) + CRC(2)
    request = bytearray([
        slave_id,
        0x03,  # 读取保持寄存器
        (register_addr >> 8) & 0xFF,
        register_addr & 0xFF,
        (count >> 8) & 0xFF,
        count & 0xFF
    ])
    # 2. 计算CRC校验
    crc = calculate_crc(request)
    request.extend(crc)
    # 3. 通信循环
    with create_serial_connection(port, 19200) as ser:
        if ser:
            # 发送请求
            send_data(ser, request)
            # 接收响应（Modbus RTU响应最小长度为5字节）
            response = receive_data(ser, 5)
            if response and len(response) >= 5:
                # 验证CRC
                received_crc = response[-2:]
                calc_crc = calculate_crc(response[:-2])
                if received_crc == calc_crc:
                    # 解析数据（跳过从站地址和功能码）
                    byte_count = response[2]
                    register_values = []
                    for i in range(byte_count // 2):
                        start_idx = 3 + i * 2
                        value = (response[start_idx] << 8) | response[start_idx + 1]
                        register_values.append(value)
                    print(f"读取到寄存器值：{register_values}")
                else:
                    print("CRC校验失败")
def calculate_crc(data):
    """Modbus CRC16计算"""
    crc = 0xFFFF
    for byte in data:
        crc ^= byte
        for _ in range(8):
            if crc & 0x0001:
                crc >>= 1
                crc ^= 0xA001
            else:
                crc >>= 1
    return bytes([crc & 0xFF, (crc >> 8) & 0xFF])

四、高级应用与优化

4.1 多线程通信架构

import threading
import queue
class SerialCommunicator:
    def __init__(self, port):
        self.ser = create_serial_connection(port)
        self.send_queue = queue.Queue()
        self.receive_thread = threading.Thread(target=self._receive_loop)
        self.receive_thread.daemon = True
        self.receive_thread.start()
    def send_async(self, data):
        self.send_queue.put(data)
    def _receive_loop(self):
        while self.ser and self.ser.is_open:
            if self.ser.in_waiting > 0:
                data = self.ser.read(self.ser.in_waiting)
                # 处理接收到的数据（可添加解析逻辑）
                print(f"异步接收：{data.hex()}")
            time.sleep(0.01)
    def close(self):
        if self.ser:
            self.ser.close()

4.2 性能优化技巧

缓冲区管理：

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

流控制配置：

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

非阻塞读取：

def non_blocking_read(ser):
    if ser.in_waiting > 0:
        return ser.read(ser.in_waiting)
    return None

五、常见问题解决方案

5.1 连接失败排查

端口占用检查：
- Windows：使用netstat -ano | findstr "COM3"
- Linux：lsof | grep /dev/ttyUSB0

权限问题：

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

5.2 数据乱码处理

编码转换：

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

帧同步策略：
- 添加起始/结束标记（如0xAA 0x55）
- 实现超时重传机制

六、工业级实现建议

看门狗机制：

class Watchdog:
    def __init__(self, timeout=5):
        self.timeout = timeout
        self.last_activity = time.time()
    def touch(self):
        self.last_activity = time.time()
    def is_alive(self):
        return (time.time() - self.last_activity) < self.timeout

日志记录系统：

import logging
logging.basicConfig(
    filename='serial_com.log',
    level=logging.DEBUG,
    format='%(asctime)s - %(levelname)s - %(message)s'
)

配置管理：

import configparser
config = configparser.ConfigParser()
config.read('serial_config.ini')
port = config.get('SERIAL', 'port')
baudrate = config.getint('SERIAL', 'baudrate')

七、总结与扩展

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

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

扩展方向：

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

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

发表评论

开发者关注产品榜

最热文章

关于作者

被阅读数
被赞数
被收藏数

活动

咨询

开发者热搜

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

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

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

二、环境准备与依赖安装

2.1 系统级配置

2.2 Python库安装

三、核心通信实现

3.1 基础连接建立

3.2 数据收发实现

发送数据（字节流）

接收数据（带校验）

3.3 完整通信流程示例

四、高级应用与优化

4.1 多线程通信架构

4.2 性能优化技巧

五、常见问题解决方案

5.1 连接失败排查

5.2 数据乱码处理

六、工业级实现建议

七、总结与扩展

相关文章推荐

文心一言接入指南：通过百度智能云千帆大模型平台API调用

从 MLOps 到 LMOps 的关键技术嬗变

Sugar BI教你怎么做数据可视化 - 拓扑图，让节点连接信息一目了然

更轻量的百度百舸，CCE Stack 智算版发布

打造合规数据闭环，加速自动驾驶技术研发

LMOps 工具链与千帆大模型平台

发表评论

开发者关注产品榜

百度千帆·大模型服务及Agent开发平台

百度千帆·数据智能平台

秒哒-生成式应用开发平台

百度智能云客悦智能客服平台

最热文章

关于作者