나중에 UHD/GNURadio를 사용하여 유한 샘플 수의 샘플을받는 방법은 무엇입니까?

Question

UHD에 GNURadio 파이썬 인터페이스를 사용하고 있으며 특정 시간을 설정하여 샘플 채취를 시작하고 특정 수의 샘플을 수집하거나 특정 시간에 샘플 수집을 중지하려고합니다. 본질적으로 샘플의 시간 측정 스냅 샷을 생성합니다. 이것은 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) 

(형식 오류 : 방법 ' 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