搜索菜单
当前位置: 首页 > Python > PC端Python通过pyserial串口控制STM32开发板

PC端Python通过pyserial串口控制STM32开发板

小满
小满
发布时间：2026-10-10 | 2人浏览过
原创

在工业测控、自动化测试与原型验证场景中，利用PC端编写自动化脚本直接调度微控制器是最实用的方案之一。本文的核心目标是使用Python的pyserial库与STM32开发板建立稳定的全双工串口通信链路，通过定义规范的数据帧格式，实现PC端对STM32外设（如LED指示灯、继电器、电机）的精准控制，并实时获取下位机返回的执行状态与采样数据。

一、物理连接与硬件电气特性

实现串口控制的前提是建立正确的物理电气连接。PC端普遍仅提供USB接口，而STM32微控制器的片上USART外设运行在TTL逻辑电平（3.3V）。因此，两者通信必须通过USB转TTL串口适配模块（例如基于CH340、CP2102或FT232芯片的转接板）进行信号电平桥接，或者使用板载自带ST-LINK虚拟串口（VCP）的开发板。

在物理接线时必须牢记交叉连接规则与共地原则：转接板的发送端（TXD）连接STM32的接收引脚（如PA10/USART1_RX），转接板的接收端（RXD）连接STM32的发送引脚（如PA9/USART1_TX）。最关键的是，两者的地线（GND）必须可靠相连，以提供统一的参考电位；未共地会导致接收端识别到的电平阈值偏移，进而产生大范围误码甚至通信中断。

PC端Python通过pyserial串口控制STM32开发板
图1：PC端上位机、USB转TTL模块与STM32开发板电气接线拓扑与信号流向

二、通信协议帧设计

纯文本字符串（如直接发送"LED_ON"）虽然调试直观，但在工业与嵌入式应用中极易因字符截断、缺失转义或未定义换行产生解析故障。设计紧凑、带校验的定长或变长二进制数据帧是保证指令100%可靠执行的标准做法。

本文采用经典的帧结构设计：

  • 帧头（Header，2字节）：固定为0xAA 0x55，接收状态机以此同步识别数据帧的起始边界。
  • 命令标识（CMD，1字节）：定义操作类别，例如0x01表示写外设输出状态，0x02表示读传感器数据。
  • 数据长度（LEN，1字节）：指示紧随其后的有效载荷字节数N。
  • 数据载荷（DATA，N字节）：具体的控制参数，如LED编号及期望电平状态。
  • 校验和（Checksum，1字节）：从CMD起至DATA最后一个字节的低8位累加和，用于检测传输误码。
  • 帧尾（Tail，2字节）：固定为0x0D 0x0A（即\r\n），提供二级边界验证。
PC端Python通过pyserial串口控制STM32开发板
图2：串口通信二进制协议帧定义及下位机状态机解析流向

三、STM32下位机固件实现逻辑

在STM32端，串口接收不能采用阻塞式的HAL_UART_Receive轮询，否则会严重阻塞主循环中的其他控制任务。推荐采用USART接收中断（RXNE中断）+ 软件状态机的方式逐字节解包，或者采用DMA + 空闲中断（IDLE）接收整包。这里展示基于HAL库中断回调的状态机解包核心代码。

/* protocol.h 核心结构体与指令定义 */
#ifndef PROTOCOL_H
#define PROTOCOL_H

#include "stm32f1xx_hal.h"
#include <stdint.h>

#define FRAME_HEADER1  0xAA
#define FRAME_HEADER2  0x55
#define FRAME_TAIL1    0x0D
#define FRAME_TAIL2    0x0A

#define CMD_SET_LED    0x01
#define CMD_GET_SENSOR 0x02

typedef struct {
    uint8_t cmd;
    uint8_t len;
    uint8_t data[32];
    uint8_t checksum;
} RxFrame_t;

void Protocol_Init(UART_HandleTypeDef *huart);
void Protocol_ProcessByte(uint8_t byte);
void Protocol_HandleCommand(RxFrame_t *frame);
void Protocol_SendResponse(uint8_t cmd, uint8_t *data, uint8_t len);

#endif

在中断处理中，每到达一个字节，就推进一次状态机，避免占用CPU长时间等待：

/* protocol.c 状态机解析与响应 */
#include "protocol.h"
#include "main.h"

static UART_HandleTypeDef *g_huart;
static uint8_t rx_byte = 0;

typedef enum {
    STATE_WAIT_HEADER1 = 0,
    STATE_WAIT_HEADER2,
    STATE_WAIT_CMD,
    STATE_WAIT_LEN,
    STATE_WAIT_DATA,
    STATE_WAIT_SUM,
    STATE_WAIT_TAIL1,
    STATE_WAIT_TAIL2
} ParseState_t;

static ParseState_t state = STATE_WAIT_HEADER1;
static RxFrame_t current_frame;
static uint8_t data_index = 0;
static uint8_t calc_checksum = 0;

void Protocol_Init(UART_HandleTypeDef *huart) {
    g_huart = huart;
    HAL_UART_Receive_IT(g_huart, &rx_byte, 1);
}

void HAL_UART_RxCpltCallback(UART_HandleTypeDef *huart) {
    if (huart->Instance == g_huart->Instance) {
        Protocol_ProcessByte(rx_byte);
        HAL_UART_Receive_IT(g_huart, &rx_byte, 1);
    }
}

void Protocol_ProcessByte(uint8_t byte) {
    switch (state) {
        case STATE_WAIT_HEADER1:
            if (byte == FRAME_HEADER1) state = STATE_WAIT_HEADER2;
            break;
        case STATE_WAIT_HEADER2:
            if (byte == FRAME_HEADER2) {
                state = STATE_WAIT_CMD;
            } else {
                state = STATE_WAIT_HEADER1;
            }
            break;
        case STATE_WAIT_CMD:
            current_frame.cmd = byte;
            calc_checksum = byte;
            state = STATE_WAIT_LEN;
            break;
        case STATE_WAIT_LEN:
            current_frame.len = byte;
            calc_checksum += byte;
            data_index = 0;
            if (current_frame.len > 32) { /* 长度超限保护 */
                state = STATE_WAIT_HEADER1;
            } else if (current_frame.len == 0) {
                state = STATE_WAIT_SUM;
            } else {
                state = STATE_WAIT_DATA;
            }
            break;
        case STATE_WAIT_DATA:
            current_frame.data[data_index++] = byte;
            calc_checksum += byte;
            if (data_index >= current_frame.len) {
                state = STATE_WAIT_SUM;
            }
            break;
        case STATE_WAIT_SUM:
            current_frame.checksum = byte;
            if (current_frame.checksum == calc_checksum) {
                state = STATE_WAIT_TAIL1;
            } else {
                state = STATE_WAIT_HEADER1; /* 校验和失败，丢弃 */
            }
            break;
        case STATE_WAIT_TAIL1:
            if (byte == FRAME_TAIL1) state = STATE_WAIT_TAIL2;
            else state = STATE_WAIT_HEADER1;
            break;
        case STATE_WAIT_TAIL2:
            if (byte == FRAME_TAIL2) {
                Protocol_HandleCommand(&current_frame);
            }
            state = STATE_WAIT_HEADER1;
            break;
    }
}

void Protocol_HandleCommand(RxFrame_t *frame) {
    if (frame->cmd == CMD_SET_LED && frame->len >= 2) {
        uint8_t led_id = frame->data[0];
        uint8_t led_state = frame->data[1];
        if (led_id == 1) {
            HAL_GPIO_WritePin(GPIOC, GPIO_PIN_13, led_state ? GPIO_PIN_RESET : GPIO_PIN_SET);
        }
        /* 返回执行回执: 0x00 代表执行成功 */
        uint8_t ack_data[2] = {led_id, 0x00};
        Protocol_SendResponse(CMD_SET_LED, ack_data, 2);
    }
}

void Protocol_SendResponse(uint8_t cmd, uint8_t *data, uint8_t len) {
    uint8_t tx_buf[40];
    tx_buf[0] = FRAME_HEADER1;
    tx_buf[1] = FRAME_HEADER2;
    tx_buf[2] = cmd;
    tx_buf[3] = len;
    uint8_t sum = cmd + len;
    for (uint8_t i = 0; i < len; i++) {
        tx_buf[4 + i] = data[i];
        sum += data[i];
    }
    tx_buf[4 + len] = sum;
    tx_buf[5 + len] = FRAME_TAIL1;
    tx_buf[6 + len] = FRAME_TAIL2;
    HAL_UART_Transmit(g_huart, tx_buf, 7 + len, 100);
}

四、PC端Python环境准备与pyserial安装

在PC端，确保Python环境为3.8或更高版本。通过包管理工具安装pyserial官方模块：

pip install pyserial

注意：PyPI中存在名为serial的历史废弃库，切勿执行pip install serial，否则会导致命名冲突并缺失核心串口功能。

五、Python上位机控制脚本实现

上位机代码封装为一个通用的STM32Controller类。类内部管理串口连接的生命周期、数据封包、累加和计算以及带超时等待的回执校验：

# stm32_controller.py
import serial
import serial.tools.list_ports
import time
import struct

class STM32Controller:
    HEADER = bytes([0xAA, 0x55])
    TAIL = bytes([0x0D, 0x0A])
    CMD_SET_LED = 0x01
    CMD_GET_SENSOR = 0x02

    def __init__(self, port=None, baudrate=115200, timeout=1.0):
        self.port = port
        self.baudrate = baudrate
        self.timeout = timeout
        self.ser = None

    @staticmethod
    def list_available_ports():
        """列出当前操作系统中所有可用串口"""
        ports = serial.tools.list_ports.comports()
        available = []
        for p in ports:
            available.append((p.device, p.description))
        return available

    def connect(self):
        """建立串口物理连接并清空缓冲区"""
        if not self.port:
            available = self.list_available_ports()
            if not available:
                raise ConnectionError("未检测到任何可用串口设备")
            self.port = available[0][0]
            print(f"自动选择串口设备: {self.port}")

        self.ser = serial.Serial(
            port=self.port,
            baudrate=self.baudrate,
            bytesize=serial.EIGHTBITS,
            parity=serial.PARITY_NONE,
            stopbits=serial.STOPBITS_ONE,
            timeout=self.timeout
        )
        # 打开串口时DTR/RTS可能导致部分STM32开发板复位，等待初始化稳定
        time.sleep(0.1)
        self.ser.reset_input_buffer()
        self.ser.reset_output_buffer()
        print(f"已连接至 {self.port}，波特率: {self.baudrate}")

    def close(self):
        if self.ser and self.ser.is_open:
            self.ser.close()
            print("串口已关闭")

    def _build_frame(self, cmd: int, payload: bytes) -> bytes:
        """构建标准协议帧并计算累加和"""
        length = len(payload)
        checksum = (cmd + length + sum(payload)) & 0xFF
        frame = self.HEADER + bytes([cmd, length]) + payload + bytes([checksum]) + self.TAIL
        return frame

    def _read_response(self) -> dict:
        """读取并解析下位机返回的确认帧"""
        start_time = time.time()
        # 搜寻帧头 0xAA 0x55
        while time.time() - start_time < self.timeout:
            b1 = self.ser.read(1)
            if b1 == bytes([0xAA]):
                b2 = self.ser.read(1)
                if b2 == bytes([0x55]):
                    break
        else:
            raise TimeoutError("读取响应超时：未收到合法帧头")

        header_info = self.ser.read(2)
        if len(header_info) < 2:
            raise TimeoutError("读取响应超时：头信息不全")
        cmd, length = header_info[0], header_info[1]

        payload = self.ser.read(length) if length > 0 else b""
        if len(payload) < length:
            raise TimeoutError("读取响应超时：载荷数据截断")

        checksum_byte = self.ser.read(1)
        tail_bytes = self.ser.read(2)
        if len(checksum_byte) < 1 or len(tail_bytes) < 2:
            raise TimeoutError("读取响应超时：尾部不完整")

        calc_sum = (cmd + length + sum(payload)) & 0xFF
        if checksum_byte[0] != calc_sum:
            raise ValueError(f"校验和错误: 计算值={calc_sum}, 收到值={checksum_byte[0]}")

        if tail_bytes != self.TAIL:
            raise ValueError("帧尾验证失败")

        return {"cmd": cmd, "payload": payload}

    def set_led_state(self, led_id: int, turn_on: bool) -> bool:
        """下发LED控制指令并验证回执"""
        state_val = 1 if turn_on else 0
        payload = bytes([led_id, state_val])
        frame = self._build_frame(self.CMD_SET_LED, payload)
        
        self.ser.write(frame)
        self.ser.flush()

        response = self._read_response()
        if response["cmd"] == self.CMD_SET_LED and len(response["payload"]) >= 2:
            ret_id, status = response["payload"][0], response["payload"][1]
            return (ret_id == led_id) and (status == 0)
        return False

# 示例业务运行逻辑
if __name__ == "__main__":
    controller = STM32Controller(baudrate=115200)
    try:
        controller.connect()
        print("准备下发指令：开启LED1...")
        success = controller.set_led_state(led_id=1, turn_on=True)
        print(f"执行结果: {'成功' if success else '失败'}")
        
        time.sleep(1)
        
        print("准备下发指令：关闭LED1...")
        success = controller.set_led_state(led_id=1, turn_on=False)
        print(f"执行结果: {'成功' if success else '失败'}")
    finally:
        controller.close()
PC端Python通过pyserial串口控制STM32开发板
图3：PC端Python控制脚本与STM32下位机闭环交互时序流程

六、常见通信故障诊断与排查决策

在实际调试过程中，通信异常通常集中在物理链路、波特率匹配以及数据流同步三个环节。遇到故障时可按照以下决策路径逐级排查：

PC端Python通过pyserial串口控制STM32开发板
图4：串口通信故障分类与系统排查诊断流程

七、串口控制关键要素与最佳实践总结

为了在长时间运行或工业环境中保持控制链路的高可靠性，需要遵循以下配置与开发准则：

PC端Python通过pyserial串口控制STM32开发板
图5：PC端与STM32串口控制体系关键配置与开发准则全景总结

综上所述，PC端通过Python与STM32开发板进行通信的核心，在于建立严格的物理电气兼容、自解释的二进制协议帧、状态机式的中断解包，以及具备确认回执的上位机驱动封装。掌握这一整套方法后，即可快速扩展出电机调速、传感器采样曲线绘制、产线自动烧录校准等各类复杂的工控应用。

相关文章

相关标签: Python pyserial

本站声明

本文内容仅供参考，不代表本站观点。转载或引用请注明出处。如发现内容问题，请通过联系我们页面反馈。

作者最新文章

全部内容

Python 相关产品和应用

全部下载
Python Selenium
Selenium

Selenium

4.51.0 · Windows

用于浏览器自动化和 Web 测试的 Python 工具。
Python 机器学习
scikit-learn

scikit-learn

1.9.1 · Windows

常用 Python 机器学习算法与模型评估工具库。
Python 代码质量
Black

Black

26.10.1 · Windows

强调一致性的 Python 自动代码格式化工具。
Python 代码质量
Ruff

Ruff

0.17.1 · macOS

由 Rust 编写的高速 Python 代码检查和格式化工具。
Python pytest
pytest

pytest

9.1.1 · Windows

简洁而强大的 Python 自动化测试框架。
Python FastAPI API
FastAPI

FastAPI

0.146.0 · Windows

高性能 Python API 框架，支持类型提示和自动接口文档。
Python Flask Web开发
Flask

Flask

3.1.3 · Windows

轻量灵活的 Python Web 应用框架。
Python Django Web开发
Django

Django

6.1.2 · Windows

成熟的 Python Web 框架，适合构建内容站点和业务系统。
Python Spyder
Spyder

Spyder

6.1.7 · Windows

面向科学计算与数据分析的 Python 集成开发环境。
Python SciPy
SciPy

SciPy

1.18.1 · Windows

面向科学、工程和技术计算的 Python 算法库。
Python Pandas 数据分析
pandas

pandas

3.0.6 · Windows

Python 数据分析库，提供 DataFrame 与丰富的数据处理能力。
Python NumPy
NumPy

NumPy

2.5.4 · Windows

Python 科学计算基础库，提供高性能多维数组与数值运算。

文章分类

专题列表

ipython.cn 学习网
人工智能 Java AI行业服务