如何在未来使用 UHD/GNURadio 接收有限数量的样本？答案

【问题标题】：How to receive a finite number of samples at a future time using UHD/GNURadio?如何在未来使用 UHD/GNURadio 接收有限数量的样本？
【发布时间】：2017-09-11 05:32:18
【问题描述】：

我正在使用 UHD 的 GNURadio python 接口，并且我正在尝试设置一个特定的时间来开始收集样本并收集特定数量的样本或在特定时间停止样本的收集。本质上，创建样本的定时快照。这类似于 C++ Ettus UHD 示例“rx_timed_sample”。

我可以让流程图在特定时间开始，但我似乎无法让它在特定时间停止（至少不会导致溢出）。我也尝试过进行有限采集，这很有效，但我无法让它在特定时间开始。所以我有点不知道下一步该做什么。

这是我对有限采集的尝试（似乎只是忽略了开始时间并收集了 0 个样本）：

num_samples = 1000

usrp = uhd.usrp_source(
    ",".join(("", "")),
    uhd.stream_args(
        cpu_format="fc32",
        channels=range(1),
    ),
)

...

usrp.set_start_time(absolute_start_time)
samples = usrp.finite_acquisition(num_samples)

我也尝试了以下一些组合但没有成功（TypeError: in method 'usrp_source_sptr_issue_stream_cmd', argument 2 of type '::uhd::stream_cmd_t const &'）：

usrp.set_command_time(absolute_start_time)
usrp.issue_stream_cmd(uhd.stream_cmd.STREAM_MODE_NUM_SAMPS_AND_DONE)

我还在流程图中尝试了以下内容：

...
usrp = flowgrah.uhd_usrp_source_0
absolute_start_time = uhd.uhd_swig.time_spec_t(start_time)
usrp.set_start_time(absolute_start_time)

flowgrah.start()

stop_cmd = uhd.stream_cmd(uhd.stream_cmd.STREAM_MODE_STOP_CONTINUOUS)
absolute_stop_time = absolute_start_time + uhd.uhd_swig.time_spec_t(collection_time)

usrp.set_command_time(absolute_stop_time)
usrp.issue_stream_cmd(stop_cmd)

无论出于何种原因，生成的流程图在任何大于 0.02 秒的收集时间内都会持续溢出。

【问题讨论】：

标签： gnuradio usrp uhd

【解决方案1】：

我遇到了类似的问题，并通过使用 head 块解决了它。

这是一个简单的示例，它从正弦波源中保存 10,000 个样本，然后退出。

#!/usr/bin/env python
# Evan Widloski - 2017-09-03
# Logging test in gnuradio

from gnuradio import gr
from gnuradio import blocks
from gnuradio import analog

class top_block(gr.top_block):
    def __init__(self, output):
        gr.top_block.__init__(self)

        sample_rate = 32e3
        num_samples = 10000
        ampl = 1

        source = analog.sig_source_f(sample_rate, analog.GR_SIN_WAVE, 100, ampl)
        head = blocks.head(4, num_samples)
        sink = blocks.file_sink(4, output)

        self.connect(source, head)
        self.connect(head, sink)

if __name__ == '__main__':
    try:
        top_block('/tmp/out').run()
    except KeyboardInterrupt:
        pass

【讨论】：

这对我来说非常有效。感谢您的回复！。