如何过滤数据流 [英] How to filter a stream of data

问题描述

我正在尝试为Raspi上的传感器读数实现过滤器.我从要使用过滤器代码进行过滤的传感器中获得了恒定的数据流.我试图通过在我的范围代码中生成随机数来模拟从传感器获得的读数.问题是由于某种原因，我看不到过滤器的任何输出，甚至不知道它是否有效.以下是用于生成随机数据流和过滤器的代码.我是在模拟传感器读数吗?(注意:我并不特别在意这些值，我只是想要一个我可以保持读取和过滤的随机数据流)

Im trying to implement a filter for sensor readings on a Raspi. I get a constant stream of data from the sensors that I want to filter using my filter code. I tried to emulate the sort of readings I would get from a sensor by generating random numbers in my rangen code. The issue is that for some reason I cannot see any output from the filter and I dont even know if it works. Below are the codes for generating a random stream of data and the filter. Am I emulating the sensor readings right?(Note: Im not particularly concerned about the values, I just want a random stream of data that I can keep reading and filter)

生成随机数据

# generate random integer values, Filename: Rangen.py
from random import seed
from random import randint
from time import sleep
# seed random number generator
seed(1)
# generate some integers
def rangen():
    value = randint(0, 100)
    return value

while True:
    data = rangen()
    sleep(0.1)

这是过滤器

from time import sleep
from statistics import mean
from Rangen import data

def AplhaTrimmedfilter(windowsize,alpha, sensordata):
    data_list = []
    Flag = True
    while Flag:
        if len(data_list)<windowsize:
            data_list.append(sensordata)
            print("data added")
            sleep(0.0010)
            break
    if len(data_list)==windowsize:
        sorted_data = data_list.sort()     
        sliced_data = sorted_data[(windowsize/2):windowsize-alpha+(alpha/2)]
        alphamean = statistics.mean(sliced_data) #sum(a) / len(a)
        data_list = []
        print(data_list)
    return alphamean

AlphaTrimmedfilter(5,2,data)

我想从生成的数据中获取5个值，对它们进行排序，切分并平均后在最后显示.我不知道我是否实现了此权限，因为控制台上的输出未显示任何内容.任何输入表示赞赏.谢谢你.

I want to take 5 values from the generated data, sort them, slice them and avergae them and display at the end. I dont understand if I implemented this right because the output on the console shows nothing. Any input is appreciated. Thank you.

正如MisterMiyagi所建议的那样，我使用了生成器来解决过滤问题，并且我使用[[item]过滤了data_data中的项目]遍历了生成器对象，但是现在我有一个奇怪的TypeError'Float'对象是不可迭代的.我该如何解决这个问题?我认为这可能是由于我传递给过滤器(主要是浮点数)的数据类型所致.以下是根据MisterMiyagi的回答修改后的过滤器的代码:

As MisterMiyagi suggested, I used generators to tackle the issue with filtering and I iterated over the generator object using [[item] for item in filtered_data] but now I have a weird TypeError 'Float' object is not iterable. How would I fix this issue? I think this might be due to the type of data I pass into the filter(which is mostly float). Here is the code for the modified filter based on MisterMiyagi's answer:

def alpha_mean(window, alpha):
    cut = alpha//2
    data = sorted(window)[cut:-cut]
    result = sum(data)/len(data)
    return result

def alphatrimmer(window_size, alpha, sensor_stream):
    window = []
    for item in sensor_stream:
        window.append(item)
        if len(window) >= window_size:
            break
    yield alpha_mean(window, alpha)

    for item in sensor_stream:
        window.pop(0)
        window.append(item)
        yield alpha_mean(window,alpha)

推荐答案

Python的原生流等效项是迭代器.您可以通过编写生成器来创建自己的迭代器.例如，我们可以将产生一个值的rangen 函数转换为产生许多值的 generator函数.

Python's native equivalent of streams are iterators. You can create your own iterator by writing a generator. For example, we can turn your rangen function producing one value into a generator function producing many values.

# stream.py
import random

random.seed(1)

def random_stream():
    """Generator producing a stream of random numbers"""
    while True:      # generators can produce values indefinitely...
        value = random.randint(0, 100)
        yield value  # ... by yield'ing them

除其他外，for语句可以使用迭代器/生成器.您可以在Python控制台上对此进行测试:

Among other things, iterators/generators can be consumed by for statements. You can test this on the Python console:

>>> from stream import random_stream
>>> data = random_stream()   # note: the generator is called using ()
>>> for idx, item in enumerate(data):
...     if idx >= 3: break
...     print(idx, ':', item)
0 : 17
1 : 72
2 : 97

---

理想情况下，消耗此类流的函数也是 生成器-而不是一次应用固定窗口，而是在流上移动窗口.不管有多长，这都可以无缝使用整个流.

---

A function that consumes such a stream ideally is also a generator - instead of applying a fixed window once, it moves a window over the stream. This allows to seamlessly consume the entire stream, no matter how long it is.

理想地，您将窗口的跟踪和计算拆分到每个窗口位置.值得注意的是，后者不是 生成器-只能在单个窗口状态下工作.

Ideally, you split the tracking of the window and the computation at each window position. Notably, the latter is not a generator - it just works on a single window state.

def alpha_mean(window, alpha):
    """Compute the mean for a given window and alpha"""
    size = len(window)
    data = sorted(window)[size//2:size - (alpha // 2)]
    return sum(data) / len(data)


def alpha_trim(window_size, alpha, sensor_stream):
    """Produce a trimmed stream based on the ``sensor_stream`` data"""
    window = []  # note: a collections.deque is more efficient for very large windows
    # fill the first window
    for item in sensor_stream:
        window.append(item)
        if len(window) >= window_size:
            break
    yield alpha_mean(window, alpha)
    # advance the window as new data comes in
    for item in sensor_stream:
        # "move" the window by discarding the oldest value
        window.pop(0)
        window.append(item)
        yield alpha_mean(window, alpha)

alpha_trim是一个生成器，它也使用一个生成器.您可以在Python控制台中再次进行测试:

The alpha_trim is a generator that also takes a generator. You can test this again in the Python console:

>>> data = alpha_trim(5, 2, rangen())
>>> for idx, item in enumerate(data):
...     if idx >= 5: break
...     print(idx, ':', item)
0 : 52.0
1 : 52.0
2 : 47.5
3 : 47.5
4 : 60.0

这篇关于如何过滤数据流的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！