基于Python和STM32的简易自动化测试上位机开发

在嵌入式电路板研发与批量出厂检测中，人工逐项测量电压、复位电路与通信接口耗时费力且容易漏检。利用PC端Python脚本构建自动化测试上位机，配合STM32微控制器作为下位机执行治具，是一种开发周期短、成本低且扩展性强的测试方案。PC端负责根据测试用例下发控制指令、收集检测数据并给出合格判定；STM32下位机负责根据指令执行GPIO电平输出、继电器供电控制、多路ADC模拟量采样以及串口回环收发。本文按照通信协议制定、STM32下位机固件编写、Python上位机自动化框架实现以及软硬件联调与异常排查的具体流程展开讲解。

一、系统总体架构与通信协议设计

自动化测试系统的核心在于上位机与下位机职责的清晰划分。上位机处于主导地位，运行在开发人员的测试工控机或个人电脑上，使用Python生态中的自动化执行逻辑控制整个测试进程；下位机以STM32F103或STM32F401等常见Cortex-M核心板为控制核心，作为连接待测板（DUT, Device Under Test）的硬件探针接口。

两者的物理连接通常采用USB转TTL串口模块（如CH340或CP2102芯片）。串口参数统一配置为：波特率 115200 bps、数据位 8 位、停止位 1 位、无奇偶校验位（115200 8-N-1）。

1.1 定长帧通信协议格式

为保证串口在干扰或断线情况下仍能稳定解析，避免因字节粘包或丢失导致控制混乱，通信采用带固定帧头、长度指示、累加校验和及换行帧尾的二进制协议结构：

• 帧头（2字节）：固定为 0xAA 0x55，用于接收端快速定位一帧数据的起始位置。
• 命令字 CMD（1字节）：指定本次交互的操作动作，如握手、设置GPIO、采集ADC等。
• 数据长度 LEN（1字节）：表示后续数据载荷（Payload）的实际字节数（0~8字节）。
• 数据载荷 PAYLOAD（N字节）：具体参数内容，例如目标通道号、输出状态值或测量数值。
• 校验和 CHKSUM（1字节）：对 CMD、LEN 以及 PAYLOAD 中所有字节进行累加，取低 8 位（即 Sum & 0xFF），用于验证数据传输完整性。
• 帧尾（2字节）：固定为 0x0D 0x0A（即 ASCII 换行符 ），作为帧传输结束边界。

1.2 功能命令分配表

本套简易自动化测试系统定义了四个核心命令字，满足对目标板的基本功能自动化验证需求：

二、STM32 下位机固件与帧解析状态机实现

STM32固件的核心任务是及时、非阻塞地接收串口数据，并通过有限状态机（FSM）逐字节解析。当校验和与帧尾均通过后，触发对应的硬件驱动执行，并将测量结果打包成应答帧发送回上位机。

2.1 下位机帧接收有限状态机

接收状态机包含 7 个标准状态：等待帧头1、等待帧头2、提取命令字、提取长度、接收数据载荷、核对校验和以及确认帧尾。任何一个步骤收到非法数据时，状态机立即重置回初始状态，从而阻断错误数据的持续扩散。

/* protocol.h - 协议定义与状态机枚举 */
#ifndef PROTOCOL_H
#define PROTOCOL_H

#include <stdint.h>

#define FRAME_HEAD1  0xAA
#define FRAME_HEAD2  0x55
#define FRAME_TAIL1  0x0D
#define FRAME_TAIL2  0x0A

typedef enum {
    STATE_WAIT_HEAD1 = 0,
    STATE_WAIT_HEAD2,
    STATE_GET_CMD,
    STATE_GET_LEN,
    STATE_GET_DATA,
    STATE_CHECK_SUM,
    STATE_WAIT_TAIL1,
    STATE_WAIT_TAIL2
} RxState_t;

typedef struct {
    uint8_t cmd;
    uint8_t len;
    uint8_t data[16];
    uint8_t chksum;
} Frame_t;

void Protocol_Init(void);
void Protocol_ProcessByte(uint8_t byte);
void Protocol_ExecuteCommand(const Frame_t *rx_frame);
void Protocol_SendResponse(uint8_t cmd, const uint8_t *payload, uint8_t len);

#endif

2.2 状态机接收与命令分发 C 语言实现

在 protocol.c 中，通过单字节输入函数逐步推动状态转换。命令执行函数针对不同命令调用相应的硬件外设驱动：

/* protocol.c - 状态机流转与执行 */
#include "protocol.h"
#include "main.h"

extern UART_HandleTypeDef huart1;
extern ADC_HandleTypeDef hadc1;

static RxState_t s_state = STATE_WAIT_HEAD1;
static Frame_t s_rx_frame;
static uint8_t s_data_index = 0;
static uint8_t s_calc_chksum = 0;

void Protocol_ProcessByte(uint8_t byte) {
    switch (s_state) {
        case STATE_WAIT_HEAD1:
            if (byte == FRAME_HEAD1) s_state = STATE_WAIT_HEAD2;
            break;
        case STATE_WAIT_HEAD2:
            if (byte == FRAME_HEAD2) {
                s_state = STATE_GET_CMD;
                s_calc_chksum = 0;
            } else {
                s_state = STATE_WAIT_HEAD1;
            }
            break;
        case STATE_GET_CMD:
            s_rx_frame.cmd = byte;
            s_calc_chksum += byte;
            s_state = STATE_GET_LEN;
            break;
        case STATE_GET_LEN:
            s_rx_frame.len = byte;
            s_calc_chksum += byte;
            s_data_index = 0;
            if (s_rx_frame.len == 0) {
                s_state = STATE_CHECK_SUM;
            } else if (s_rx_frame.len <= 16) {
                s_state = STATE_GET_DATA;
            } else {
                s_state = STATE_WAIT_HEAD1; /* 长度超限，异常重置 */
            }
            break;
        case STATE_GET_DATA:
            s_rx_frame.data[s_data_index++] = byte;
            s_calc_chksum += byte;
            if (s_data_index >= s_rx_frame.len) {
                s_state = STATE_CHECK_SUM;
            }
            break;
        case STATE_CHECK_SUM:
            if (byte == (s_calc_chksum & 0xFF)) {
                s_state = STATE_WAIT_TAIL1;
            } else {
                s_state = STATE_WAIT_HEAD1; /* 校验和错误，丢弃 */
            }
            break;
        case STATE_WAIT_TAIL1:
            if (byte == FRAME_TAIL1) {
                s_state = STATE_WAIT_TAIL2;
            } else {
                s_state = STATE_WAIT_HEAD1;
            }
            break;
        case STATE_WAIT_TAIL2:
            if (byte == FRAME_TAIL2) {
                Protocol_ExecuteCommand(&s_rx_frame);
            }
            s_state = STATE_WAIT_HEAD1;
            break;
        default:
            s_state = STATE_WAIT_HEAD1;
            break;
    }
}

void Protocol_SendResponse(uint8_t cmd, const uint8_t *payload, uint8_t len) {
    uint8_t tx_buf[24];
    uint8_t chk = cmd + len;
    uint8_t idx = 0;

    tx_buf[idx++] = FRAME_HEAD1;
    tx_buf[idx++] = FRAME_HEAD2;
    tx_buf[idx++] = cmd;
    tx_buf[idx++] = len;

    for (uint8_t i = 0; i < len; i++) {
        tx_buf[idx++] = payload[i];
        chk += payload[i];
    }

    tx_buf[idx++] = chk & 0xFF;
    tx_buf[idx++] = FRAME_TAIL1;
    tx_buf[idx++] = FRAME_TAIL2;

    HAL_UART_Transmit(&huart1, tx_buf, idx, 100);
}

void Protocol_ExecuteCommand(const Frame_t *rx) {
    uint8_t resp_payload[4];

    switch (rx->cmd) {
        case 0x01: /* PING 握手探活 */
            resp_payload[0] = 0x00; /* 0x00 代表正常就绪 */
            Protocol_SendResponse(0x01, resp_payload, 1);
            break;

        case 0x02: /* 继电器供电控制 (PA1) */
            if (rx->len >= 1) {
                GPIO_PinState pin_state = (rx->data[0] == 0x01) ? GPIO_PIN_SET : GPIO_PIN_RESET;
                HAL_GPIO_WritePin(GPIOA, GPIO_PIN_1, pin_state);
                HAL_Delay(50); /* 等待触点与滤波电容稳定 */
                resp_payload[0] = (uint8_t)HAL_GPIO_ReadPin(GPIOA, GPIO_PIN_1);
                Protocol_SendResponse(0x02, resp_payload, 1);
            }
            break;

        case 0x03: /* ADC 电压采集 (PA0) */
            {
                uint32_t adc_accum = 0;
                for (int i = 0; i < 8; i++) {
                    HAL_ADC_Start(&hadc1);
                    HAL_ADC_PollForConversion(&hadc1, 10);
                    adc_accum += HAL_ADC_GetValue(&hadc1);
                }
                uint16_t adc_avg = (uint16_t)(adc_accum / 8);
                /* 假设 3.3V 供电参考电压，12位ADC分辨率: mV = (raw * 3300) / 4095 */
                uint16_t mv = (uint16_t)(((uint32_t)adc_avg * 3300) / 4095);
                resp_payload[0] = (mv >> 8) & 0xFF;
                resp_payload[1] = mv & 0xFF;
                Protocol_SendResponse(0x03, resp_payload, 2);
            }
            break;

        default:
            break;
    }
}

三、Python 上位机自动化测试框架开发

上位机测试框架采用 Python 3 编写，主要依赖串口库 pyserial。执行环境可通过常规命令准备：

pip install pyserial

自动化测试脚本由两个类构成：STM32Controller 封装了底层协议的封包、发送、超时等待、校验和解包；TestSuiteRunner 负责组织具体的测试用例逻辑、判定上下限阈值，并输出最终测试统计表。

3.1 上位机完整实现源码

以下为 test_runner.py 的完整代码：

# test_runner.py - 基于 Python 的简易自动化测试执行器
import time
import serial
from typing import Optional, Tuple, List, Dict

class STM32Controller:
    """负责与 STM32 下位机进行定长帧通信的控制器"""
    HEAD1 = 0xAA
    HEAD2 = 0x55
    TAIL = b'\r\n'

    def __init__(self, port: str, baudrate: int = 115200, timeout: float = 0.5):
        self.port = port
        self.baudrate = baudrate
        self.timeout = timeout
        self.ser: Optional[serial.Serial] = None

    def connect(self) -> bool:
        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
            )
            # 开启串口时清空输入输出缓冲区残留字节
            self.ser.reset_input_buffer()
            self.ser.reset_output_buffer()
            time.sleep(0.05)
            return True
        except serial.SerialException as e:
            print(f"[错误] 打开串口 {self.port} 失败: {e}")
            return False

    def close(self):
        if self.ser and self.ser.is_open:
            self.ser.close()

    def send_command(self, cmd: int, payload: bytes = b'', max_retries: int = 2) -> Optional[bytes]:
        """封包、发送指令并等待合法应答帧，超时自动重试"""
        if not self.ser or not self.ser.is_open:
            return None

        # 构造数据帧：HEAD1(1) + HEAD2(1) + CMD(1) + LEN(1) + DATA(N) + CHKSUM(1) + TAIL(2)
        length = len(payload)
        chksum = (cmd + length + sum(payload)) & 0xFF
        frame = bytes([self.HEAD1, self.HEAD2, cmd, length]) + payload + bytes([chksum]) + self.TAIL

        for attempt in range(max_retries + 1):
            try:
                self.ser.reset_input_buffer()
                self.ser.write(frame)
                self.ser.flush()

                # 读取应答：先以 \r\n 结尾读取
                resp = self.ser.read_until(self.TAIL)
                if not resp or len(resp) < 7:
                    if attempt < max_retries:
                        time.sleep(0.05)
                        continue
                    return None

                # 验证帧头与帧尾
                if resp[0] != self.HEAD1 or resp[1] != self.HEAD2:
                    continue
                if not resp.endswith(self.TAIL):
                    continue

                resp_cmd = resp[2]
                resp_len = resp[3]
                if len(resp) != 4 + resp_len + 1 + 2:
                    continue

                resp_payload = resp[4:4 + resp_len]
                resp_chksum = resp[4 + resp_len]

                # 校验和比对
                calc_chk = (resp_cmd + resp_len + sum(resp_payload)) & 0xFF
                if calc_chk == resp_chksum and resp_cmd == cmd:
                    return resp_payload

            except serial.SerialException:
                pass
            time.sleep(0.05)

        return None

    def ping(self) -> bool:
        resp = self.send_command(cmd=0x01, payload=b'')
        return resp is not None and len(resp) >= 1 and resp[0] == 0x00

    def set_relay(self, enable: bool) -> bool:
        val = 0x01 if enable else 0x00
        resp = self.send_command(cmd=0x02, payload=bytes([val]))
        return resp is not None and len(resp) >= 1 and resp[0] == val

    def read_voltage_mv(self, channel: int = 0) -> Optional[int]:
        resp = self.send_command(cmd=0x03, payload=bytes([channel]))
        if resp and len(resp) >= 2:
            return (resp[0] << 8) | resp[1]
        return None


class TestSuiteRunner:
    """自动化测试用例运行调度与判定引擎"""
    def __init__(self, controller: STM32Controller):
        self.ctrl = controller
        self.results: List[Dict[str, str]] = []

    def log_result(self, item_name: str, measured: str, standard: str, passed: bool):
        verdict = "PASS" if passed else "FAIL"
        self.results.append({
            "name": item_name,
            "measured": measured,
            "standard": standard,
            "verdict": verdict
        })
        print(f"[{verdict:4s}] {item_name}: 测量值={measured}, 标准范围={standard}")

    def run_all(self) -> bool:
        print("=" * 60)
        print("开始执行嵌入式自动化测试流程...")
        print("=" * 60)

        # 步骤 1：下位机通信在线探测
        is_online = self.ctrl.ping()
        self.log_result("01. 下位机在线探活 (PING)", "在线" if is_online else "无响应", "必须在线", is_online)
        if not is_online:
            print("[中断] 下位机通信失败，终止后续硬件测试。")
            return False

        # 步骤 2：待测板供电控制测试
        pwr_on = self.ctrl.set_relay(True)
        self.log_result("02. 继电器上电控制 (POWER ON)", "开启成功" if pwr_on else "失败", "成功开启", pwr_on)
        if not pwr_on:
            print("[中断] 待测板供电未建立，终止后续测试。")
            return False

        # 延时等待待测板各电源轨电容充饱、系统稳定
        time.sleep(0.3)

        # 步骤 3：3.3V 关键测试点模拟电压检测
        mv = self.ctrl.read_voltage_mv(channel=0)
        if mv is not None:
            # 允许 3.3V 稳压芯片波动范围：3150mV ~ 3450mV (±4.5%)
            v_ok = (3150 <= mv <= 3450)
            self.log_result("03. 待测板3.3V导轨电压测量", f"{mv} mV", "3150 ~ 3450 mV", v_ok)
        else:
            self.log_result("03. 待测板3.3V导轨电压测量", "读取超时", "3150 ~ 3450 mV", False)

        # 步骤 4：测试完成后断电保护
        pwr_off = self.ctrl.set_relay(False)
        self.log_result("04. 继电器断电下电 (POWER OFF)", "关闭成功" if pwr_off else "失败", "成功关闭", pwr_off)

        # 打印总结报告
        print("=" * 60)
        print("自动化测试汇总报告:")
        total = len(self.results)
        passed_cnt = sum(1 for r in self.results if r["verdict"] == "PASS")
        all_passed = (total == passed_cnt)
        print(f"总测试项: {total}, 通过: {passed_cnt}, 失败: {total - passed_cnt}")
        print(f"最终判定: {'【ALL PASS】合格' if all_passed else '【FAIL】不合格'}")
        print("=" * 60)
        return all_passed

if __name__ == "__main__":
    # 执行前需根据操作系统修改串口号：
    # Windows: "COM3" / Linux: "/dev/ttyUSB0" / macOS: "/dev/tty.usbserial-110"
    TARGET_PORT = "/dev/ttyUSB0"
    mcu = STM32Controller(port=TARGET_PORT, baudrate=115200, timeout=0.5)

    if mcu.connect():
        runner = TestSuiteRunner(mcu)
        runner.run_all()
        mcu.close()

四、联调验证与测试运行示例

在开展联调测试前，按照以下接线表完成硬件连接：

4.1 执行自动化测试脚本

将待测板放置在测试治具针床上，在上位机终端执行运行命令：

python test_runner.py

4.2 实际控制台输出示例

当硬件接线良好、待测板各指标正常时，控制台呈现如下标准执行结果输出：

============================================================
开始执行嵌入式自动化测试流程...
============================================================
[PASS] 01. 下位机在线探活 (PING): 测量值=在线, 标准范围=必须在线
[PASS] 02. 继电器上电控制 (POWER ON): 测量值=开启成功, 标准范围=成功开启
[PASS] 03. 待测板3.3V导轨电压测量: 测量值=3312 mV, 标准范围=3150 ~ 3450 mV
[PASS] 04. 继电器断电下电 (POWER OFF): 测量值=关闭成功, 标准范围=成功关闭
============================================================
自动化测试汇总报告:
总测试项: 4, 通过: 4, 失败: 0
最终判定: 【ALL PASS】合格
============================================================

五、常见通信与同步问题排查

在实际搭建上位机测试系统过程中，初学者常会遇到通信中断或测量不稳定的情况。以下针对三种高频故障给出定位与解决方法：

5.1 串口频繁超时且无应答

• 排查地线连接：PC、USB-TTL模块、STM32核心板与DUT如果未接共地线（GND），两者信号电平参考点不一致，会导致逻辑电平识别混乱。务必使用短粗导线牢固连接各模块GND。
• 排查TX/RX交叉接线：检查物理接线是否为发送端接接收端（即 TX 接 RX，RX 接 TX）。
• 排查波特率晶振误差：若使用STM32内部RC振荡器（HSI）作为系统时钟源，在环境温差变化较大时波特率漂移可能超过3%，导致采样错误。建议使用外部高精石英晶振（HSE）并重新核准系统分频配置。

5.2 校验和不匹配或频繁丢帧

• 校验和溢出处理差异：Python的 sum() 返回无符号整数，必须使用 & 0xFF 进行单字节截断，才能与C语言中 uint8_t 的自然溢出行为保持一致。
• 串口接收超时时间偏短：若STM32在执行ADC连续采样或继电器防抖延时较长，上位机 timeout 设置过小（例如小于50ms）会在下位机尚未回应前直接判断超时并重发，造成帧序列错位。建议根据下位机最长单步执行时间将超时设定在 0.3s ~ 0.5s。

5.3 继电器动作瞬间导致下位机复位