Python 中的实时移动平均线

Question

我需要使用 Python 计算串行端口上出现的传感器数据的移动平均值。在程序启动之前，我能找到的所有关于 numpy 的示例都使用文件中的数据或数组中的硬编码数据。

就我而言，程序启动时我没有任何数据。数据每一秒都是实时的。我想在数据到达串行端口时对其进行平滑处理。

我在 Arduino 上有这个工作，但在 Python 中也需要它。有人可以请我指出实时（随时间变化的单个值）样本而不是批量样本。

Answer 1

作为一个类实现：

class StreamingMovingAverage:
    def __init__(self, window_size):
        self.window_size = window_size
        self.values = []
        self.sum = 0

    def process(self, value):
        self.values.append(value)
        self.sum += value
        if len(self.values) > self.window_size:
            self.sum -= self.values.pop(0)
        return float(self.sum) / len(self.values)

Answer 2

下面介绍了如何一次将一个读数添加到正在运行的读数集合中并返回平均值。我预先填充了阅读列表以显示它的实际效果，但在您的程序中，您只需从一个空列表开始：readings = []

我假设您希望在平均值中包含最后的 x 个读数，而不是包含所有读数。这就是max_samples参数的用途。

没有麻木：

readings = [1, 2, 3, 4, 5, 6, 7, 8, 9]
reading = 10
max_samples = 10

def mean(nums):
    return float(sum(nums)) / max(len(nums), 1)

readings.append(reading)
avg = mean(readings)

print 'current average =', avg
print 'readings used for average:', readings

if len(readings) == max_samples:
    readings.pop(0)

print 'readings saved for next time:', readings

结果：

current average = 5.5
readings used for average: [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
readings saved for next time: [2, 3, 4, 5, 6, 7, 8, 9, 10]

与麻木：

import numpy as np

readings = np.array([1, 2, 3, 4, 5, 6, 7, 8, 9])
reading = 10
max_samples = 10

readings = np.append(readings, reading)
avg = np.mean(readings)

print 'current average =', avg
print 'readings used for average:', readings

if len(readings) == max_samples:
    readings = np.delete(readings, 0)

print 'readings saved for next time:', readings

结果：

current average = 5.5
readings used for average: [ 1  2  3  4  5  6  7  8  9 10]
readings saved for next time: [ 2  3  4  5  6  7  8  9 10]