实现带超时重置的滑动3值窗口均值计算问题求助

核心需求回顾 #

• 滑动窗口大小为3，计算窗口内数值的均值
• 处理到当前数组倒数第三位时，保存最后2个值作为续接缓存
• 若新数组在2秒内到达，将缓存的2个值与新数组拼接后继续滑动计算
• 超时则清空缓存，重新开始计算
• 已实现绘图功能，但原时间校验逻辑导致窗口冻结

问题根源 #

原代码大概率用了阻塞式的等待（比如time.sleep()），直接阻塞了GUI主线程，导致窗口无响应。必须用非阻塞式的定时器或者异步/多线程来处理超时逻辑，避免占用绘图线程。

修复后的代码示例（以Python + Matplotlib为例） #

假设原无时间校验的可运行示例基于Matplotlib实时绘图，以下加入非阻塞超时逻辑：

import numpy as np
import matplotlib.pyplot as plt
from matplotlib.animation import FuncAnimation
import time
from threading import Timer

# 全局变量：续接缓存、超时定时器、最后数据接收时间
cache = []
timeout_timer = None
last_receive_time = time.time()

def calculate_sliding_mean(data, window_size=3):
    global cache
    # 拼接缓存与新数据
    combined_data = cache + data
    means = []
    # 计算滑动窗口均值
    for i in range(len(combined_data) - window_size + 1):
        window = combined_data[i:i+window_size]
        means.append(np.mean(window))
    # 更新缓存：处理后剩余数据>=2则保留最后2个，否则清空
    cache = combined_data[-2:] if len(combined_data) >= 2 else []
    return means

def reset_cache():
    """超时后清空缓存"""
    global cache
    cache = []
    print("超时，清空缓存")

def handle_new_data(new_data):
    global timeout_timer, last_receive_time
    # 取消之前的超时定时器
    if timeout_timer is not None:
        timeout_timer.cancel()
    # 计算当前批次均值
    current_means = calculate_sliding_mean(new_data)
    # 更新最后接收时间
    last_receive_time = time.time()
    # 启动新的2秒超时定时器
    timeout_timer = Timer(2.0, reset_cache)
    timeout_timer.start()
    return current_means

# ---------------------- 绘图部分（保留原实现逻辑）----------------------
fig, ax = plt.subplots()
line, = ax.plot([], [], 'b-')
ax.set_xlim(0, 50)
ax.set_ylim(0, 10)
x_data, y_data = [], []

def update(frame):
    # 模拟随机生成新数据（实际场景替换为真实数据接收逻辑）
    if np.random.rand() < 0.3:
        new_data = np.random.randint(1, 10, size=np.random.randint(2, 6))
        means = handle_new_data(new_data.tolist())
        y_data.extend(means)
        x_data.extend(range(len(x_data), len(x_data)+len(means)))
    # 更新绘图数据
    line.set_data(x_data, y_data)
    # 自动调整x轴显示范围
    if len(x_data) > 0:
        ax.set_xlim(max(0, len(x_data)-50), len(x_data))
    return line,

ani = FuncAnimation(fig, update, interval=100, blit=True)
plt.show()

关键逻辑说明 #

1. 非阻塞超时处理：用threading.Timer实现超时逻辑，每次收到新数据时取消旧定时器、重启新定时器，定时器到期自动清空缓存，全程不阻塞GUI主线程。
2. 缓存续接逻辑：每次计算前拼接缓存与新数据，计算完成后保留最后2个值作为下次续接的基础。
3. 窗口响应保障：绘图用FuncAnimation的非阻塞更新，数据处理逻辑放在定时器回调中，不占用绘图主线程。

注意事项 #

• 如果使用PyQt、Tkinter等其他GUI框架，需替换为框架自带的定时器（如QTimer、tkinter.Timer），避免跨线程操作GUI控件。
• 真实场景中，数据接收（串口、网络等）逻辑也应放在单独线程，进一步避免阻塞主线程。

内容的提问来源于stack exchange，提问作者tenebris in kux