# STM32串口波形数据接收的Python程序编写指南 ## 问题分析与解决方案 接收STM32发送的采样波形数据需要从**硬件连接配置**、**数据协议设计**、**Python串口通信**和**数据可视化**四个层面进行系统性设计 [ref_1]。下面通过完整的代码示例和详细解析来说明实现方法。 ## 1. 环境准备与依赖安装 首先需要安装必要的Python库： ```python # 安装必要的依赖库 pip install pyserial # 串口通信库 pip install pyqtgraph # 高性能绘图库 pip install PyQt5 # GUI框架 pip install numpy # 数值计算库 ``` ## 2. 基础串口数据接收程序 ### 2.1 简单文本数据接收 ```python import serial import time import threading class STM32SerialReceiver: def __init__(self, port='COM3', baudrate=115200, timeout=1): """ 初始化串口接收器 :param port: 串口端口，Windows为COMx，Linux为/dev/ttyUSBx :param baudrate: 波特率，需与STM32保持一致 :param timeout: 读取超时时间 """ self.ser = serial.Serial(port, baudrate, timeout=timeout) self.is_running = False self.data_buffer = [] def start_receiving(self): """开始接收数据""" self.is_running = True print(f"开始接收STM32数据，端口: {self.ser.port}") while self.is_running: try: # 读取一行数据 if self.ser.in_waiting > 0: raw_data = self.ser.readline().decode('utf-8').strip() if raw_data: self.process_data(raw_data) except serial.SerialException as e: print(f"串口错误: {e}") break except UnicodeDecodeError: print("数据解码错误，跳过该数据包") def process_data(self, data): """ 处理接收到的数据 :param data: 原始字符串数据 """ try: # 去除可能的回车换行符 cleaned_data = data.replace('\r', '').replace('\n', '') # 假设STM32发送的是逗号分隔的数值 if ',' in cleaned_data: values = [float(x) for x in cleaned_data.split(',')] self.data_buffer.extend(values) print(f"接收到数据: {values}") else: # 单数值处理 value = float(cleaned_data) self.data_buffer.append(value) print(f"接收到数据点: {value}") # 限制缓冲区大小 if len(self.data_buffer) > 1000: self.data_buffer = self.data_buffer[-500:] except ValueError as e: print(f"数据转换错误: {e}, 原始数据: {data}") def stop_receiving(self): """停止接收数据""" self.is_running = False self.ser.close() print("数据接收已停止") # 使用示例 if __name__ == "__main__": receiver = STM32SerialReceiver(port='COM3', baudrate=115200) # 在新线程中运行接收器 receive_thread = threading.Thread(target=receiver.start_receiving) receive_thread.daemon = True receive_thread.start() try: # 主程序保持运行 while True: time.sleep(1) except KeyboardInterrupt: receiver.stop_receiving() print("程序退出") ``` ## 3. 实时波形显示程序 ### 3.1 基于PyQtGraph的波形显示器 ```python import sys import serial import numpy as np from PyQt5.QtWidgets import QApplication, QMainWindow, QVBoxLayout, QWidget, QLabel, QComboBox from PyQt5.QtCore import QTimer import pyqtgraph as pg class RealTimeWaveformDisplay(QMainWindow): def __init__(self, port='COM3', baudrate=115200, buffer_size=500): super().__init__() # 串口配置 self.serial_port = port self.baudrate = baudrate self.buffer_size = buffer_size # 数据缓冲区 self.waveform_data = np.zeros(buffer_size) self.data_index = 0 # 初始化UI和串口 self.init_ui() self.init_serial() def init_serial(self): """初始化串口连接""" try: self.ser = serial.Serial( port=self.serial_port, baudrate=self.baudrate, timeout=0.1 # 短超时以实现实时更新 ) print(f"串口连接成功: {self.serial_port}") except serial.SerialException as e: print(f"串口连接失败: {e}") self.ser = None def init_ui(self): """初始化用户界面""" self.setWindowTitle("STM32波形数据实时显示") self.setGeometry(100, 100, 1200, 600) # 中央部件 central_widget = QWidget() self.setCentralWidget(central_widget) layout = QVBoxLayout(central_widget) # 状态显示 self.status_label = QLabel("等待STM32数据...") layout.addWidget(self.status_label) # 绘图部件 self.plot_widget = pg.PlotWidget() self.plot_widget.setLabel('left', '幅度', 'V') self.plot_widget.setLabel('bottom', '采样点') self.plot_widget.showGrid(x=True, y=True, alpha=0.3) layout.addWidget(self.plot_widget) # 波形曲线 self.curve = self.plot_widget.plot( pen=pg.mkPen(color='b', width=2), name='STM32波形' ) # 定时器用于实时更新 self.timer = QTimer() self.timer.timeout.connect(self.update_display) self.timer.start(50) # 20Hz刷新率 def read_serial_data(self): """从串口读取并处理数据""" if not self.ser or not self.ser.is_open: return None try: # 读取所有可用数据 while self.ser.in_waiting > 0: raw_data = self.ser.readline().decode('utf-8').strip() if raw_data: # 数据处理 - 根据STM32实际数据格式调整 processed_data = self.parse_waveform_data(raw_data) return processed_data except Exception as e: print(f"数据读取错误: {e}") return None def parse_waveform_data(self, raw_data): """ 解析波形数据 支持多种格式：单数值、逗号分隔多数值、JSON格式等 """ try: # 方法1: 单数值格式 "123.45" if raw_data.replace('.', '').replace('-', '').isdigit(): return [float(raw_data)] # 方法2: 逗号分隔格式 "1.2,3.4,5.6" elif ',' in raw_data: values = [float(x.strip()) for x in raw_data.split(',')] return values # 方法3: 支持其他自定义格式 else: # 这里可以添加更多的数据格式解析逻辑 print(f"无法解析的数据格式: {raw_data}") return None except ValueError as e: print(f"数据解析错误: {e}, 数据: {raw_data}") return None def update_display(self): """更新波形显示""" new_data = self.read_serial_data() if new_data: # 更新数据缓冲区 for value in new_data: self.waveform_data[self.data_index] = value self.data_index = (self.data_index + 1) % self.buffer_size # 更新曲线显示 self.curve.setData(self.waveform_data) # 更新状态 self.status_label.setText(f"实时显示中... 最新值: {new_data[-1]:.3f}") def closeEvent(self, event): """关闭事件处理""" if self.ser and self.ser.is_open: self.ser.close() event.accept() # 主程序入口 if __name__ == '__main__': app = QApplication(sys.argv) # 可根据实际情况修改串口参数 window = RealTimeWaveformDisplay(port='COM3', baudrate=115200) window.show() sys.exit(app.exec_()) ``` ## 4. 高级功能扩展 ### 4.1 多通道数据接收与显示 ```python import json import serial import numpy as np from collections import deque class MultiChannelReceiver: def __init__(self, port='COM3', baudrate=115200, channels=4): self.ser = serial.Serial(port, baudrate, timeout=0.1) self.channels = channels self.data_buffers = [deque(maxlen=1000) for _ in range(channels)] def parse_json_data(self, raw_data): """ 解析JSON格式的多通道数据 STM32发送格式示例: {"ch1": 1.23, "ch2": 2.34, "ch3": 3.45, "timestamp": 123456} """ try: data_dict = json.loads(raw_data) channel_data = [] for i in range(1, self.channels + 1): channel_key = f"ch{i}" if channel_key in data_dict: channel_data.append(data_dict[channel_key]) return channel_data if channel_data else None except json.JSONDecodeError: return self.parse_legacy_format(raw_data) def parse_legacy_format(self, raw_data): """解析传统格式数据""" # 这里可以添加其他自定义格式的解析逻辑 return self.parse_csv_format(raw_data) def start_realtime_processing(self): """启动实时数据处理""" while True: if self.ser.in_waiting > 0: raw_data = self.ser.readline().decode('utf-8').strip() channel_data = self.parse_json_data(raw_data) if channel_data and len(channel_data) == self.channels: for i, value in enumerate(channel_data): self.data_buffers[i].append(value) ``` ## 5. STM32端数据发送建议 为确保Python程序正确接收数据，STM32端应采用以下数据格式之一： ### 5.1 文本格式（推荐） ```c // C语言示例 - 单数值发送 printf("%.3f\n", adc_value); // 多通道数据发送 printf("%.3f,%.3f,%.3f\n", ch1_value, ch2_value, ch3_value); ``` ### 5.2 JSON格式（结构化数据） ```c // 发送结构化数据 printf("{\"ch1\":%.3f,\"ch2\":%.3f,\"timestamp\":%lu}\n", adc_value1, adc_value2, HAL_GetTick()); ``` ## 6. 调试与故障排除 ### 6.1 常见问题解决方案 | 问题现象 | 可能原因 | 解决方案 | |---------|----------|----------| | 无法连接串口 | 端口被占用或不存在 | 检查设备管理器，确认正确端口 | | 数据乱码 | 波特率不匹配 | 确保Python与STM32波特率一致 | | 数据丢失 | 缓冲区溢出 | 增加STM32发送间隔或Python读取频率 | | 程序卡死 | 串口异常断开 | 添加异常处理机制 | ### 6.2 调试代码示例 ```python def debug_serial_communication(port='COM3', baudrate=115200): """串口通信调试工具""" try: with serial.Serial(port, baudrate, timeout=1) as ser: print(f"串口 {port} 打开成功") while True: if ser.in_waiting > 0: raw_data = ser.readline() print(f"原始数据: {raw_data}") print(f"解码后: {raw_data.decode('utf-8', errors='ignore')}") except Exception as e: print(f"调试过程中出错: {e}") # 运行调试 # debug_serial_communication() ``` 通过以上完整的Python程序实现，您可以可靠地接收STM32发送的采样波形数据，并实现实时显示和数据处理功能 [ref_1]。关键是要确保数据格式的一致性，并做好异常处理以保证程序的稳定性。