使用'embedded python block'创建缓冲区

Question

我正试图在GNU Radio中创建一个程序，将信号中的数据存储到Filesink中。但是，通过这样做，文件链接中的大多数数据将由噪声组成。我的目标是只存储主信号和同一信号前后的一点噪声。

为了实现这一点，我试图使用"嵌入式python块"构建一个缓冲区，该缓冲区只存储主信号，如上所述，还存储该信号前后的一点点噪声(存储的噪声大小由缓冲区的大小决定(。

为此，我创建了这个GNU无线电程序GNU无线电计划，在最后我将一个文件链接连接到静噪块，并将一个数据链接连接到缓冲块，这样我就可以使用频谱分析仪比较这两个文件链接。

(注意：我用发射器和arduino发射信号。arduino被编程为每秒发送一条"Hello World"消息。然后我用SDR Pluto捕获该信号/消息(。

主要的问题是，在我运行程序后，无论程序运行多长时间，静噪文件链接都只有64KB。另一方面，随着时间的推移，缓冲区文件链接将存储数据，但是，当我在频谱分析仪上检查缓冲区文件链时，频谱是空的。

如果有人能帮助我了解问题所在，我将不胜感激。这是我的缓冲代码：

import numpy as np
from gnuradio import gr

class blk(gr.basic_block):  # other base classes are basic_block, decim_block, interp_block
def __init__(self, buffer_size_param=10000.0):  # only default arguments here
"""arguments to this function show up as parameters in GRC"""
gr.basic_block.__init__(
self,
name='Embedded Python Block (Buffer)',  # will show up in GRC
in_sig=[np.complex64, np.complex64],
out_sig=[np.complex64]
)

self.buffer = [] 
self.buffer_size_param = buffer_size_param
self.counter = self.buffer_size_param

def general_work(self, input_items, output_items):
squelch = input_items[0]
throttle = input_items[1]

for i in range(len(squelch)):       

if (squelch [i] == 0):
self.counter = self.counter + 1
if (self.counter <= buffer_size_param):
output_items.append(throttle[i])
else:
self.buffer.append(throttle [i])

if len (self.buffer) > buffer_size_param:
self.buffer.pop (0)

else:

output_items = output_items + self.buffer
output_items.append(throttle[i])

self.buffer = []
self.counter = 0

return len(output_items)

Answer 1

在这里，我用numpy实现了一个循环缓冲区，它运行得很好。您可以使用Numpy生成缓冲区，这对于纯python编写的代码来说是非常快的。如果你看到任何错误，请告诉我https://github.com/OkanErturk16/Misc/blob/master/circularBuffer.py