diff options
Diffstat (limited to '')
| -rwxr-xr-x | tests/sockets/Test_Bit_Errorrate.py (renamed from tests/zmq/zmqtest.py) | 33 |
1 files changed, 14 insertions, 19 deletions
diff --git a/tests/zmq/zmqtest.py b/tests/sockets/Test_Bit_Errorrate.py index a046c13..6a989df 100755 --- a/tests/zmq/zmqtest.py +++ b/tests/sockets/Test_Bit_Errorrate.py @@ -5,10 +5,11 @@ # SPDX-License-Identifier: GPL-3.0 # # GNU Radio Python Flow Graph -# Title: ZMQ test -# Author: Naoki Pross +# Title: Bit Error Rate test +# Author: Sara Halter # GNU Radio version: 3.8.2.0 +from gnuradio import analog from gnuradio import blocks from gnuradio import gr from gnuradio.filter import firdes @@ -17,13 +18,14 @@ import signal from argparse import ArgumentParser from gnuradio.eng_arg import eng_float, intx from gnuradio import eng_notation -from gnuradio import zeromq +import fadingui +import numpy as np -class zmqtest(gr.top_block): +class Test_Bit_Errorrate(gr.top_block): def __init__(self): - gr.top_block.__init__(self, "ZMQ test") + gr.top_block.__init__(self, "Bit Error Rate test ") ################################################## # Variables @@ -33,17 +35,17 @@ class zmqtest(gr.top_block): ################################################## # Blocks ################################################## - self.zeromq_req_source_0 = zeromq.req_source(gr.sizeof_gr_complex, 1, '', 100, False, -1) - self.zeromq_rep_sink_0 = zeromq.rep_sink(gr.sizeof_gr_complex, 1, '', 100, False, -1) - self.blocks_throttle_0 = blocks.throttle(gr.sizeof_gr_complex*1, samp_rate,True) + self.fadingui_netsink_0 = fadingui.netsink(address='udp://localhost:31415', dtype="float", vlen=1) + self.blocks_throttle_1 = blocks.throttle(gr.sizeof_float*1, samp_rate,True) + self.analog_noise_source_x_0 = analog.noise_source_f(analog.GR_GAUSSIAN, 1, 0) ################################################## # Connections ################################################## - self.connect((self.blocks_throttle_0, 0), (self.zeromq_rep_sink_0, 0)) - self.connect((self.zeromq_req_source_0, 0), (self.blocks_throttle_0, 0)) + self.connect((self.analog_noise_source_x_0, 0), (self.blocks_throttle_1, 0)) + self.connect((self.blocks_throttle_1, 0), (self.fadingui_netsink_0, 0)) def get_samp_rate(self): @@ -51,15 +53,13 @@ class zmqtest(gr.top_block): def set_samp_rate(self, samp_rate): self.samp_rate = samp_rate - self.blocks_throttle_0.set_sample_rate(self.samp_rate) + self.blocks_throttle_1.set_sample_rate(self.samp_rate) -def main(top_block_cls=zmqtest, options=None): - if gr.enable_realtime_scheduling() != gr.RT_OK: - print("Error: failed to enable real-time scheduling.") +def main(top_block_cls=Test_Bit_Errorrate, options=None): tb = top_block_cls() def sig_handler(sig=None, frame=None): @@ -73,11 +73,6 @@ def main(top_block_cls=zmqtest, options=None): tb.start() - try: - input('Press Enter to quit: ') - except EOFError: - pass - tb.stop() tb.wait() |