Python读取STM32传感器数据并保存为CSV文件

在嵌入式开发与数据监测场景中，将STM32单片机采集的传感器数据传输到上位机并归档，是实现数据可视化与离线分析的基本链路。最稳定且通用的途径是：STM32将传感器数据格式化为文本行，经串口（USART/UART）通过USB转TTL模块发送到计算机；计算机端运行Python脚本，利用pyserial库持续监听串口数据流，完成字符串校验与字段拆分，最后通过Python内置的csv模块将包含时间戳的结构化数据追加写入本地CSV文件中。

系统整体硬件拓扑与数据流动

整套采集链路分为物理层信号传输与逻辑层数据解析两个阶段。传感器（例如温湿度、气压、陀螺仪等模块）通过I2C、SPI或ADC总线连接至STM32主控芯片。STM32完成底层模数转换与校准计算后，通过硬件串口以固定波特率向外输出。计算机通过CH340、CP2102或FT232等USB转TTL芯片接入该信号，在操作系统层面映射为虚拟串口（COM端口或ttyUSB设备）。

图1：STM32至PC的数据采集与物理通信拓扑

在物理接线时，有三个关键点必须核实：

• 交叉连接：STM32的TXD（发送端）必须连接USB转TTL模块的RXD（接收端）；STM32的RXD连接模块的TXD。
• 共地连接（GND）：STM32开发板的地线必须与USB转TTL模块的GND可靠连接，否则参考电平不一致会导致接收端出现大面积乱码或丢帧。
• 逻辑电平兼容：多数STM32芯片引脚电平为3.3V，应确保转串口模块跳线帽切换至3.3V挡位，避免5V直接输入导致引脚过压受损。

通信协议设计与STM32输出规范

为了让上位机稳定解析，下位机发送的字符流必须具备明确的分隔符和帧边界。最常见且鲁棒性较好的格式是纯文本定界符协议（ASCII Comma-Separated Values），每条有效数据以回车换行符\r\n结束。

例如，若STM32采集温度（24.5℃）、湿度（60.2%）与光照度（350 lx），其单次串口发送的字符串格式应设计为：

24.50,60.20,350\r\n

在STM32工程代码中（以HAL库为例），可通过重定向printf或直接调用HAL_UART_Transmit定时向外发送数据：

/* STM32端示例逻辑：每隔1秒采集并向串口打印一行数据 */
char tx_buffer[64];
float temperature = 24.5f;
float humidity = 60.2f;
int light_level = 350;

snprintf(tx_buffer, sizeof(tx_buffer), "%.2f,%.2f,%d\r\n", temperature, humidity, light_level);
HAL_UART_Transmit(&huart1, (uint8_t*)tx_buffer, strlen(tx_buffer), 100);
HAL_Delay(1000);

Python端完整实现步骤与逻辑

在上位机端，整个Python程序的生命周期可以拆解为配置初始化、串口开启、行监听、格式校验与实时写入五大环节。

图2：从串口监听、校验清洗到CSV持久化各步骤逻辑

1. 依赖库安装

Python标准库自带csv、time和datetime模块，唯一需要安装的第三方依赖是pyserial：

pip install pyserial

2. 编写Python完整采集脚本

以下脚本具备自动创建表头、时间戳附加、脏数据容错过滤以及中断安全退出的完整逻辑：

import os
import csv
import time
from datetime import datetime
import serial

# --- 参数配置 ---
SERIAL_PORT = "COM3"       # Windows如 'COM3', Linux/macOS如 '/dev/ttyUSB0'
BAUD_RATE = 115200         # 必须与STM32串口波特率一致
OUTPUT_CSV_FILE = "sensor_data.csv"
EXPECTED_FIELDS_COUNT = 3  # 预期字段数：温度, 湿度, 光照

def init_csv_file(filepath):
    """初始化CSV文件，如果文件不存在则写入表头"""
    file_exists = os.path.exists(filepath)
    if not file_exists:
        with open(filepath, mode="w", newline="", encoding="utf-8") as f:
            writer = csv.writer(f)
            writer.writerow(["Timestamp", "Temperature_C", "Humidity_Pct", "Light_Lux"])
            print(f"[系统] 创建新数据文件并写入表头: {filepath}")

def main():
    init_csv_file(OUTPUT_CSV_FILE)
    
    print(f"[系统] 正在尝试连接串口: {SERIAL_PORT} (波特率: {BAUD_RATE})...")
    try:
        ser = serial.Serial(
            port=SERIAL_PORT,
            baudrate=BAUD_RATE,
            bytesize=serial.EIGHTBITS,
            parity=serial.PARITY_NONE,
            stopbits=serial.STOPBITS_ONE,
            timeout=2.0  # 读取超时时间，防止无数据时程序无限挂起
        )
    except serial.SerialException as e:
        print(f"[错误] 无法打开串口 {SERIAL_PORT}: {e}")
        return

    # 首次打开串口时，清空历史残余缓冲区数据
    ser.reset_input_buffer()
    time.sleep(1.0)
    print(f"[系统] 串口连接成功，开始持续监听并写入 {OUTPUT_CSV_FILE}（按 Ctrl+C 停止）...")

    # 以追加模式持续保持文件打开
    try:
        with open(OUTPUT_CSV_FILE, mode="a", newline="", encoding="utf-8") as csv_file:
            csv_writer = csv.writer(csv_file)
            
            while True:
                # 阻塞读取直到遇到换行符 \n 或超时
                raw_bytes = ser.readline()
                if not raw_bytes:
                    continue

                try:
                    line_str = raw_bytes.decode("utf-8", errors="ignore").strip()
                except Exception:
                    continue

                if not line_str:
                    continue

                # 拆分逗号字段
                fields = line_str.split(",")
                if len(fields) != EXPECTED_FIELDS_COUNT:
                    print(f"[警告] 丢弃异常格式帧: {line_str}")
                    continue

                # 验证字段数值合法性
                try:
                    temp_val = float(fields[0])
                    humi_val = float(fields[1])
                    light_val = int(fields[2])
                except ValueError:
                    print(f"[警告] 数据转换失败，包含非法字符: {line_str}")
                    continue

                # 获取当前系统毫秒级时间戳
                current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")[:-3]
                
                # 写入CSV记录
                row_data = [current_time, temp_val, humi_val, light_val]
                csv_writer.writerow(row_data)
                
                # 关键：强制刷新缓冲区，确保实时持久化，防止突然断电或程序崩溃丢失数据
                csv_file.flush()
                
                print(f"[记录] {current_time} -> 温度:{temp_val}℃, 湿度:{humi_val}%, 光照:{light_val}lx")

    except KeyboardInterrupt:
        print("\n[系统] 接收到退出信号，正在安全关闭...")
    finally:
        if ser.is_open:
            ser.close()
            print("[系统] 串口已安全释放。")

if __name__ == "__main__":
    main()

数据拆分、校验与行记录组装解析

在传感器数据采集过程中，由于上位机刚刚连接串口时STM32可能正在发送某个字符的中间阶段，第一帧收到的原始数据常常是半截断的残帧。此外，外界电磁干扰偶尔也会导致串口比特错位。因此，解析模块必须具备健壮的容错机制：

图3：字符串拆分、类型转换与时间戳注入装配为CSV行数据

在数据处理阶段，程序依次执行三道防线：

1. 解码过滤：使用decode("utf-8", errors="ignore")防止因单字节噪点导致整个程序崩溃抛出UnicodeDecodeError。
2. 元素数量校验：通过len(fields) == EXPECTED_FIELDS_COUNT快速剔除开头由于截断产生的半行残缺文本。
3. 显式类型转换：对解析出的字段强制进行float()或int()类型转换，不仅保证存入数据的合规性，还能有效滤除含乱码字母的损坏行。

常见问题排查与运行验证

在实际联调过程中，常见异常现象与定位解决方式如下表所示：

总结与长期稳定采集的最佳实践

通过PySerial与CSV模块的组合，可以以极轻量的方式构建一套工业级传感器数据采集持久化工具。为了确保采集系统能连续数天稳定运行，有以下总结性建议：

图4：长期稳定采集与工程落地最佳实践要点总结

在构建传感器数据流时，优先采用清晰的文本逗号定界协议，并在Python端辅以显式的行缓冲刷新与异常捕获，是嵌入式设备与计算机之间进行可靠数据传输和保存的最佳起点。