UHD/GNURadioを使用して将来有限個のサンプルを受信する方法は？

Question

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例外：メソッドで " usrp_source_sptr_issue_stream_cmd '型の引数2' :: 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) 

何らかの理由で、生成されたフローグラフが.02sの収集時間を超えるものに対して一貫してオーバーフローします。

Answer 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