根据你的代码和上位机需求，我将提供**精准、对应、可直接使用的修改方案**，分为**下位机（STM32）优化**和**上位机（Python）处理**两部分。 ## 一、下位机代码优化（STM32端） ### 1. **ADC采样优化** - 增加硬件滤波和多次采样平均 修改 `ADC.C` 文件： ```c // ADC.C - 修改后的版本 #include "adc.h" #include "stm32f10x.h" #include "math.h" #define SAMPLE_TIMES 8 // 采样次数，用于平均滤波 void ADC1_Init(void) { GPIO_InitTypeDef GPIO_InitStructure; ADC_InitTypeDef ADC_InitStructure; RCC_APB2PeriphClockCmd(RCC_APB2Periph_GPIOA | RCC_APB2Periph_ADC1, ENABLE); RCC_ADCCLKConfig(RCC_PCLK2_Div6); // PA1 通道1 (注意：ADC_Channel_1对应PA1) GPIO_InitStructure.GPIO_Pin = GPIO_Pin_1; GPIO_InitStructure.GPIO_Mode = GPIO_Mode_AIN; GPIO_Init(GPIOA, &GPIO_InitStructure); // PA4 通道4 GPIO_InitStructure.GPIO_Pin = GPIO_Pin_4; GPIO_InitStructure.GPIO_Mode = GPIO_Mode_AIN; GPIO_Init(GPIOA, &GPIO_InitStructure); ADC_InitStructure.ADC_Mode = ADC_Mode_Independent; ADC_InitStructure.ADC_ScanConvMode = DISABLE; ADC_InitStructure.ADC_ContinuousConvMode = DISABLE; ADC_InitStructure.ADC_ExternalTrigConv = ADC_ExternalTrigConv_None; ADC_InitStructure.ADC_DataAlign = ADC_DataAlign_Right; ADC_InitStructure.ADC_NbrOfChannel = 1; ADC_Init(ADC1, &ADC_InitStructure); ADC_Cmd(ADC1, ENABLE); ADC_ResetCalibration(ADC1); while(ADC_GetResetCalibrationStatus(ADC1)); ADC_StartCalibration(ADC1); while(ADC_GetCalibrationStatus(ADC1)); } // 改进的读取函数：多次采样 + 去极值平均法 static u16 ADC_ReadChannel(u8 channel) { u32 sum = 0; u16 samples[SAMPLE_TIMES]; u16 min, max; u8 i; // 采样SAMPLE_TIMES次 for(i = 0; i < SAMPLE_TIMES; i++) { ADC_RegularChannelConfig(ADC1, channel, 1, ADC_SampleTime_239Cycles5); ADC_SoftwareStartConvCmd(ADC1, ENABLE); while(!ADC_GetFlagStatus(ADC1, ADC_FLAG_EOC)); samples[i] = ADC_GetConversionValue(ADC1); Delay_us(10); // 小延时避免连续采样干扰 } // 找最小最大值（去极值） min = samples[0]; max = samples[0]; for(i = 1; i < SAMPLE_TIMES; i++) { if(samples[i] < min) min = samples[i]; if(samples[i] > max) max = samples[i]; } // 去掉最小最大值后求平均 for(i = 0; i < SAMPLE_TIMES; i++) { if(samples[i] != min && samples[i] != max) { sum += samples[i]; } } return sum / (SAMPLE_TIMES - 2); } // 通道1：PA4 = ADC_Channel_4 u8 Get_FSR1_Pressure(void) { u16 adc = ADC_ReadChannel(ADC_Channel_4); // 限幅处理，防止异常值 if(adc > 4095) adc = 4095; return (adc * 100) / 4095; } // 通道2：PA1 = ADC_Channel_1 u8 Get_FSR2_Pressure(void) { u16 adc = ADC_ReadChannel(ADC_Channel_1); // 限幅处理，防止异常值 if(adc > 4095) adc = 4095; return (adc * 100) / 4095; } ``` ### 2. **MAIN.C 优化** - 增加数据发送频率控制 ```c // MAIN.C - 修改后的版本 #include "stm32f10x.h" #include "adc.h" #include "usart.h" #include "delay.h" // 数据缓冲区，用于平滑处理 #define BUFFER_SIZE 5 u8 fsr1_buffer[BUFFER_SIZE] = {0}; u8 fsr2_buffer[BUFFER_SIZE] = {0}; u8 buffer_index = 0; // 移动平均滤波函数 u8 Moving_Average(u8 buffer[], u8 new_value) { buffer[buffer_index] = new_value; buffer_index = (buffer_index + 1) % BUFFER_SIZE; u32 sum = 0; for(u8 i = 0; i < BUFFER_SIZE; i++) { sum += buffer[i]; } return sum / BUFFER_SIZE; } int main(void) { u8 fsr1_raw, fsr2_raw; u8 fsr1_smooth, fsr2_smooth; Delay_Init(); ADC1_Init(); USART1_Init(9600); // 初始化缓冲区 for(u8 i = 0; i < BUFFER_SIZE; i++) { fsr1_raw = Get_FSR1_Pressure(); fsr2_raw = Get_FSR2_Pressure(); fsr1_buffer[i] = fsr1_raw; fsr2_buffer[i] = fsr2_raw; Delay_ms(50); } while(1) { // 读取原始数据 fsr1_raw = Get_FSR1_Pressure(); fsr2_raw = Get_FSR2_Pressure(); // 移动平均滤波 fsr1_smooth = Moving_Average(fsr1_buffer, fsr1_raw); fsr2_smooth = Moving_Average(fsr2_buffer, fsr2_raw); // 发送平滑后的数据给上位机 Usart1_SendTwoNum(fsr1_smooth, fsr2_smooth); // 调整采样间隔（根据需求调整） Delay_ms(100); // 100ms采样间隔，10Hz采样率 } } ``` ## 二、上位机Python代码优化（关键！） ### 1. **改进的串口读取和数据处理类** ```python # serial_processor.py import serial import numpy as np from collections import deque from scipy.signal import savgol_filter import threading import time class SmoothPressureProcessor: def __init__(self, port='COM3', baudrate=9600, buffer_size=50): """ 初始化平滑压力处理器 :param port: 串口端口 :param baudrate: 波特率 :param buffer_size: 数据缓冲区大小 """ self.serial_port = None self.port = port self.baudrate = baudrate self.buffer_size = buffer_size # 数据缓冲区 self.ch1_raw_buffer = deque(maxlen=buffer_size) self.ch2_raw_buffer = deque(maxlen=buffer_size) self.ch1_smooth_buffer = deque(maxlen=buffer_size) self.ch2_smooth_buffer = deque(maxlen=buffer_size) # 时间戳 self.time_buffer = deque(maxlen=buffer_size) self.start_time = time.time() # 线程控制 self.running = False self.thread = None # 平滑参数 self.savgol_window = 15 # Savitzky-Golay窗口大小 self.savgol_order = 3 # 多项式阶数 # 异常值过滤阈值 self.max_jump = 30 # 相邻点最大允许跳变值 def connect(self): """连接串口""" try: self.serial_port = serial.Serial( port=self.port, baudrate=self.baudrate, timeout=0.1 # 缩短超时时间 ) print(f"已连接到 {self.port}") return True except Exception as e: print(f"连接失败: {e}") return False def disconnect(self): """断开连接""" self.stop() if self.serial_port and self.serial_port.is_open: self.serial_port.close() def _read_serial_data(self): """读取串口数据的内部方法""" if not self.serial_port or not self.serial_port.is_open: return None, None try: line = self.serial_port.readline().decode('ascii', errors='ignore').strip() if line: parts = line.split(',') if len(parts) == 2: ch1 = int(parts[0]) ch2 = int(parts[1]) # 有效性检查 if 0 <= ch1 <= 100 and 0 <= ch2 <= 100: return ch1, ch2 except: pass return None, None def _smooth_data(self, data_buffer): """使用Savitzky-Golay滤波器平滑数据""" if len(data_buffer) < self.savgol_window: # 数据不足时使用移动平均 return np.mean(list(data_buffer)) if data_buffer else 0 # 转换为数组并应用滤波 data_array = np.array(list(data_buffer)) # 使用Savitzky-Golay滤波 smoothed = savgol_filter( data_array, window_length=self.savgol_window, polyorder=self.savgol_order, mode='nearest' # 边界处理模式 ) return smoothed[-1] # 返回最新的平滑值 def _remove_outliers(self, new_val, last_val): """去除异常值""" if last_val is None: return new_val # 如果跳变过大，认为是异常值，返回上一个值 if abs(new_val - last_val) > self.max_jump: return last_val return new_val def start(self): """启动数据采集线程""" if self.running: return self.running = True self.thread = threading.Thread(target=self._update_loop) self.thread.daemon = True self.thread.start() def stop(self): """停止数据采集""" self.running = False if self.thread: self.thread.join(timeout=1) def _update_loop(self): """数据更新循环""" last_ch1 = None last_ch2 = None while self.running: # 读取数据 ch1_raw, ch2_raw = self._read_serial_data() if ch1_raw is not None and ch2_raw is not None: # 去除异常值 ch1_filtered = self._remove_outliers(ch1_raw, last_ch1) ch2_filtered = self._remove_outliers(ch2_raw, last_ch2) # 更新原始缓冲区 self.ch1_raw_buffer.append(ch1_filtered) self.ch2_raw_buffer.append(ch2_filtered) # 更新时间戳 current_time = time.time() - self.start_time self.time_buffer.append(current_time) # 平滑处理 if len(self.ch1_raw_buffer) >= 5: # 至少有5个点才开始平滑 ch1_smooth = self._smooth_data(self.ch1_raw_buffer) ch2_smooth = self._smooth_data(self.ch2_raw_buffer) self.ch1_smooth_buffer.append(ch1_smooth) self.ch2_smooth_buffer.append(ch2_smooth) last_ch1 = ch1_filtered last_ch2 = ch2_filtered time.sleep(0.01) # 10ms循环间隔 def get_latest_data(self): """获取最新数据""" if not self.ch1_smooth_buffer or not self.ch2_smooth_buffer: return 0, 0, 0, 0 # 返回原始值和平滑值 ch1_raw = self.ch1_raw_buffer[-1] if self.ch1_raw_buffer else 0 ch2_raw = self.ch2_raw_buffer[-1] if self.ch2_raw_buffer else 0 ch1_smooth = self.ch1_smooth_buffer[-1] if self.ch1_smooth_buffer else ch1_raw ch2_smooth = self.ch2_smooth_buffer[-1] if self.ch2_smooth_buffer else ch2_raw return ch1_raw, ch2_raw, ch1_smooth, ch2_smooth def get_all_data(self): """获取所有缓冲数据""" times = list(self.time_buffer) ch1_raw = list(self.ch1_raw_buffer) ch2_raw = list(self.ch2_raw_buffer) ch1_smooth = list(self.ch1_smooth_buffer) ch2_smooth = list(self.ch2_smooth_buffer) return times, ch1_raw, ch2_raw, ch1_smooth, ch2_smooth ``` ### 2. **改进的实时绘图代码** ```python # pressure_plotter.py import matplotlib.pyplot as plt import matplotlib.animation as animation from serial_processor import SmoothPressureProcessor import numpy as np class RealTimePressurePlotter: def __init__(self, port='COM3'): self.processor = SmoothPressureProcessor(port=port) self.fig, self.axes = plt.subplots(2, 2, figsize=(12, 8)) self.setup_plots() # 数据存储 self.max_points = 200 # 显示的最大点数 def setup_plots(self): """设置子图""" # 通道1原始vs平滑 self.ax1 = self.axes[0, 0] self.ax1.set_title('通道1 - 原始 vs 平滑') self.ax1.set_xlabel('时间 (s)') self.ax1.set_ylabel('压力值') self.ax1.set_ylim(0, 100) self.ax1.grid(True, alpha=0.3) self.line1_raw, = self.ax1.plot([], [], 'r-', alpha=0.5, linewidth=1, label='原始') self.line1_smooth, = self.ax1.plot([], [], 'b-', linewidth=2, label='平滑') self.ax1.legend() # 通道2原始vs平滑 self.ax2 = self.axes[0, 1] self.ax2.set_title('通道2 - 原始 vs 平滑') self.ax2.set_xlabel('时间 (s)') self.ax2.set_ylabel('压力值') self.ax2.set_ylim(0, 100) self.ax2.grid(True, alpha=0.3) self.line2_raw, = self.ax2.plot([], [], 'g-', alpha=0.5, linewidth=1, label='原始') self.line2_smooth, = self.ax2.plot([], [], 'm-', linewidth=2, label='平滑') self.ax2.legend() # 双通道对比（平滑后） self.ax3 = self.axes[1, 0] self.ax3.set_title('双通道对比（平滑）') self.ax3.set_xlabel('时间 (s)') self.ax3.set_ylabel('压力值') self.ax3.set_ylim(0, 100) self.ax3.grid(True, alpha=0.3) self.line3_ch1, = self.ax3.plot([], [], 'b-', linewidth=2, label='通道1') self.line3_ch2, = self.ax3.plot([], [], 'm-', linewidth=2, label='通道2') self.ax3.legend() # 数值显示 self.ax4 = self.axes[1, 1] self.ax4.axis('off') self.text_display = self.ax4.text(0.1, 0.5, '', fontsize=12, verticalalignment='center', bbox=dict(boxstyle='round', facecolor='wheat', alpha=0.5)) plt.tight_layout() def update_plot(self, frame): """更新绘图""" # 获取数据 times, ch1_raw, ch2_raw, ch1_smooth, ch2_smooth = self.processor.get_all_data() if len(times) > 0: # 只显示最近的数据 start_idx = max(0, len(times) - self.max_points) times_display = times[start_idx:] # 更新通道1 if len(ch1_raw) > 0: ch1_raw_display = ch1_raw[start_idx:] self.line1_raw.set_data(times_display, ch1_raw_display) if len(ch1_smooth) > 0: ch1_smooth_display = ch1_smooth[start_idx:] self