#!/usr/bin/env python2 # -*- coding: utf-8 -*- ################################################## # GNU Radio Python Flow Graph # Title: Plutosagr # Generated: Fri Mar 15 12:16:18 2019 ################################################## from gnuradio import analog from gnuradio import blocks from gnuradio import eng_notation from gnuradio import filter from gnuradio import gr from gnuradio import iio from gnuradio.eng_option import eng_option from gnuradio.filter import firdes from optparse import OptionParser class plutosagr(gr.top_block): def __init__(self, plutoip='192.168.2.1', video_rms_alpha=0.0003): gr.top_block.__init__(self, "Plutosagr") ################################################## # Parameters ################################################## self.plutoip = plutoip self.video_rms_alpha = video_rms_alpha ################################################## # Variables ################################################## self.tx_rf_freq = tx_rf_freq = 144000000 self.bb_bw = bb_bw = 1024000 self.video_bw = video_bw = 10000 self.tx_level = tx_level = 0.1 self.tx_bb_freq = tx_bb_freq = 10000 self.rx_rf_freq = rx_rf_freq = tx_rf_freq self.rx_gain = rx_gain = 64 self.rf_bw = rf_bw = bb_bw*4 self.if_samp_rate = if_samp_rate = 128000 self.if_bw = if_bw = 60000 self.bb_samp_rate = bb_samp_rate = bb_bw*2 ################################################## # Blocks ################################################## self.video_rms_nlog10_probe = blocks.probe_signal_f() self.video_rms_nlog10 = blocks.nlog10_ff(20, 1, 2) self.video_rms_det = blocks.rms_ff(video_rms_alpha) self.video_lowpass_filter = filter.fir_filter_fff(1, firdes.low_pass( 1, if_samp_rate, video_bw, int(video_bw*0.1), firdes.WIN_HAMMING, 6.76)) self.tx_bb_carrier_freq = analog.sig_source_f(0, analog.GR_CONST_WAVE, 0, 0, tx_bb_freq*2*3.14) self.rx_if_lopass_filter = filter.fir_filter_ccf(1, firdes.low_pass( 1, if_samp_rate, if_bw, int(if_bw*0.1), firdes.WIN_HAMMING, 6.76)) self.pluto_source_0 = iio.pluto_source(plutoip, int(rx_rf_freq), int(bb_samp_rate), int(rf_bw), 0x8000, True, True, True, "manual", rx_gain, '', True) self.pluto_sink_0 = iio.pluto_sink(plutoip, int(tx_rf_freq), int(bb_samp_rate), int(rf_bw), 0x8000, False, 10, '', True) self.fir_filter_xxx_0 = filter.fir_filter_ccc(bb_samp_rate/if_samp_rate, (firdes.low_pass(1,bb_samp_rate,bb_samp_rate/(2*if_samp_rate), 1000))) self.fir_filter_xxx_0.declare_sample_delay(0) self.dc_blocker_xx_0 = filter.dc_blocker_cc(1024, True) self.blocks_multiply_const_vxx_0 = blocks.multiply_const_vcc((tx_level, )) self.blocks_multiply_conjugate_cc_0 = blocks.multiply_conjugate_cc(1) self.blocks_complex_to_mag_0 = blocks.complex_to_mag(1) self.bb_tx_vco = blocks.vco_c(bb_samp_rate, 1, 1) ################################################## # Connections ################################################## self.connect((self.bb_tx_vco, 0), (self.blocks_multiply_conjugate_cc_0, 0)) self.connect((self.bb_tx_vco, 0), (self.blocks_multiply_const_vxx_0, 0)) self.connect((self.blocks_complex_to_mag_0, 0), (self.video_lowpass_filter, 0)) self.connect((self.blocks_multiply_conjugate_cc_0, 0), (self.fir_filter_xxx_0, 0)) self.connect((self.blocks_multiply_const_vxx_0, 0), (self.pluto_sink_0, 0)) self.connect((self.dc_blocker_xx_0, 0), (self.blocks_multiply_conjugate_cc_0, 1)) self.connect((self.fir_filter_xxx_0, 0), (self.rx_if_lopass_filter, 0)) self.connect((self.pluto_source_0, 0), (self.dc_blocker_xx_0, 0)) self.connect((self.rx_if_lopass_filter, 0), (self.blocks_complex_to_mag_0, 0)) self.connect((self.tx_bb_carrier_freq, 0), (self.bb_tx_vco, 0)) self.connect((self.video_lowpass_filter, 0), (self.video_rms_det, 0)) self.connect((self.video_rms_det, 0), (self.video_rms_nlog10, 0)) self.connect((self.video_rms_nlog10, 0), (self.video_rms_nlog10_probe, 0)) def get_plutoip(self): return self.plutoip def set_plutoip(self, plutoip): self.plutoip = plutoip def get_video_rms_alpha(self): return self.video_rms_alpha def set_video_rms_alpha(self, video_rms_alpha): self.video_rms_alpha = video_rms_alpha self.video_rms_det.set_alpha(self.video_rms_alpha) def get_tx_rf_freq(self): return self.tx_rf_freq def set_tx_rf_freq(self, tx_rf_freq): self.tx_rf_freq = tx_rf_freq self.set_rx_rf_freq(self.tx_rf_freq) self.pluto_sink_0.set_params(int(self.tx_rf_freq), int(self.bb_samp_rate), int(self.rf_bw), 10, '', True) def get_bb_bw(self): return self.bb_bw def set_bb_bw(self, bb_bw): self.bb_bw = bb_bw self.set_rf_bw(self.bb_bw*4) self.set_bb_samp_rate(self.bb_bw*2) def get_video_bw(self): return self.video_bw def set_video_bw(self, video_bw): self.video_bw = video_bw self.video_lowpass_filter.set_taps(firdes.low_pass(1, self.if_samp_rate, self.video_bw, int(self.video_bw*0.1), firdes.WIN_HAMMING, 6.76)) def get_tx_level(self): return self.tx_level def set_tx_level(self, tx_level): self.tx_level = tx_level self.blocks_multiply_const_vxx_0.set_k((self.tx_level, )) def get_tx_bb_freq(self): return self.tx_bb_freq def set_tx_bb_freq(self, tx_bb_freq): self.tx_bb_freq = tx_bb_freq self.tx_bb_carrier_freq.set_offset(self.tx_bb_freq*2*3.14) def get_rx_rf_freq(self): return self.rx_rf_freq def set_rx_rf_freq(self, rx_rf_freq): self.rx_rf_freq = rx_rf_freq self.pluto_source_0.set_params(int(self.rx_rf_freq), int(self.bb_samp_rate), int(self.rf_bw), True, True, True, "manual", self.rx_gain, '', True) def get_rx_gain(self): return self.rx_gain def set_rx_gain(self, rx_gain): self.rx_gain = rx_gain self.pluto_source_0.set_params(int(self.rx_rf_freq), int(self.bb_samp_rate), int(self.rf_bw), True, True, True, "manual", self.rx_gain, '', True) def get_rf_bw(self): return self.rf_bw def set_rf_bw(self, rf_bw): self.rf_bw = rf_bw self.pluto_source_0.set_params(int(self.rx_rf_freq), int(self.bb_samp_rate), int(self.rf_bw), True, True, True, "manual", self.rx_gain, '', True) self.pluto_sink_0.set_params(int(self.tx_rf_freq), int(self.bb_samp_rate), int(self.rf_bw), 10, '', True) def get_if_samp_rate(self): return self.if_samp_rate def set_if_samp_rate(self, if_samp_rate): self.if_samp_rate = if_samp_rate self.video_lowpass_filter.set_taps(firdes.low_pass(1, self.if_samp_rate, self.video_bw, int(self.video_bw*0.1), firdes.WIN_HAMMING, 6.76)) self.rx_if_lopass_filter.set_taps(firdes.low_pass(1, self.if_samp_rate, self.if_bw, int(self.if_bw*0.1), firdes.WIN_HAMMING, 6.76)) self.fir_filter_xxx_0.set_taps((firdes.low_pass(1,self.bb_samp_rate,self.bb_samp_rate/(2*self.if_samp_rate), 1000))) def get_if_bw(self): return self.if_bw def set_if_bw(self, if_bw): self.if_bw = if_bw self.rx_if_lopass_filter.set_taps(firdes.low_pass(1, self.if_samp_rate, self.if_bw, int(self.if_bw*0.1), firdes.WIN_HAMMING, 6.76)) def get_bb_samp_rate(self): return self.bb_samp_rate def set_bb_samp_rate(self, bb_samp_rate): self.bb_samp_rate = bb_samp_rate self.pluto_source_0.set_params(int(self.rx_rf_freq), int(self.bb_samp_rate), int(self.rf_bw), True, True, True, "manual", self.rx_gain, '', True) self.pluto_sink_0.set_params(int(self.tx_rf_freq), int(self.bb_samp_rate), int(self.rf_bw), 10, '', True) self.fir_filter_xxx_0.set_taps((firdes.low_pass(1,self.bb_samp_rate,self.bb_samp_rate/(2*self.if_samp_rate), 1000))) def argument_parser(): parser = OptionParser(usage="%prog: [options]", option_class=eng_option) parser.add_option( "", "--plutoip", dest="plutoip", type="string", default='192.168.2.1', help="Set Pluto Device [default=%default]") return parser def main(top_block_cls=plutosagr, options=None): if options is None: options, _ = argument_parser().parse_args() tb = top_block_cls(plutoip=options.plutoip) tb.start() tb.wait() if __name__ == '__main__': main()