diff options
Diffstat (limited to 'gr-uhd/apps/uhd_siggen_base.py')
| -rw-r--r-- | gr-uhd/apps/uhd_siggen_base.py | 395 |
1 files changed, 395 insertions, 0 deletions
diff --git a/gr-uhd/apps/uhd_siggen_base.py b/gr-uhd/apps/uhd_siggen_base.py new file mode 100644 index 000000000..3c16440ab --- /dev/null +++ b/gr-uhd/apps/uhd_siggen_base.py @@ -0,0 +1,395 @@ +#!/usr/bin/env python +# +# Copyright 2008,2009,2011,2012 Free Software Foundation, Inc. +# +# This file is part of GNU Radio +# +# GNU Radio is free software; you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation; either version 3, or (at your option) +# any later version. +# +# GNU Radio is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with GNU Radio; see the file COPYING. If not, write to +# the Free Software Foundation, Inc., 51 Franklin Street, +# Boston, MA 02110-1301, USA. +# + +DESC_KEY = 'desc' +SAMP_RATE_KEY = 'samp_rate' +LINK_RATE_KEY = 'link_rate' +GAIN_KEY = 'gain' +TX_FREQ_KEY = 'tx_freq' +DSP_FREQ_KEY = 'dsp_freq' +RF_FREQ_KEY = 'rf_freq' +AMPLITUDE_KEY = 'amplitude' +AMPL_RANGE_KEY = 'ampl_range' +WAVEFORM_FREQ_KEY = 'waveform_freq' +WAVEFORM_OFFSET_KEY = 'waveform_offset' +WAVEFORM2_FREQ_KEY = 'waveform2_freq' +FREQ_RANGE_KEY = 'freq_range' +GAIN_RANGE_KEY = 'gain_range' +TYPE_KEY = 'type' + +def setter(ps, key, val): ps[key] = val + +from gnuradio import gr, gru, uhd, eng_notation +from gnuradio.gr.pubsub import pubsub +from gnuradio.eng_option import eng_option +from optparse import OptionParser +import sys +import math + +n2s = eng_notation.num_to_str + +waveforms = { gr.GR_SIN_WAVE : "Complex Sinusoid", + gr.GR_CONST_WAVE : "Constant", + gr.GR_GAUSSIAN : "Gaussian Noise", + gr.GR_UNIFORM : "Uniform Noise", + "2tone" : "Two Tone", + "sweep" : "Sweep" } + +# +# GUI-unaware GNU Radio flowgraph. This may be used either with command +# line applications or GUI applications. +# +class top_block(gr.top_block, pubsub): + def __init__(self, options, args): + gr.top_block.__init__(self) + pubsub.__init__(self) + self._verbose = options.verbose + + #initialize values from options + self._setup_usrpx(options) + self[SAMP_RATE_KEY] = options.samp_rate + self[TX_FREQ_KEY] = options.tx_freq + self[AMPLITUDE_KEY] = options.amplitude + self[WAVEFORM_FREQ_KEY] = options.waveform_freq + self[WAVEFORM_OFFSET_KEY] = options.offset + self[WAVEFORM2_FREQ_KEY] = options.waveform2_freq + self[DSP_FREQ_KEY] = 0 + self[RF_FREQ_KEY] = 0 + + #subscribe set methods + self.subscribe(SAMP_RATE_KEY, self.set_samp_rate) + self.subscribe(GAIN_KEY, self.set_gain) + self.subscribe(TX_FREQ_KEY, self.set_freq) + self.subscribe(AMPLITUDE_KEY, self.set_amplitude) + self.subscribe(WAVEFORM_FREQ_KEY, self.set_waveform_freq) + self.subscribe(WAVEFORM2_FREQ_KEY, self.set_waveform2_freq) + self.subscribe(TYPE_KEY, self.set_waveform) + + #force update on pubsub keys + for key in (SAMP_RATE_KEY, GAIN_KEY, TX_FREQ_KEY, + AMPLITUDE_KEY, WAVEFORM_FREQ_KEY, + WAVEFORM_OFFSET_KEY, WAVEFORM2_FREQ_KEY): + self[key] = self[key] + self[TYPE_KEY] = options.type #set type last + + def _setup_usrpx(self, options): + self._u = uhd.usrp_sink(device_addr=options.args, stream_args=uhd.stream_args('fc32')) + self._u.set_samp_rate(options.samp_rate) + + # Set the subdevice spec + if(options.spec): + self._u.set_subdev_spec(options.spec, 0) + + # Set the gain on the usrp from options + if(options.gain): + self._u.set_gain(options.gain) + + # Set the antenna + if(options.antenna): + self._u.set_antenna(options.antenna, 0) + + # Setup USRP Configuration value + try: + usrp_info = self._u.get_usrp_info() + mboard_id = usrp_info.get("mboard_id").split(" ")[0] + mboard_serial = usrp_info.get("mboard_serial") + if mboard_serial == "": + mboard_serial = "no serial" + dboard_id = usrp_info.get("tx_id").split(" ")[0].split(",")[0] + dboard_serial = usrp_info.get("tx_serial") + if dboard_serial == "": + dboard_serial = "no serial" + subdev = self._u.get_subdev_spec() + antenna = self._u.get_antenna() + + desc_key_str = "Motherboard: %s [%s]\n" % (mboard_id, mboard_serial) + desc_key_str += "Daughterboard: %s [%s]\n" % (dboard_id, dboard_serial) + desc_key_str += "Subdev: %s\n" % subdev + desc_key_str += "Antenna: %s" % antenna + except: + desc_key_str = "USRP configuration output not implemented in this version" + + self.publish(DESC_KEY, lambda: desc_key_str) + self.publish(FREQ_RANGE_KEY, self._u.get_freq_range) + self.publish(GAIN_RANGE_KEY, self._u.get_gain_range) + self.publish(GAIN_KEY, self._u.get_gain) + + print "UHD Signal Generator" + print "Version: %s" % uhd.get_version_string() + print "\nUsing USRP configuration:" + print desc_key_str + "\n" + + # Direct asynchronous notifications to callback function + if options.show_async_msg: + self.async_msgq = gr.msg_queue(0) + self.async_src = uhd.amsg_source("", self.async_msgq) + self.async_rcv = gru.msgq_runner(self.async_msgq, self.async_callback) + + def async_callback(self, msg): + md = self.async_src.msg_to_async_metadata_t(msg) + print "Channel: %i Time: %f Event: %i" % (md.channel, md.time_spec.get_real_secs(), md.event_code) + + def _set_tx_amplitude(self, ampl): + """ + Sets the transmit amplitude sent to the USRP + @param ampl the amplitude or None for automatic + """ + ampl_range = self[AMPL_RANGE_KEY] + if ampl is None: + ampl = (ampl_range[1] - ampl_range[0])*0.15 + ampl_range[0] + self[AMPLITUDE_KEY] = max(ampl_range[0], min(ampl, ampl_range[1])) + + def set_samp_rate(self, sr): + self._u.set_samp_rate(sr) + sr = self._u.get_samp_rate() + + if self[TYPE_KEY] in (gr.GR_SIN_WAVE, gr.GR_CONST_WAVE): + self._src.set_sampling_freq(self[SAMP_RATE_KEY]) + elif self[TYPE_KEY] == "2tone": + self._src1.set_sampling_freq(self[SAMP_RATE_KEY]) + self._src2.set_sampling_freq(self[SAMP_RATE_KEY]) + elif self[TYPE_KEY] == "sweep": + self._src1.set_sampling_freq(self[SAMP_RATE_KEY]) + self._src2.set_sampling_freq(self[WAVEFORM_FREQ_KEY]*2*math.pi/self[SAMP_RATE_KEY]) + else: + return True # Waveform not yet set + + if self._verbose: + print "Set sample rate to:", sr + + return True + + def set_gain(self, gain): + if gain is None: + g = self[GAIN_RANGE_KEY] + gain = float(g.start()+g.stop())/2 + if self._verbose: + print "Using auto-calculated mid-point TX gain" + self[GAIN_KEY] = gain + return + self._u.set_gain(gain) + if self._verbose: + print "Set TX gain to:", gain + + def set_freq(self, target_freq): + + if target_freq is None: + f = self[FREQ_RANGE_KEY] + target_freq = float(f.start()+f.stop())/2.0 + if self._verbose: + print "Using auto-calculated mid-point frequency" + self[TX_FREQ_KEY] = target_freq + return + + tr = self._u.set_center_freq(target_freq) + fs = "%sHz" % (n2s(target_freq),) + if tr is not None: + self._freq = target_freq + self[DSP_FREQ_KEY] = tr.actual_dsp_freq + self[RF_FREQ_KEY] = tr.actual_rf_freq + if self._verbose: + print "Set center frequency to", self._u.get_center_freq() + print "Tx RF frequency: %sHz" % (n2s(tr.actual_rf_freq),) + print "Tx DSP frequency: %sHz" % (n2s(tr.actual_dsp_freq),) + elif self._verbose: + print "Failed to set freq." + return tr + + def set_waveform_freq(self, freq): + if self[TYPE_KEY] == gr.GR_SIN_WAVE: + self._src.set_frequency(freq) + elif self[TYPE_KEY] == "2tone": + self._src1.set_frequency(freq) + elif self[TYPE_KEY] == 'sweep': + #there is no set sensitivity, redo fg + self[TYPE_KEY] = self[TYPE_KEY] + return True + + def set_waveform2_freq(self, freq): + if freq is None: + self[WAVEFORM2_FREQ_KEY] = -self[WAVEFORM_FREQ_KEY] + return + if self[TYPE_KEY] == "2tone": + self._src2.set_frequency(freq) + elif self[TYPE_KEY] == "sweep": + self._src1.set_frequency(freq) + return True + + def set_waveform(self, type): + self.lock() + self.disconnect_all() + if type == gr.GR_SIN_WAVE or type == gr.GR_CONST_WAVE: + self._src = gr.sig_source_c(self[SAMP_RATE_KEY], # Sample rate + type, # Waveform type + self[WAVEFORM_FREQ_KEY], # Waveform frequency + self[AMPLITUDE_KEY], # Waveform amplitude + self[WAVEFORM_OFFSET_KEY]) # Waveform offset + elif type == gr.GR_GAUSSIAN or type == gr.GR_UNIFORM: + self._src = gr.noise_source_c(type, self[AMPLITUDE_KEY]) + elif type == "2tone": + self._src1 = gr.sig_source_c(self[SAMP_RATE_KEY], + gr.GR_SIN_WAVE, + self[WAVEFORM_FREQ_KEY], + self[AMPLITUDE_KEY]/2.0, + 0) + if(self[WAVEFORM2_FREQ_KEY] is None): + self[WAVEFORM2_FREQ_KEY] = -self[WAVEFORM_FREQ_KEY] + + self._src2 = gr.sig_source_c(self[SAMP_RATE_KEY], + gr.GR_SIN_WAVE, + self[WAVEFORM2_FREQ_KEY], + self[AMPLITUDE_KEY]/2.0, + 0) + self._src = gr.add_cc() + self.connect(self._src1,(self._src,0)) + self.connect(self._src2,(self._src,1)) + elif type == "sweep": + # rf freq is center frequency + # waveform_freq is total swept width + # waveform2_freq is sweep rate + # will sweep from (rf_freq-waveform_freq/2) to (rf_freq+waveform_freq/2) + if self[WAVEFORM2_FREQ_KEY] is None: + self[WAVEFORM2_FREQ_KEY] = 0.1 + + self._src1 = gr.sig_source_f(self[SAMP_RATE_KEY], + gr.GR_TRI_WAVE, + self[WAVEFORM2_FREQ_KEY], + 1.0, + -0.5) + self._src2 = gr.frequency_modulator_fc(self[WAVEFORM_FREQ_KEY]*2*math.pi/self[SAMP_RATE_KEY]) + self._src = gr.multiply_const_cc(self[AMPLITUDE_KEY]) + self.connect(self._src1,self._src2,self._src) + else: + raise RuntimeError("Unknown waveform type") + + self.connect(self._src, self._u) + self.unlock() + + if self._verbose: + print "Set baseband modulation to:", waveforms[type] + if type == gr.GR_SIN_WAVE: + print "Modulation frequency: %sHz" % (n2s(self[WAVEFORM_FREQ_KEY]),) + print "Initial phase:", self[WAVEFORM_OFFSET_KEY] + elif type == "2tone": + print "Tone 1: %sHz" % (n2s(self[WAVEFORM_FREQ_KEY]),) + print "Tone 2: %sHz" % (n2s(self[WAVEFORM2_FREQ_KEY]),) + elif type == "sweep": + print "Sweeping across %sHz to %sHz" % (n2s(-self[WAVEFORM_FREQ_KEY]/2.0),n2s(self[WAVEFORM_FREQ_KEY]/2.0)) + print "Sweep rate: %sHz" % (n2s(self[WAVEFORM2_FREQ_KEY]),) + print "TX amplitude:", self[AMPLITUDE_KEY] + + + def set_amplitude(self, amplitude): + if amplitude < 0.0 or amplitude > 1.0: + if self._verbose: + print "Amplitude out of range:", amplitude + return False + + if self[TYPE_KEY] in (gr.GR_SIN_WAVE, gr.GR_CONST_WAVE, gr.GR_GAUSSIAN, gr.GR_UNIFORM): + self._src.set_amplitude(amplitude) + elif self[TYPE_KEY] == "2tone": + self._src1.set_amplitude(amplitude/2.0) + self._src2.set_amplitude(amplitude/2.0) + elif self[TYPE_KEY] == "sweep": + self._src.set_k(amplitude) + else: + return True # Waveform not yet set + + if self._verbose: + print "Set amplitude to:", amplitude + return True + +def get_options(): + usage="%prog: [options]" + + parser = OptionParser(option_class=eng_option, usage=usage) + parser.add_option("-a", "--args", type="string", default="", + help="UHD device address args , [default=%default]") + parser.add_option("", "--spec", type="string", default=None, + help="Subdevice of UHD device where appropriate") + parser.add_option("-A", "--antenna", type="string", default=None, + help="select Rx Antenna where appropriate") + parser.add_option("-s", "--samp-rate", type="eng_float", default=1e6, + help="set sample rate (bandwidth) [default=%default]") + parser.add_option("-g", "--gain", type="eng_float", default=None, + help="set gain in dB (default is midpoint)") + parser.add_option("-f", "--tx-freq", type="eng_float", default=None, + help="Set carrier frequency to FREQ [default=mid-point]", + metavar="FREQ") + parser.add_option("-x", "--waveform-freq", type="eng_float", default=0, + help="Set baseband waveform frequency to FREQ [default=%default]") + parser.add_option("-y", "--waveform2-freq", type="eng_float", default=None, + help="Set 2nd waveform frequency to FREQ [default=%default]") + parser.add_option("--sine", dest="type", action="store_const", const=gr.GR_SIN_WAVE, + help="Generate a carrier modulated by a complex sine wave", + default=gr.GR_SIN_WAVE) + parser.add_option("--const", dest="type", action="store_const", const=gr.GR_CONST_WAVE, + help="Generate a constant carrier") + parser.add_option("--offset", type="eng_float", default=0, + help="Set waveform phase offset to OFFSET [default=%default]") + parser.add_option("--gaussian", dest="type", action="store_const", const=gr.GR_GAUSSIAN, + help="Generate Gaussian random output") + parser.add_option("--uniform", dest="type", action="store_const", const=gr.GR_UNIFORM, + help="Generate Uniform random output") + parser.add_option("--2tone", dest="type", action="store_const", const="2tone", + help="Generate Two Tone signal for IMD testing") + parser.add_option("--sweep", dest="type", action="store_const", const="sweep", + help="Generate a swept sine wave") + parser.add_option("", "--amplitude", type="eng_float", default=0.15, + help="Set output amplitude to AMPL (0.0-1.0) [default=%default]", + metavar="AMPL") + parser.add_option("-v", "--verbose", action="store_true", default=False, + help="Use verbose console output [default=%default]") + parser.add_option("", "--show-async-msg", action="store_true", default=False, + help="Show asynchronous message notifications from UHD [default=%default]") + + (options, args) = parser.parse_args() + + return (options, args) + +# If this script is executed, the following runs. If it is imported, +# the below does not run. +def test_main(): + if gr.enable_realtime_scheduling() != gr.RT_OK: + print "Note: failed to enable realtime scheduling, continuing" + + # Grab command line options and create top block + try: + (options, args) = get_options() + tb = top_block(options, args) + + except RuntimeError, e: + print e + sys.exit(1) + + tb.start() + raw_input('Press Enter to quit: ') + tb.stop() + tb.wait() + +# Make sure to create the top block (tb) within a function: +# That code in main will allow tb to go out of scope on return, +# which will call the decontructor on usrp and stop transmit. +# Whats odd is that grc works fine with tb in the __main__, +# perhaps its because the try/except clauses around tb. +if __name__ == "__main__": + test_main() |