diff options
Diffstat (limited to 'gr-wxgui/src/python')
| -rw-r--r-- | gr-wxgui/src/python/Makefile.am | 7 | ||||
| -rwxr-xr-x | gr-wxgui/src/python/fftsink.py | 488 | ||||
| -rwxr-xr-x | gr-wxgui/src/python/numbersink.py | 614 | ||||
| -rwxr-xr-x | gr-wxgui/src/python/scopesink.py | 650 | ||||
| -rw-r--r-- | gr-wxgui/src/python/stdgui.py | 90 | ||||
| -rwxr-xr-x | gr-wxgui/src/python/waterfallsink.py | 475 |
6 files changed, 1 insertions, 2323 deletions
diff --git a/gr-wxgui/src/python/Makefile.am b/gr-wxgui/src/python/Makefile.am index f466db163..bda7c362c 100644 --- a/gr-wxgui/src/python/Makefile.am +++ b/gr-wxgui/src/python/Makefile.am @@ -1,5 +1,5 @@ # -# Copyright 2004,2005 Free Software Foundation, Inc. +# Copyright 2004,2005,2008 Free Software Foundation, Inc. # # This file is part of GNU Radio # @@ -31,16 +31,11 @@ ourlibdir = $(grpyexecdir)/wxgui ourpython_PYTHON = \ __init__.py \ form.py \ - fftsink.py \ fftsink2.py \ plot.py \ powermate.py \ - scopesink.py \ scopesink2.py \ - waterfallsink.py \ waterfallsink2.py \ slider.py \ - stdgui.py \ stdgui2.py \ - numbersink.py \ numbersink2.py diff --git a/gr-wxgui/src/python/fftsink.py b/gr-wxgui/src/python/fftsink.py deleted file mode 100755 index 85e3a1f9e..000000000 --- a/gr-wxgui/src/python/fftsink.py +++ /dev/null @@ -1,488 +0,0 @@ -#!/usr/bin/env python -# -# Copyright 2003,2004,2005,2006,2007 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. -# - -from gnuradio import gr, gru, window -from gnuradio.wxgui import stdgui -import wx -import gnuradio.wxgui.plot as plot -import numpy -import threading -import math - -default_fftsink_size = (640,240) -default_fft_rate = gr.prefs().get_long('wxgui', 'fft_rate', 15) - -class fft_sink_base(object): - def __init__(self, input_is_real=False, baseband_freq=0, y_per_div=10, ref_level=50, - sample_rate=1, fft_size=512, - fft_rate=default_fft_rate, - average=False, avg_alpha=None, title='', peak_hold=False): - - # initialize common attributes - self.baseband_freq = baseband_freq - self.y_divs = 8 - self.y_per_div=y_per_div - self.ref_level = ref_level - self.sample_rate = sample_rate - self.fft_size = fft_size - self.fft_rate = fft_rate - self.average = average - if avg_alpha is None: - self.avg_alpha = 2.0 / fft_rate - else: - self.avg_alpha = avg_alpha - self.title = title - self.peak_hold = peak_hold - self.input_is_real = input_is_real - self.msgq = gr.msg_queue(2) # queue that holds a maximum of 2 messages - - def set_y_per_div(self, y_per_div): - self.y_per_div = y_per_div - - def set_ref_level(self, ref_level): - self.ref_level = ref_level - - def set_average(self, average): - self.average = average - if average: - self.avg.set_taps(self.avg_alpha) - self.set_peak_hold(False) - else: - self.avg.set_taps(1.0) - - def set_peak_hold(self, enable): - self.peak_hold = enable - if enable: - self.set_average(False) - self.win.set_peak_hold(enable) - - def set_avg_alpha(self, avg_alpha): - self.avg_alpha = avg_alpha - - def set_baseband_freq(self, baseband_freq): - self.baseband_freq = baseband_freq - - def set_sample_rate(self, sample_rate): - self.sample_rate = sample_rate - self._set_n() - - def _set_n(self): - self.one_in_n.set_n(max(1, int(self.sample_rate/self.fft_size/self.fft_rate))) - - -class fft_sink_f(gr.hier_block, fft_sink_base): - def __init__(self, fg, parent, baseband_freq=0, - y_per_div=10, ref_level=50, sample_rate=1, fft_size=512, - fft_rate=default_fft_rate, average=False, avg_alpha=None, - title='', size=default_fftsink_size, peak_hold=False): - - fft_sink_base.__init__(self, input_is_real=True, baseband_freq=baseband_freq, - y_per_div=y_per_div, ref_level=ref_level, - sample_rate=sample_rate, fft_size=fft_size, - fft_rate=fft_rate, - average=average, avg_alpha=avg_alpha, title=title, - peak_hold=peak_hold) - - s2p = gr.stream_to_vector(gr.sizeof_float, self.fft_size) - self.one_in_n = gr.keep_one_in_n(gr.sizeof_float * self.fft_size, - max(1, int(self.sample_rate/self.fft_size/self.fft_rate))) - - mywindow = window.blackmanharris(self.fft_size) - fft = gr.fft_vfc(self.fft_size, True, mywindow) - power = 0 - for tap in mywindow: - power += tap*tap - - c2mag = gr.complex_to_mag(self.fft_size) - self.avg = gr.single_pole_iir_filter_ff(1.0, self.fft_size) - - # FIXME We need to add 3dB to all bins but the DC bin - log = gr.nlog10_ff(20, self.fft_size, - -20*math.log10(self.fft_size)-10*math.log10(power/self.fft_size)) - sink = gr.message_sink(gr.sizeof_float * self.fft_size, self.msgq, True) - - fg.connect (s2p, self.one_in_n, fft, c2mag, self.avg, log, sink) - gr.hier_block.__init__(self, fg, s2p, sink) - - self.win = fft_window(self, parent, size=size) - self.set_average(self.average) - - -class fft_sink_c(gr.hier_block, fft_sink_base): - def __init__(self, fg, parent, baseband_freq=0, - y_per_div=10, ref_level=50, sample_rate=1, fft_size=512, - fft_rate=default_fft_rate, average=False, avg_alpha=None, - title='', size=default_fftsink_size, peak_hold=False): - - fft_sink_base.__init__(self, input_is_real=False, baseband_freq=baseband_freq, - y_per_div=y_per_div, ref_level=ref_level, - sample_rate=sample_rate, fft_size=fft_size, - fft_rate=fft_rate, - average=average, avg_alpha=avg_alpha, title=title, - peak_hold=peak_hold) - - s2p = gr.stream_to_vector(gr.sizeof_gr_complex, self.fft_size) - self.one_in_n = gr.keep_one_in_n(gr.sizeof_gr_complex * self.fft_size, - max(1, int(self.sample_rate/self.fft_size/self.fft_rate))) - mywindow = window.blackmanharris(self.fft_size) - power = 0 - for tap in mywindow: - power += tap*tap - - fft = gr.fft_vcc(self.fft_size, True, mywindow) - c2mag = gr.complex_to_mag(fft_size) - self.avg = gr.single_pole_iir_filter_ff(1.0, fft_size) - log = gr.nlog10_ff(20, self.fft_size, - -20*math.log10(self.fft_size)-10*math.log10(power/self.fft_size)) - sink = gr.message_sink(gr.sizeof_float * fft_size, self.msgq, True) - - fg.connect(s2p, self.one_in_n, fft, c2mag, self.avg, log, sink) - gr.hier_block.__init__(self, fg, s2p, sink) - - self.win = fft_window(self, parent, size=size) - self.set_average(self.average) - - -# ------------------------------------------------------------------------ - -myDATA_EVENT = wx.NewEventType() -EVT_DATA_EVENT = wx.PyEventBinder (myDATA_EVENT, 0) - - -class DataEvent(wx.PyEvent): - def __init__(self, data): - wx.PyEvent.__init__(self) - self.SetEventType (myDATA_EVENT) - self.data = data - - def Clone (self): - self.__class__ (self.GetId()) - - -class input_watcher (threading.Thread): - def __init__ (self, msgq, fft_size, event_receiver, **kwds): - threading.Thread.__init__ (self, **kwds) - self.setDaemon (1) - self.msgq = msgq - self.fft_size = fft_size - self.event_receiver = event_receiver - self.keep_running = True - self.start () - - def run (self): - while (self.keep_running): - msg = self.msgq.delete_head() # blocking read of message queue - itemsize = int(msg.arg1()) - nitems = int(msg.arg2()) - - s = msg.to_string() # get the body of the msg as a string - - # There may be more than one FFT frame in the message. - # If so, we take only the last one - if nitems > 1: - start = itemsize * (nitems - 1) - s = s[start:start+itemsize] - - complex_data = numpy.fromstring (s, numpy.float32) - de = DataEvent (complex_data) - wx.PostEvent (self.event_receiver, de) - del de - - -class fft_window (plot.PlotCanvas): - def __init__ (self, fftsink, parent, id = -1, - pos = wx.DefaultPosition, size = wx.DefaultSize, - style = wx.DEFAULT_FRAME_STYLE, name = ""): - plot.PlotCanvas.__init__ (self, parent, id, pos, size, style, name) - - self.y_range = None - self.fftsink = fftsink - self.peak_hold = False - self.peak_vals = None - - self.SetEnableGrid (True) - # self.SetEnableZoom (True) - # self.SetBackgroundColour ('black') - - self.build_popup_menu() - - EVT_DATA_EVENT (self, self.set_data) - wx.EVT_CLOSE (self, self.on_close_window) - self.Bind(wx.EVT_RIGHT_UP, self.on_right_click) - - self.input_watcher = input_watcher(fftsink.msgq, fftsink.fft_size, self) - - - def on_close_window (self, event): - print "fft_window:on_close_window" - self.keep_running = False - - - def set_data (self, evt): - dB = evt.data - L = len (dB) - - if self.peak_hold: - if self.peak_vals is None: - self.peak_vals = dB - else: - self.peak_vals = numpy.maximum(dB, self.peak_vals) - dB = self.peak_vals - - x = max(abs(self.fftsink.sample_rate), abs(self.fftsink.baseband_freq)) - if x >= 1e9: - sf = 1e-9 - units = "GHz" - elif x >= 1e6: - sf = 1e-6 - units = "MHz" - else: - sf = 1e-3 - units = "kHz" - - if self.fftsink.input_is_real: # only plot 1/2 the points - x_vals = ((numpy.arange (L/2) - * (self.fftsink.sample_rate * sf / L)) - + self.fftsink.baseband_freq * sf) - points = numpy.zeros((len(x_vals), 2), numpy.float64) - points[:,0] = x_vals - points[:,1] = dB[0:L/2] - else: - # the "negative freqs" are in the second half of the array - x_vals = ((numpy.arange (-L/2, L/2) - * (self.fftsink.sample_rate * sf / L)) - + self.fftsink.baseband_freq * sf) - points = numpy.zeros((len(x_vals), 2), numpy.float64) - points[:,0] = x_vals - points[:,1] = numpy.concatenate ((dB[L/2:], dB[0:L/2])) - - - lines = plot.PolyLine (points, colour='BLUE') - - graphics = plot.PlotGraphics ([lines], - title=self.fftsink.title, - xLabel = units, yLabel = "dB") - - self.Draw (graphics, xAxis=None, yAxis=self.y_range) - self.update_y_range () - - def set_peak_hold(self, enable): - self.peak_hold = enable - self.peak_vals = None - - def update_y_range (self): - ymax = self.fftsink.ref_level - ymin = self.fftsink.ref_level - self.fftsink.y_per_div * self.fftsink.y_divs - self.y_range = self._axisInterval ('min', ymin, ymax) - - def on_average(self, evt): - # print "on_average" - self.fftsink.set_average(evt.IsChecked()) - - def on_peak_hold(self, evt): - # print "on_peak_hold" - self.fftsink.set_peak_hold(evt.IsChecked()) - - def on_incr_ref_level(self, evt): - # print "on_incr_ref_level" - self.fftsink.set_ref_level(self.fftsink.ref_level - + self.fftsink.y_per_div) - - def on_decr_ref_level(self, evt): - # print "on_decr_ref_level" - self.fftsink.set_ref_level(self.fftsink.ref_level - - self.fftsink.y_per_div) - - def on_incr_y_per_div(self, evt): - # print "on_incr_y_per_div" - self.fftsink.set_y_per_div(next_up(self.fftsink.y_per_div, (1,2,5,10,20))) - - def on_decr_y_per_div(self, evt): - # print "on_decr_y_per_div" - self.fftsink.set_y_per_div(next_down(self.fftsink.y_per_div, (1,2,5,10,20))) - - def on_y_per_div(self, evt): - # print "on_y_per_div" - Id = evt.GetId() - if Id == self.id_y_per_div_1: - self.fftsink.set_y_per_div(1) - elif Id == self.id_y_per_div_2: - self.fftsink.set_y_per_div(2) - elif Id == self.id_y_per_div_5: - self.fftsink.set_y_per_div(5) - elif Id == self.id_y_per_div_10: - self.fftsink.set_y_per_div(10) - elif Id == self.id_y_per_div_20: - self.fftsink.set_y_per_div(20) - - - def on_right_click(self, event): - menu = self.popup_menu - for id, pred in self.checkmarks.items(): - item = menu.FindItemById(id) - item.Check(pred()) - self.PopupMenu(menu, event.GetPosition()) - - - def build_popup_menu(self): - self.id_incr_ref_level = wx.NewId() - self.id_decr_ref_level = wx.NewId() - self.id_incr_y_per_div = wx.NewId() - self.id_decr_y_per_div = wx.NewId() - self.id_y_per_div_1 = wx.NewId() - self.id_y_per_div_2 = wx.NewId() - self.id_y_per_div_5 = wx.NewId() - self.id_y_per_div_10 = wx.NewId() - self.id_y_per_div_20 = wx.NewId() - self.id_average = wx.NewId() - self.id_peak_hold = wx.NewId() - - self.Bind(wx.EVT_MENU, self.on_average, id=self.id_average) - self.Bind(wx.EVT_MENU, self.on_peak_hold, id=self.id_peak_hold) - self.Bind(wx.EVT_MENU, self.on_incr_ref_level, id=self.id_incr_ref_level) - self.Bind(wx.EVT_MENU, self.on_decr_ref_level, id=self.id_decr_ref_level) - self.Bind(wx.EVT_MENU, self.on_incr_y_per_div, id=self.id_incr_y_per_div) - self.Bind(wx.EVT_MENU, self.on_decr_y_per_div, id=self.id_decr_y_per_div) - self.Bind(wx.EVT_MENU, self.on_y_per_div, id=self.id_y_per_div_1) - self.Bind(wx.EVT_MENU, self.on_y_per_div, id=self.id_y_per_div_2) - self.Bind(wx.EVT_MENU, self.on_y_per_div, id=self.id_y_per_div_5) - self.Bind(wx.EVT_MENU, self.on_y_per_div, id=self.id_y_per_div_10) - self.Bind(wx.EVT_MENU, self.on_y_per_div, id=self.id_y_per_div_20) - - - # make a menu - menu = wx.Menu() - self.popup_menu = menu - menu.AppendCheckItem(self.id_average, "Average") - menu.AppendCheckItem(self.id_peak_hold, "Peak Hold") - menu.Append(self.id_incr_ref_level, "Incr Ref Level") - menu.Append(self.id_decr_ref_level, "Decr Ref Level") - # menu.Append(self.id_incr_y_per_div, "Incr dB/div") - # menu.Append(self.id_decr_y_per_div, "Decr dB/div") - menu.AppendSeparator() - # we'd use RadioItems for these, but they're not supported on Mac - menu.AppendCheckItem(self.id_y_per_div_1, "1 dB/div") - menu.AppendCheckItem(self.id_y_per_div_2, "2 dB/div") - menu.AppendCheckItem(self.id_y_per_div_5, "5 dB/div") - menu.AppendCheckItem(self.id_y_per_div_10, "10 dB/div") - menu.AppendCheckItem(self.id_y_per_div_20, "20 dB/div") - - self.checkmarks = { - self.id_average : lambda : self.fftsink.average, - self.id_peak_hold : lambda : self.fftsink.peak_hold, - self.id_y_per_div_1 : lambda : self.fftsink.y_per_div == 1, - self.id_y_per_div_2 : lambda : self.fftsink.y_per_div == 2, - self.id_y_per_div_5 : lambda : self.fftsink.y_per_div == 5, - self.id_y_per_div_10 : lambda : self.fftsink.y_per_div == 10, - self.id_y_per_div_20 : lambda : self.fftsink.y_per_div == 20, - } - - -def next_up(v, seq): - """ - Return the first item in seq that is > v. - """ - for s in seq: - if s > v: - return s - return v - -def next_down(v, seq): - """ - Return the last item in seq that is < v. - """ - rseq = list(seq[:]) - rseq.reverse() - - for s in rseq: - if s < v: - return s - return v - - -# ---------------------------------------------------------------- -# Deprecated interfaces -# ---------------------------------------------------------------- - -# returns (block, win). -# block requires a single input stream of float -# win is a subclass of wxWindow - -def make_fft_sink_f(fg, parent, title, fft_size, input_rate, ymin = 0, ymax=50): - - block = fft_sink_f(fg, parent, title=title, fft_size=fft_size, sample_rate=input_rate, - y_per_div=(ymax - ymin)/8, ref_level=ymax) - return (block, block.win) - -# returns (block, win). -# block requires a single input stream of gr_complex -# win is a subclass of wxWindow - -def make_fft_sink_c(fg, parent, title, fft_size, input_rate, ymin=0, ymax=50): - block = fft_sink_c(fg, parent, title=title, fft_size=fft_size, sample_rate=input_rate, - y_per_div=(ymax - ymin)/8, ref_level=ymax) - return (block, block.win) - - -# ---------------------------------------------------------------- -# Standalone test app -# ---------------------------------------------------------------- - -class test_app_flow_graph (stdgui.gui_flow_graph): - def __init__(self, frame, panel, vbox, argv): - stdgui.gui_flow_graph.__init__ (self, frame, panel, vbox, argv) - - fft_size = 256 - - # build our flow graph - input_rate = 20.48e3 - - # Generate a complex sinusoid - #src1 = gr.sig_source_c (input_rate, gr.GR_SIN_WAVE, 2e3, 1) - src1 = gr.sig_source_c (input_rate, gr.GR_CONST_WAVE, 5.75e3, 1) - - # We add these throttle blocks so that this demo doesn't - # suck down all the CPU available. Normally you wouldn't use these. - thr1 = gr.throttle(gr.sizeof_gr_complex, input_rate) - - sink1 = fft_sink_c (self, panel, title="Complex Data", fft_size=fft_size, - sample_rate=input_rate, baseband_freq=100e3, - ref_level=0, y_per_div=20) - vbox.Add (sink1.win, 1, wx.EXPAND) - self.connect (src1, thr1, sink1) - - #src2 = gr.sig_source_f (input_rate, gr.GR_SIN_WAVE, 2e3, 1) - src2 = gr.sig_source_f (input_rate, gr.GR_CONST_WAVE, 5.75e3, 1) - thr2 = gr.throttle(gr.sizeof_float, input_rate) - sink2 = fft_sink_f (self, panel, title="Real Data", fft_size=fft_size*2, - sample_rate=input_rate, baseband_freq=100e3, - ref_level=0, y_per_div=20) - vbox.Add (sink2.win, 1, wx.EXPAND) - self.connect (src2, thr2, sink2) - -def main (): - app = stdgui.stdapp (test_app_flow_graph, - "FFT Sink Test App") - app.MainLoop () - -if __name__ == '__main__': - main () diff --git a/gr-wxgui/src/python/numbersink.py b/gr-wxgui/src/python/numbersink.py deleted file mode 100755 index 5427d2a40..000000000 --- a/gr-wxgui/src/python/numbersink.py +++ /dev/null @@ -1,614 +0,0 @@ -#!/usr/bin/env python -# -# Copyright 2003,2004,2005,2006,2007 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., 59 Temple Place - Suite 330, -# Boston, MA 02111-1307, USA. -# - -from gnuradio import gr, gru, window -from gnuradio.wxgui import stdgui -import wx -#from wx import StaticText -import gnuradio.wxgui.plot as plot -import numpy -import threading -import math - -default_numbersink_size = (640,240) -default_number_rate = gr.prefs().get_long('wxgui', 'number_rate', 15) - -class number_sink_base(object): - def __init__(self, input_is_real=False, unit='',base_value=0, minval=-100.0,maxval=100.0,factor=1.0,decimal_places=10, ref_level=50, - sample_rate=1, - number_rate=default_number_rate, - average=False, avg_alpha=None, label='', peak_hold=False): - - # initialize common attributes - self.unit=unit - self.base_value = base_value - self.minval=minval - self.maxval=maxval - self.factor=factor - self.y_divs = 8 - self.decimal_places=decimal_places - self.ref_level = ref_level - self.sample_rate = sample_rate - number_size=1 - self.number_size = number_size - self.number_rate = number_rate - self.average = average - if avg_alpha is None: - self.avg_alpha = 2.0 / number_rate - else: - self.avg_alpha = avg_alpha - self.label = label - self.peak_hold = peak_hold - self.show_gauge = True - self.input_is_real = input_is_real - self.msgq = gr.msg_queue(2) # queue that holds a maximum of 2 messages - - def set_decimal_places(self, decimal_places): - self.decimal_places = decimal_places - - def set_ref_level(self, ref_level): - self.ref_level = ref_level - - def print_current_value(self, comment): - print comment,self.win.current_value - - def set_average(self, average): - self.average = average - if average: - self.avg.set_taps(self.avg_alpha) - self.set_peak_hold(False) - else: - self.avg.set_taps(1.0) - - def set_peak_hold(self, enable): - self.peak_hold = enable - if enable: - self.set_average(False) - self.win.set_peak_hold(enable) - - def set_show_gauge(self, enable): - self.show_gauge = enable - self.win.set_show_gauge(enable) - - def set_avg_alpha(self, avg_alpha): - self.avg_alpha = avg_alpha - - def set_base_value(self, base_value): - self.base_value = base_value - - -class number_sink_f(gr.hier_block, number_sink_base): - def __init__(self, fg, parent, unit='',base_value=0,minval=-100.0,maxval=100.0,factor=1.0, - decimal_places=10, ref_level=50, sample_rate=1, #number_size=512, - number_rate=default_number_rate, average=False, avg_alpha=None, - label='', size=default_numbersink_size, peak_hold=False): - - number_sink_base.__init__(self, unit=unit, input_is_real=True, base_value=base_value, - minval=minval,maxval=maxval,factor=factor, - decimal_places=decimal_places, ref_level=ref_level, - sample_rate=sample_rate, #number_size=number_size, - number_rate=number_rate, - average=average, avg_alpha=avg_alpha, label=label, - peak_hold=peak_hold) - - number_size=1 - #s2p = gr.stream_to_vector(gr.sizeof_float, number_size) - one_in_n = gr.keep_one_in_n(gr.sizeof_float, - max(1, int(sample_rate/number_rate))) - - - #c2mag = gr.complex_to_mag(number_size) - self.avg = gr.single_pole_iir_filter_ff(1.0, number_size) - - # FIXME We need to add 3dB to all bins but the DC bin - #log = gr.nlog10_ff(20, number_size, - # -20*math.log10(number_size)-10*math.log10(power/number_size)) - sink = gr.message_sink(gr.sizeof_float , self.msgq, True) - - #fg.connect (s2p, one_in_n, fft, c2mag, self.avg, log, sink) - fg.connect(self.avg,one_in_n,sink) - gr.hier_block.__init__(self, fg, self.avg, sink) - self.win = number_window(self, parent, size=size,label=label) - self.set_average(self.average) - - -class number_sink_c(gr.hier_block, number_sink_base): - def __init__(self, fg, parent, unit='',base_value=0,minval=-100.0,maxval=100.0,factor=1.0, - decimal_places=10, ref_level=50, sample_rate=1, #number_size=512, - number_rate=default_number_rate, average=False, avg_alpha=None, - label='', size=default_numbersink_size, peak_hold=False): - - number_sink_base.__init__(self, unit=unit, input_is_real=False, base_value=base_value,factor=factor, - minval=minval,maxval=maxval,decimal_places=decimal_places, ref_level=ref_level, - sample_rate=sample_rate, #number_size=number_size, - number_rate=number_rate, - average=average, avg_alpha=avg_alpha, label=label, - peak_hold=peak_hold) - - number_size=1 - one_in_n = gr.keep_one_in_n(gr.sizeof_gr_complex, - max(1, int(sample_rate/number_rate))) - - - #c2mag = gr.complex_to_mag(number_size) - self.avg = gr.single_pole_iir_filter_cc(1.0, number_size) - - # FIXME We need to add 3dB to all bins but the DC bin - #log = gr.nlog10_ff(20, number_size, - # -20*math.log10(number_size)-10*math.log10(power/number_size)) - sink = gr.message_sink(gr.sizeof_gr_complex , self.msgq, True) - - #fg.connect (s2p, one_in_n, fft, c2mag, self.avg, log, sink) - fg.connect(self.avg,one_in_n,sink) - gr.hier_block.__init__(self, fg, self.avg, sink) - self.win = number_window(self, parent, size=size,label=label) - self.set_average(self.average) - - -# ------------------------------------------------------------------------ - -myDATA_EVENT = wx.NewEventType() -EVT_DATA_EVENT = wx.PyEventBinder (myDATA_EVENT, 0) - - -class DataEvent(wx.PyEvent): - def __init__(self, data): - wx.PyEvent.__init__(self) - self.SetEventType (myDATA_EVENT) - self.data = data - - def Clone (self): - self.__class__ (self.GetId()) - - -class input_watcher (threading.Thread): - def __init__ (self, msgq, number_size, event_receiver, **kwds): - threading.Thread.__init__ (self, **kwds) - self.setDaemon (1) - self.msgq = msgq - self.number_size = number_size - self.event_receiver = event_receiver - self.keep_running = True - self.start () - - def run (self): - while (self.keep_running): - msg = self.msgq.delete_head() # blocking read of message queue - itemsize = int(msg.arg1()) - nitems = int(msg.arg2()) - - s = msg.to_string() # get the body of the msg as a string - - # There may be more than one number in the message. - # If so, we take only the last one - if nitems > 1: - start = itemsize * (nitems - 1) - s = s[start:start+itemsize] - - complex_data = numpy.fromstring (s, numpy.float32) - de = DataEvent (complex_data) - wx.PostEvent (self.event_receiver, de) - del de - -#======================================================================================== -class static_text_window (wx.StaticText): #plot.PlotCanvas): - def __init__ (self, parent, numbersink,id = -1,label="number", - pos = wx.DefaultPosition, size = wx.DefaultSize, - style = wx.DEFAULT_FRAME_STYLE, name = ""): - #plot.PlotCanvas.__init__ (self, parent, id, pos, size, style, name) - wx.StaticText.__init__(self, parent, id, label, pos, size, style, name) - #self.static_text=wx.StaticText( parent, id, label, pos, (size[0]/2,size[1]/2), style, name) - #gauge_style = wx.GA_HORIZONTAL - #self.gauge=wx.Gauge( parent, id, range=1000, pos=(pos[0],pos[1]+size[1]/2),size=(size[0]/2,size[1]/2), style=gauge_style, name = "gauge") - #wx.BoxSizer.__init__ (self,wx.VERTICAL) - #self.Add (self.static_text, 0, wx.EXPAND) - #self.Add (self.gauge, 1, wx.EXPAND) - self.parent=parent - self.label=label - #self.y_range = None - self.numbersink = numbersink - self.peak_hold = False - self.peak_vals = None - - #self.SetEnableGrid (True) - # self.SetEnableZoom (True) - # self.SetBackgroundColour ('black') - - self.build_popup_menu() - - #EVT_DATA_EVENT (self, self.set_data) - #wx.EVT_CLOSE (self, self.on_close_window) - #self.Bind(wx.EVT_RIGHT_UP, self.on_right_click) - self.Bind(wx.EVT_RIGHT_UP, self.on_right_click) - - #self.input_watcher = input_watcher(numbersink.msgq, numbersink.number_size, self) - - - def on_close_window (self, event): - print "number_window:on_close_window" - self.keep_running = False - - - def set_peak_hold(self, enable): - self.peak_hold = enable - self.peak_vals = None - - def update_y_range (self): - ymax = self.numbersink.ref_level - ymin = self.numbersink.ref_level - self.numbersink.decimal_places * self.numbersink.y_divs - self.y_range = self._axisInterval ('min', ymin, ymax) - - def on_average(self, evt): - # print "on_average" - self.numbersink.set_average(evt.IsChecked()) - - def on_peak_hold(self, evt): - # print "on_peak_hold" - self.numbersink.set_peak_hold(evt.IsChecked()) - - def on_show_gauge(self, evt): - # print "on_show_gauge" - #if evt.IsChecked(): - self.numbersink.set_show_gauge(evt.IsChecked()) - print evt.IsChecked() - # print "show gauge" - #else: - # self.parent.gauge.Hide() - # print "hide gauge" - - def on_incr_ref_level(self, evt): - # print "on_incr_ref_level" - self.numbersink.set_ref_level(self.numbersink.ref_level - + self.numbersink.decimal_places) - - def on_decr_ref_level(self, evt): - # print "on_decr_ref_level" - self.numbersink.set_ref_level(self.numbersink.ref_level - - self.numbersink.decimal_places) - - def on_incr_decimal_places(self, evt): - # print "on_incr_decimal_places" - self.numbersink.set_decimal_places(self.numbersink.decimal_places+1) #next_up(self.numbersink.decimal_places, (1,2,5,10,20))) - - def on_decr_decimal_places(self, evt): - # print "on_decr_decimal_places" - self.numbersink.set_decimal_places(max(self.numbersink.decimal_places-1,0)) #next_down(self.numbersink.decimal_places, (1,2,5,10,20))) - - def on_decimal_places(self, evt): - # print "on_decimal_places" - Id = evt.GetId() - if Id == self.id_decimal_places_0: - self.numbersink.set_decimal_places(0) - elif Id == self.id_decimal_places_1: - self.numbersink.set_decimal_places(1) - elif Id == self.id_decimal_places_2: - self.numbersink.set_decimal_places(2) - elif Id == self.id_decimal_places_3: - self.numbersink.set_decimal_places(3) - elif Id == self.id_decimal_places_6: - self.numbersink.set_decimal_places(6) - elif Id == self.id_decimal_places_9: - self.numbersink.set_decimal_places(9) - - - def on_right_click(self, event): - menu = self.popup_menu - for id, pred in self.checkmarks.items(): - item = menu.FindItemById(id) - item.Check(pred()) - self.PopupMenu(menu, event.GetPosition()) - - - def build_popup_menu(self): - #self.id_hide_gauge = wx.NewId() - self.id_show_gauge = wx.NewId() - self.id_incr_ref_level = wx.NewId() - self.id_decr_ref_level = wx.NewId() - self.id_incr_decimal_places = wx.NewId() - self.id_decr_decimal_places = wx.NewId() - self.id_decimal_places_0 = wx.NewId() - self.id_decimal_places_1 = wx.NewId() - self.id_decimal_places_2 = wx.NewId() - self.id_decimal_places_3 = wx.NewId() - self.id_decimal_places_6 = wx.NewId() - self.id_decimal_places_9 = wx.NewId() - self.id_average = wx.NewId() - self.id_peak_hold = wx.NewId() - - self.Bind(wx.EVT_MENU, self.on_average, id=self.id_average) - self.Bind(wx.EVT_MENU, self.on_peak_hold, id=self.id_peak_hold) - #self.Bind(wx.EVT_MENU, self.on_hide_gauge, id=self.id_hide_gauge) - self.Bind(wx.EVT_MENU, self.on_show_gauge, id=self.id_show_gauge) - self.Bind(wx.EVT_MENU, self.on_incr_ref_level, id=self.id_incr_ref_level) - self.Bind(wx.EVT_MENU, self.on_decr_ref_level, id=self.id_decr_ref_level) - self.Bind(wx.EVT_MENU, self.on_incr_decimal_places, id=self.id_incr_decimal_places) - self.Bind(wx.EVT_MENU, self.on_decr_decimal_places, id=self.id_decr_decimal_places) - self.Bind(wx.EVT_MENU, self.on_decimal_places, id=self.id_decimal_places_0) - self.Bind(wx.EVT_MENU, self.on_decimal_places, id=self.id_decimal_places_1) - self.Bind(wx.EVT_MENU, self.on_decimal_places, id=self.id_decimal_places_2) - self.Bind(wx.EVT_MENU, self.on_decimal_places, id=self.id_decimal_places_3) - self.Bind(wx.EVT_MENU, self.on_decimal_places, id=self.id_decimal_places_6) - self.Bind(wx.EVT_MENU, self.on_decimal_places, id=self.id_decimal_places_9) - - - # make a menu - menu = wx.Menu() - self.popup_menu = menu - menu.AppendCheckItem(self.id_average, "Average") - menu.AppendCheckItem(self.id_peak_hold, "Peak Hold") - #menu.Append(self.id_hide_gauge, "Hide gauge") - menu.AppendCheckItem(self.id_show_gauge, "Show gauge") - menu.Append(self.id_incr_ref_level, "Incr Ref Level") - menu.Append(self.id_decr_ref_level, "Decr Ref Level") - menu.Append(self.id_incr_decimal_places, "Incr decimal places") - menu.Append(self.id_decr_decimal_places, "Decr decimal places") - menu.AppendSeparator() - # we'd use RadioItems for these, but they're not supported on Mac - menu.AppendCheckItem(self.id_decimal_places_0, "0 decimal places") - menu.AppendCheckItem(self.id_decimal_places_1, "1 decimal places") - menu.AppendCheckItem(self.id_decimal_places_2, "2 decimal places") - menu.AppendCheckItem(self.id_decimal_places_3, "3 decimal places") - menu.AppendCheckItem(self.id_decimal_places_6, "6 decimal places") - menu.AppendCheckItem(self.id_decimal_places_9, "9 decimal places") - - self.checkmarks = { - self.id_average : lambda : self.numbersink.average, - self.id_peak_hold : lambda : self.numbersink.peak_hold,# self.id_hide_gauge : lambda : self.numbersink.hide_gauge, - self.id_show_gauge : lambda : self.numbersink.show_gauge, - self.id_decimal_places_0 : lambda : self.numbersink.decimal_places == 0, - self.id_decimal_places_1 : lambda : self.numbersink.decimal_places == 1, - self.id_decimal_places_2 : lambda : self.numbersink.decimal_places == 2, - self.id_decimal_places_3 : lambda : self.numbersink.decimal_places == 3, - self.id_decimal_places_6 : lambda : self.numbersink.decimal_places == 6, - self.id_decimal_places_9 : lambda : self.numbersink.decimal_places == 9, - } - - -def next_up(v, seq): - """ - Return the first item in seq that is > v. - """ - for s in seq: - if s > v: - return s - return v - -def next_down(v, seq): - """ - Return the last item in seq that is < v. - """ - rseq = list(seq[:]) - rseq.reverse() - - for s in rseq: - if s < v: - return s - return v - - -#======================================================================================== -class number_window (plot.PlotCanvas): - def __init__ (self, numbersink, parent, id = -1,label="number", - pos = wx.DefaultPosition, size = wx.DefaultSize, - style = wx.DEFAULT_FRAME_STYLE, name = ""): - plot.PlotCanvas.__init__ (self, parent, id, pos, size, style, name) - #wx.StaticText.__init__(self, parent, id, label, pos, (size[0]/2,size[1]/2), style, name) - #print 'parent',parent - self.static_text=static_text_window( self, numbersink,id, label, pos, (size[0]/2,size[1]/2), style, name) - gauge_style = wx.GA_HORIZONTAL - vbox=wx.BoxSizer(wx.VERTICAL) - vbox.Add (self.static_text, 0, wx.EXPAND) - self.current_value=None - if numbersink.input_is_real: - self.gauge=wx.Gauge( self, id, range=1000, pos=(pos[0],pos[1]+size[1]/2),size=(size[0]/2,size[1]/2), style=gauge_style, name = "gauge") - vbox.Add (self.gauge, 1, wx.EXPAND) - else: - self.gauge=wx.Gauge( self, id, range=1000, pos=(pos[0],pos[1]+size[1]/3),size=(size[0]/2,size[1]/3), style=gauge_style, name = "gauge") - #hbox=wx.BoxSizer(wx.HORIZONTAL) - self.gauge_imag=wx.Gauge( self, id, range=1000, pos=(pos[0],pos[1]+size[1]*2/3),size=(size[0]/2,size[1]/3), style=gauge_style, name = "gauge_imag") - vbox.Add (self.gauge, 1, wx.EXPAND) - vbox.Add (self.gauge_imag, 1, wx.EXPAND) - #vbox.Add (hbox, 1, wx.EXPAND) - self.sizer = vbox - self.SetSizer (self.sizer) - self.SetAutoLayout (True) - self.sizer.Fit (self) - - self.label=label - #self.y_range = None - self.numbersink = numbersink - self.peak_hold = False - self.peak_vals = None - - #self.SetEnableGrid (True) - # self.SetEnableZoom (True) - # self.SetBackgroundColour ('black') - - #self.build_popup_menu() - - EVT_DATA_EVENT (self, self.set_data) - wx.EVT_CLOSE (self, self.on_close_window) - #self.Bind(wx.EVT_RIGHT_UP, self.on_right_click) - #self.Bind(wx.EVT_RIGHT_UP, self.on_right_click) - - self.input_watcher = input_watcher(numbersink.msgq, numbersink.number_size, self) - - - def on_close_window (self, event): - print "number_window:on_close_window" - self.keep_running = False - - def set_show_gauge(self, enable): - self.show_gauge = enable - if enable: - self.gauge.Show() - if not self.numbersink.input_is_real: - self.gauge_imag.Show() - #print 'show' - else: - self.gauge.Hide() - if not self.numbersink.input_is_real: - self.gauge_imag.Hide() - #print 'hide' - - def set_data (self, evt): - numbers = evt.data - L = len (numbers) - - if self.peak_hold: - if self.peak_vals is None: - self.peak_vals = numbers - else: - self.peak_vals = numpy.maximum(numbers, self.peak_vals) - numbers = self.peak_vals - - if self.numbersink.input_is_real: - real_value=numbers[0]*self.numbersink.factor + self.numbersink.base_value - imag_value=0.0 - self.current_value=real_value - else: - real_value=numbers[0]*self.numbersink.factor + self.numbersink.base_value - imag_value=numbers[1]*self.numbersink.factor + self.numbersink.base_value - self.current_value=complex(real_value,imag_value) - #x = max(abs(self.numbersink.sample_rate), abs(self.numbersink.base_value)) - x = max(real_value, imag_value) - if x >= 1e9: - sf = 1e-9 - unit_prefix = "G" - elif x >= 1e6: - sf = 1e-6 - unit_prefix = "M" - elif x>= 1e3: - sf = 1e-3 - unit_prefix = "k" - else : - sf = 1 - unit_prefix = "" - #self.update_y_range () - if self.numbersink.input_is_real: - showtext = "%s: %.*f %s%s" % (self.label, self.numbersink.decimal_places,real_value*sf,unit_prefix,self.numbersink.unit) - else: - showtext = "%s: %.*f,%.*f %s%s" % (self.label, self.numbersink.decimal_places,real_value*sf, - self.numbersink.decimal_places,imag_value*sf,unit_prefix,self.numbersink.unit) - self.static_text.SetLabel(showtext) - #print (int(float((real_value-self.numbersink.base_value)*1000.0/(self.numbersink.maxval-self.numbersink.minval)))+500) - self.gauge.SetValue(int(float((real_value-self.numbersink.base_value)*1000.0/(self.numbersink.maxval-self.numbersink.minval)))+500) - if not self.numbersink.input_is_real: - self.gauge.SetValue(int(float((imag_value-self.numbersink.base_value)*1000.0/(self.numbersink.maxval-self.numbersink.minval)))+500) - - def set_peak_hold(self, enable): - self.peak_hold = enable - self.peak_vals = None - - def update_y_range (self): - ymax = self.numbersink.ref_level - ymin = self.numbersink.ref_level - self.numbersink.decimal_places * self.numbersink.y_divs - self.y_range = self._axisInterval ('min', ymin, ymax) - - def on_average(self, evt): - # print "on_average" - self.numbersink.set_average(evt.IsChecked()) - - def on_peak_hold(self, evt): - # print "on_peak_hold" - self.numbersink.set_peak_hold(evt.IsChecked()) - - - - - - - - - - -# ---------------------------------------------------------------- -# Deprecated interfaces -# ---------------------------------------------------------------- - -# returns (block, win). -# block requires a single input stream of float -# win is a subclass of wxWindow - -def make_number_sink_f(fg, parent, label, number_size, input_rate, ymin = 0, ymax=50): - - block = number_sink_f(fg, parent, label=label, number_size=number_size, sample_rate=input_rate, - decimal_places=(ymax - ymin)/8, ref_level=ymax) - return (block, block.win) - -# returns (block, win). -# block requires a single input stream of gr_complex -# win is a subclass of wxWindow - -def make_number_sink_c(fg, parent, label, number_size, input_rate, ymin=0, ymax=50): - block = number_sink_c(fg, parent, label=label, number_size=number_size, sample_rate=input_rate, - decimal_places=(ymax - ymin)/8, ref_level=ymax) - return (block, block.win) - - -# ---------------------------------------------------------------- -# Standalone test app -# ---------------------------------------------------------------- - -class test_app_flow_graph (stdgui.gui_flow_graph): - def __init__(self, frame, panel, vbox, argv): - stdgui.gui_flow_graph.__init__ (self, frame, panel, vbox, argv) - - #number_size = 256 - - # build our flow graph - input_rate = 20.48e3 - - # Generate a complex sinusoid - src1 = gr.sig_source_c (input_rate, gr.GR_SIN_WAVE, 2e3, 1) - #src1 = gr.sig_source_c (input_rate, gr.GR_CONST_WAVE, 5.75e3, 1) - - # We add these throttle blocks so that this demo doesn't - # suck down all the CPU available. Normally you wouldn't use these. - thr1 = gr.throttle(gr.sizeof_gr_complex, input_rate) - - #sink1 = number_sink_c (self, panel, label="Complex Data", number_size=number_size, - # sample_rate=input_rate, base_value=100e3, - # ref_level=0, decimal_places=3) - #vbox.Add (sink1.win, 1, wx.EXPAND) - #self.connect (src1, thr1, sink1) - - src2 = gr.sig_source_f (input_rate, gr.GR_SIN_WAVE, 2e3, 1) - #src2 = gr.sig_source_f (input_rate, gr.GR_CONST_WAVE, 5.75e3, 1) - thr2 = gr.throttle(gr.sizeof_float, input_rate) - sink2 = number_sink_f (self, panel, unit='Hz',label="Real Data", avg_alpha=0.001,#number_size=number_size*2, - sample_rate=input_rate, base_value=100e3, - ref_level=0, decimal_places=3) - vbox.Add (sink2.win, 1, wx.EXPAND) - sink3 = number_sink_c (self, panel, unit='V',label="Complex Data", avg_alpha=0.001,#number_size=number_size*2, - sample_rate=input_rate, base_value=0, - ref_level=0, decimal_places=3) - vbox.Add (sink3.win, 1, wx.EXPAND) - self.connect (src2, thr2, sink2) - self.connect (src1, thr1, sink3) -def main (): - app = stdgui.stdapp (test_app_flow_graph, - "Number Sink Test App") - app.MainLoop () - -if __name__ == '__main__': - main () diff --git a/gr-wxgui/src/python/scopesink.py b/gr-wxgui/src/python/scopesink.py deleted file mode 100755 index ffd4529bc..000000000 --- a/gr-wxgui/src/python/scopesink.py +++ /dev/null @@ -1,650 +0,0 @@ -#!/usr/bin/env python -# -# Copyright 2003,2004,2006,2007 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. -# - -from gnuradio import gr, gru, eng_notation -from gnuradio.wxgui import stdgui -import wx -import gnuradio.wxgui.plot as plot -import numpy -import threading -import struct - -default_scopesink_size = (640, 240) -default_v_scale = 1000 -default_frame_decim = gr.prefs().get_long('wxgui', 'frame_decim', 1) - -class scope_sink_f(gr.hier_block): - def __init__(self, fg, parent, title='', sample_rate=1, - size=default_scopesink_size, frame_decim=default_frame_decim, - v_scale=default_v_scale, t_scale=None): - msgq = gr.msg_queue(2) # message queue that holds at most 2 messages - self.guts = gr.oscope_sink_f(sample_rate, msgq) - gr.hier_block.__init__(self, fg, self.guts, self.guts) - self.win = scope_window(win_info (msgq, sample_rate, frame_decim, - v_scale, t_scale, self.guts, title), parent) - - def set_sample_rate(self, sample_rate): - self.guts.set_sample_rate(sample_rate) - self.win.info.set_sample_rate(sample_rate) - -class scope_sink_c(gr.hier_block): - def __init__(self, fg, parent, title='', sample_rate=1, - size=default_scopesink_size, frame_decim=default_frame_decim, - v_scale=default_v_scale, t_scale=None): - msgq = gr.msg_queue(2) # message queue that holds at most 2 messages - c2f = gr.complex_to_float() - self.guts = gr.oscope_sink_f(sample_rate, msgq) - fg.connect((c2f, 0), (self.guts, 0)) - fg.connect((c2f, 1), (self.guts, 1)) - gr.hier_block.__init__(self, fg, c2f, self.guts) - self.win = scope_window(win_info(msgq, sample_rate, frame_decim, - v_scale, t_scale, self.guts, title), parent) - - def set_sample_rate(self, sample_rate): - self.guts.set_sample_rate(sample_rate) - self.win.info.set_sample_rate(sample_rate) - -# ======================================================================== -# This is the deprecated interface, retained for compatibility... -# -# returns (block, win). -# block requires a N input stream of float -# win is a subclass of wxWindow - -def make_scope_sink_f (fg, parent, label, input_rate): - block = scope_sink_f(fg, parent, title=label, sample_rate=input_rate) - return (block, block.win) - -# ======================================================================== - - -time_base_list = [ # time / division - 1.0e-7, # 100ns / div - 2.5e-7, - 5.0e-7, - 1.0e-6, # 1us / div - 2.5e-6, - 5.0e-6, - 1.0e-5, # 10us / div - 2.5e-5, - 5.0e-5, - 1.0e-4, # 100us / div - 2.5e-4, - 5.0e-4, - 1.0e-3, # 1ms / div - 2.5e-3, - 5.0e-3, - 1.0e-2, # 10ms / div - 2.5e-2, - 5.0e-2 - ] - -v_scale_list = [ # counts / div, LARGER gains are SMALLER /div, appear EARLIER - 2.0e-3, # 2m / div, don't call it V/div it's actually counts/div - 5.0e-3, - 1.0e-2, - 2.0e-2, - 5.0e-2, - 1.0e-1, - 2.0e-1, - 5.0e-1, - 1.0e+0, - 2.0e+0, - 5.0e+0, - 1.0e+1, - 2.0e+1, - 5.0e+1, - 1.0e+2, - 2.0e+2, - 5.0e+2, - 1.0e+3, - 2.0e+3, - 5.0e+3, - 1.0e+4 # 10000 /div, USRP full scale is -/+ 32767 - ] - - -wxDATA_EVENT = wx.NewEventType() - -def EVT_DATA_EVENT(win, func): - win.Connect(-1, -1, wxDATA_EVENT, func) - -class DataEvent(wx.PyEvent): - def __init__(self, data): - wx.PyEvent.__init__(self) - self.SetEventType (wxDATA_EVENT) - self.data = data - - def Clone (self): - self.__class__ (self.GetId()) - - -class win_info (object): - __slots__ = ['msgq', 'sample_rate', 'frame_decim', 'v_scale', - 'scopesink', 'title', - 'time_scale_cursor', 'v_scale_cursor', 'marker', 'xy', - 'autorange', 'running'] - - def __init__ (self, msgq, sample_rate, frame_decim, v_scale, t_scale, - scopesink, title = "Oscilloscope"): - self.msgq = msgq - self.sample_rate = sample_rate - self.frame_decim = frame_decim - self.scopesink = scopesink - self.title = title; - - self.time_scale_cursor = gru.seq_with_cursor(time_base_list, initial_value = t_scale) - self.v_scale_cursor = gru.seq_with_cursor(v_scale_list, initial_value = v_scale) - - self.marker = 'line' - self.xy = False - if v_scale == None: # 0 and None are both False, but 0 != None - self.autorange = True - else: - self.autorange = False # 0 is a valid v_scale - self.running = True - - def get_time_per_div (self): - return self.time_scale_cursor.current () - - def get_volts_per_div (self): - return self.v_scale_cursor.current () - - def set_sample_rate(self, sample_rate): - self.sample_rate = sample_rate - - def get_sample_rate (self): - return self.sample_rate - - def get_decimation_rate (self): - return 1.0 - - def set_marker (self, s): - self.marker = s - - def get_marker (self): - return self.marker - - -class input_watcher (threading.Thread): - def __init__ (self, msgq, event_receiver, frame_decim, **kwds): - threading.Thread.__init__ (self, **kwds) - self.setDaemon (1) - self.msgq = msgq - self.event_receiver = event_receiver - self.frame_decim = frame_decim - self.iscan = 0 - self.keep_running = True - self.start () - - def run (self): - # print "input_watcher: pid = ", os.getpid () - while (self.keep_running): - msg = self.msgq.delete_head() # blocking read of message queue - if self.iscan == 0: # only display at frame_decim - self.iscan = self.frame_decim - - nchan = int(msg.arg1()) # number of channels of data in msg - nsamples = int(msg.arg2()) # number of samples in each channel - - s = msg.to_string() # get the body of the msg as a string - - bytes_per_chan = nsamples * gr.sizeof_float - - records = [] - for ch in range (nchan): - - start = ch * bytes_per_chan - chan_data = s[start:start+bytes_per_chan] - rec = numpy.fromstring (chan_data, numpy.float32) - records.append (rec) - - # print "nrecords = %d, reclen = %d" % (len (records),nsamples) - - de = DataEvent (records) - wx.PostEvent (self.event_receiver, de) - records = [] - del de - - # end if iscan == 0 - self.iscan -= 1 - - -class scope_window (wx.Panel): - - def __init__ (self, info, parent, id = -1, - pos = wx.DefaultPosition, size = wx.DefaultSize, name = ""): - wx.Panel.__init__ (self, parent, -1) - self.info = info - - vbox = wx.BoxSizer (wx.VERTICAL) - - self.graph = graph_window (info, self, -1) - - vbox.Add (self.graph, 1, wx.EXPAND) - vbox.Add (self.make_control_box(), 0, wx.EXPAND) - vbox.Add (self.make_control2_box(), 0, wx.EXPAND) - - self.sizer = vbox - self.SetSizer (self.sizer) - self.SetAutoLayout (True) - self.sizer.Fit (self) - self.set_autorange(self.info.autorange) - - - # second row of control buttons etc. appears BELOW control_box - def make_control2_box (self): - ctrlbox = wx.BoxSizer (wx.HORIZONTAL) - - self.inc_v_button = wx.Button (self, 1101, " < ", style=wx.BU_EXACTFIT) - self.inc_v_button.SetToolTipString ("Increase vertical range") - wx.EVT_BUTTON (self, 1101, self.incr_v_scale) # ID matches button ID above - - self.dec_v_button = wx.Button (self, 1100, " > ", style=wx.BU_EXACTFIT) - self.dec_v_button.SetToolTipString ("Decrease vertical range") - wx.EVT_BUTTON (self, 1100, self.decr_v_scale) - - self.v_scale_label = wx.StaticText (self, 1002, "None") # vertical /div - self.update_v_scale_label () - - self.autorange_checkbox = wx.CheckBox (self, 1102, "Autorange") - self.autorange_checkbox.SetToolTipString ("Select autorange on/off") - wx.EVT_CHECKBOX(self, 1102, self.autorange_checkbox_event) - - ctrlbox.Add ((5,0) ,0) # left margin space - ctrlbox.Add (self.inc_v_button, 0, wx.EXPAND) - ctrlbox.Add (self.dec_v_button, 0, wx.EXPAND) - ctrlbox.Add (self.v_scale_label, 0, wx.ALIGN_CENTER) - ctrlbox.Add ((20,0) ,0) # spacer - ctrlbox.Add (self.autorange_checkbox, 0, wx.ALIGN_CENTER) - - return ctrlbox - - def make_control_box (self): - ctrlbox = wx.BoxSizer (wx.HORIZONTAL) - - tb_left = wx.Button (self, 1001, " < ", style=wx.BU_EXACTFIT) - tb_left.SetToolTipString ("Increase time base") - wx.EVT_BUTTON (self, 1001, self.incr_timebase) - - - tb_right = wx.Button (self, 1000, " > ", style=wx.BU_EXACTFIT) - tb_right.SetToolTipString ("Decrease time base") - wx.EVT_BUTTON (self, 1000, self.decr_timebase) - - self.time_base_label = wx.StaticText (self, 1002, "") - self.update_timebase_label () - - ctrlbox.Add ((5,0) ,0) - # ctrlbox.Add (wx.StaticText (self, -1, "Horiz Scale: "), 0, wx.ALIGN_CENTER) - ctrlbox.Add (tb_left, 0, wx.EXPAND) - ctrlbox.Add (tb_right, 0, wx.EXPAND) - ctrlbox.Add (self.time_base_label, 0, wx.ALIGN_CENTER) - - ctrlbox.Add ((10,0) ,1) # stretchy space - - ctrlbox.Add (wx.StaticText (self, -1, "Trig: "), 0, wx.ALIGN_CENTER) - self.trig_chan_choice = wx.Choice (self, 1004, - choices = ['Ch1', 'Ch2', 'Ch3', 'Ch4']) - self.trig_chan_choice.SetToolTipString ("Select channel for trigger") - wx.EVT_CHOICE (self, 1004, self.trig_chan_choice_event) - ctrlbox.Add (self.trig_chan_choice, 0, wx.ALIGN_CENTER) - - self.trig_mode_choice = wx.Choice (self, 1005, - choices = ['Auto', 'Pos', 'Neg']) - self.trig_mode_choice.SetToolTipString ("Select trigger slope or Auto (untriggered roll)") - wx.EVT_CHOICE (self, 1005, self.trig_mode_choice_event) - ctrlbox.Add (self.trig_mode_choice, 0, wx.ALIGN_CENTER) - - trig_level50 = wx.Button (self, 1006, "50%") - trig_level50.SetToolTipString ("Set trigger level to 50%") - wx.EVT_BUTTON (self, 1006, self.set_trig_level50) - ctrlbox.Add (trig_level50, 0, wx.EXPAND) - - run_stop = wx.Button (self, 1007, "Run/Stop") - run_stop.SetToolTipString ("Toggle Run/Stop mode") - wx.EVT_BUTTON (self, 1007, self.run_stop) - ctrlbox.Add (run_stop, 0, wx.EXPAND) - - ctrlbox.Add ((10, 0) ,1) # stretchy space - - ctrlbox.Add (wx.StaticText (self, -1, "Fmt: "), 0, wx.ALIGN_CENTER) - self.marker_choice = wx.Choice (self, 1002, choices = self._marker_choices) - self.marker_choice.SetToolTipString ("Select plotting with lines, pluses or dots") - wx.EVT_CHOICE (self, 1002, self.marker_choice_event) - ctrlbox.Add (self.marker_choice, 0, wx.ALIGN_CENTER) - - self.xy_choice = wx.Choice (self, 1003, choices = ['X:t', 'X:Y']) - self.xy_choice.SetToolTipString ("Select X vs time or X vs Y display") - wx.EVT_CHOICE (self, 1003, self.xy_choice_event) - ctrlbox.Add (self.xy_choice, 0, wx.ALIGN_CENTER) - - return ctrlbox - - _marker_choices = ['line', 'plus', 'dot'] - - def update_timebase_label (self): - time_per_div = self.info.get_time_per_div () - s = ' ' + eng_notation.num_to_str (time_per_div) + 's/div' - self.time_base_label.SetLabel (s) - - def decr_timebase (self, evt): - self.info.time_scale_cursor.prev () - self.update_timebase_label () - - def incr_timebase (self, evt): - self.info.time_scale_cursor.next () - self.update_timebase_label () - - def update_v_scale_label (self): - volts_per_div = self.info.get_volts_per_div () - s = ' ' + eng_notation.num_to_str (volts_per_div) + '/div' # Not V/div - self.v_scale_label.SetLabel (s) - - def decr_v_scale (self, evt): - self.info.v_scale_cursor.prev () - self.update_v_scale_label () - - def incr_v_scale (self, evt): - self.info.v_scale_cursor.next () - self.update_v_scale_label () - - def marker_choice_event (self, evt): - s = evt.GetString () - self.set_marker (s) - - def set_autorange(self, on): - if on: - self.v_scale_label.SetLabel(" (auto)") - self.info.autorange = True - self.autorange_checkbox.SetValue(True) - self.inc_v_button.Enable(False) - self.dec_v_button.Enable(False) - else: - if self.graph.y_range: - (l,u) = self.graph.y_range # found by autorange - self.info.v_scale_cursor.set_index_by_value((u-l)/8.0) - self.update_v_scale_label() - self.info.autorange = False - self.autorange_checkbox.SetValue(False) - self.inc_v_button.Enable(True) - self.dec_v_button.Enable(True) - - def autorange_checkbox_event(self, evt): - if evt.Checked(): - self.set_autorange(True) - else: - self.set_autorange(False) - - def set_marker (self, s): - self.info.set_marker (s) # set info for drawing routines - i = self.marker_choice.FindString (s) - assert i >= 0, "Hmmm, set_marker problem" - self.marker_choice.SetSelection (i) - - def set_format_line (self): - self.set_marker ('line') - - def set_format_dot (self): - self.set_marker ('dot') - - def set_format_plus (self): - self.set_marker ('plus') - - def xy_choice_event (self, evt): - s = evt.GetString () - self.info.xy = s == 'X:Y' - - def trig_chan_choice_event (self, evt): - s = evt.GetString () - ch = int (s[-1]) - 1 - self.info.scopesink.set_trigger_channel (ch) - - def trig_mode_choice_event (self, evt): - sink = self.info.scopesink - s = evt.GetString () - if s == 'Pos': - sink.set_trigger_mode (gr.gr_TRIG_POS_SLOPE) - elif s == 'Neg': - sink.set_trigger_mode (gr.gr_TRIG_NEG_SLOPE) - elif s == 'Auto': - sink.set_trigger_mode (gr.gr_TRIG_AUTO) - else: - assert 0, "Bad trig_mode_choice string" - - def set_trig_level50 (self, evt): - self.info.scopesink.set_trigger_level_auto () - - def run_stop (self, evt): - self.info.running = not self.info.running - - -class graph_window (plot.PlotCanvas): - - channel_colors = ['BLUE', 'RED', - 'CYAN', 'MAGENTA', 'GREEN', 'YELLOW'] - - def __init__ (self, info, parent, id = -1, - pos = wx.DefaultPosition, size = (640, 240), - style = wx.DEFAULT_FRAME_STYLE, name = ""): - plot.PlotCanvas.__init__ (self, parent, id, pos, size, style, name) - - self.SetXUseScopeTicks (True) - self.SetEnableGrid (True) - self.SetEnableZoom (True) - self.SetEnableLegend(True) - # self.SetBackgroundColour ('black') - - self.info = info; - self.y_range = None - self.x_range = None - self.avg_y_min = None - self.avg_y_max = None - self.avg_x_min = None - self.avg_x_max = None - - EVT_DATA_EVENT (self, self.format_data) - - self.input_watcher = input_watcher (info.msgq, self, info.frame_decim) - - def channel_color (self, ch): - return self.channel_colors[ch % len(self.channel_colors)] - - def format_data (self, evt): - if not self.info.running: - return - - if self.info.xy: - self.format_xy_data (evt) - return - - info = self.info - records = evt.data - nchannels = len (records) - npoints = len (records[0]) - - objects = [] - - Ts = 1.0 / (info.get_sample_rate () / info.get_decimation_rate ()) - x_vals = Ts * numpy.arange (-npoints/2, npoints/2) - - # preliminary clipping based on time axis here, instead of in graphics code - time_per_window = self.info.get_time_per_div () * 10 - n = int (time_per_window / Ts + 0.5) - n = n & ~0x1 # make even - n = max (2, min (n, npoints)) - - self.SetXUseScopeTicks (True) # use 10 divisions, no labels - - for ch in range(nchannels): - r = records[ch] - - # plot middle n points of record - - lb = npoints/2 - n/2 - ub = npoints/2 + n/2 - # points = zip (x_vals[lb:ub], r[lb:ub]) - points = numpy.zeros ((ub-lb, 2), numpy.float64) - points[:,0] = x_vals[lb:ub] - points[:,1] = r[lb:ub] - - m = info.get_marker () - if m == 'line': - objects.append (plot.PolyLine (points, - colour=self.channel_color (ch), - legend=('Ch%d' % (ch+1,)))) - else: - objects.append (plot.PolyMarker (points, - marker=m, - colour=self.channel_color (ch), - legend=('Ch%d' % (ch+1,)))) - - graphics = plot.PlotGraphics (objects, - title=self.info.title, - xLabel = '', yLabel = '') - - time_per_div = info.get_time_per_div () - x_range = (-5.0 * time_per_div, 5.0 * time_per_div) # ranges are tuples! - volts_per_div = info.get_volts_per_div () - if not self.info.autorange: - self.y_range = (-4.0 * volts_per_div, 4.0 * volts_per_div) - self.Draw (graphics, xAxis=x_range, yAxis=self.y_range) - self.update_y_range () # autorange to self.y_range - - - def format_xy_data (self, evt): - info = self.info - records = evt.data - nchannels = len (records) - npoints = len (records[0]) - - if nchannels < 2: - return - - objects = [] - # points = zip (records[0], records[1]) - points = numpy.zeros ((len(records[0]), 2), numpy.float32) - points[:,0] = records[0] - points[:,1] = records[1] - - self.SetXUseScopeTicks (False) - - m = info.get_marker () - if m == 'line': - objects.append (plot.PolyLine (points, - colour=self.channel_color (0))) - else: - objects.append (plot.PolyMarker (points, - marker=m, - colour=self.channel_color (0))) - - graphics = plot.PlotGraphics (objects, - title=self.info.title, - xLabel = 'I', yLabel = 'Q') - - self.Draw (graphics, xAxis=self.x_range, yAxis=self.y_range) - self.update_y_range () - self.update_x_range () - - - def update_y_range (self): - alpha = 1.0/25 - graphics = self.last_draw[0] - p1, p2 = graphics.boundingBox () # min, max points of graphics - - if self.avg_y_min: # prevent vertical scale from jumping abruptly --? - self.avg_y_min = p1[1] * alpha + self.avg_y_min * (1 - alpha) - self.avg_y_max = p2[1] * alpha + self.avg_y_max * (1 - alpha) - else: # initial guess - self.avg_y_min = p1[1] # -500.0 workaround, sometimes p1 is ~ 10^35 - self.avg_y_max = p2[1] # 500.0 - - self.y_range = self._axisInterval ('auto', self.avg_y_min, self.avg_y_max) - # print "p1 %s p2 %s y_min %s y_max %s y_range %s" \ - # % (p1, p2, self.avg_y_min, self.avg_y_max, self.y_range) - - - def update_x_range (self): - alpha = 1.0/25 - graphics = self.last_draw[0] - p1, p2 = graphics.boundingBox () # min, max points of graphics - - if self.avg_x_min: - self.avg_x_min = p1[0] * alpha + self.avg_x_min * (1 - alpha) - self.avg_x_max = p2[0] * alpha + self.avg_x_max * (1 - alpha) - else: - self.avg_x_min = p1[0] - self.avg_x_max = p2[0] - - self.x_range = self._axisInterval ('auto', self.avg_x_min, self.avg_x_max) - - -# ---------------------------------------------------------------- -# Stand-alone test application -# ---------------------------------------------------------------- - -class test_app_flow_graph (stdgui.gui_flow_graph): - def __init__(self, frame, panel, vbox, argv): - stdgui.gui_flow_graph.__init__ (self, frame, panel, vbox, argv) - - if len(argv) > 1: - frame_decim = int(argv[1]) - else: - frame_decim = 1 - - if len(argv) > 2: - v_scale = float(argv[2]) # start up at this v_scale value - else: - v_scale = None # start up in autorange mode, default - - if len(argv) > 3: - t_scale = float(argv[3]) # start up at this t_scale value - else: - t_scale = None # old behavior - - print "frame decim %s v_scale %s t_scale %s" % (frame_decim,v_scale,t_scale) - - input_rate = 1e6 - - # Generate a complex sinusoid - src0 = gr.sig_source_c (input_rate, gr.GR_SIN_WAVE, 25.1e3, 1e3) - - # We add this throttle block so that this demo doesn't suck down - # all the CPU available. You normally wouldn't use it... - throttle = gr.throttle(gr.sizeof_gr_complex, input_rate) - - scope = scope_sink_c (self, panel,"Secret Data",sample_rate=input_rate, - frame_decim=frame_decim, - v_scale=v_scale, t_scale=t_scale) - vbox.Add (scope.win, 1, wx.EXPAND) - - # wire the blocks together - self.connect (src0, throttle, scope) - -def main (): - app = stdgui.stdapp (test_app_flow_graph, "O'Scope Test App") - app.MainLoop () - -if __name__ == '__main__': - main () - -# ---------------------------------------------------------------- diff --git a/gr-wxgui/src/python/stdgui.py b/gr-wxgui/src/python/stdgui.py deleted file mode 100644 index 178739547..000000000 --- a/gr-wxgui/src/python/stdgui.py +++ /dev/null @@ -1,90 +0,0 @@ -# -# Copyright 2004 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. -# - -'''A simple wx gui for GNU Radio applications''' - -import wx -import sys -from gnuradio import gr - - -class stdapp (wx.App): - def __init__ (self, flow_graph_maker, title="GNU Radio", nstatus=2): - self.flow_graph_maker = flow_graph_maker - self.title = title - self._nstatus = nstatus - # All our initialization must come before calling wx.App.__init__. - # OnInit is called from somewhere in the guts of __init__. - wx.App.__init__ (self, redirect=False) - - def OnInit (self): - frame = stdframe (self.flow_graph_maker, self.title, self._nstatus) - frame.Show (True) - self.SetTopWindow (frame) - return True - - -class stdframe (wx.Frame): - def __init__ (self, flow_graph_maker, title="GNU Radio", nstatus=2): - # print "stdframe.__init__" - wx.Frame.__init__(self, None, -1, title) - - self.CreateStatusBar (nstatus) - mainmenu = wx.MenuBar () - - menu = wx.Menu () - item = menu.Append (200, 'E&xit', 'Exit') - self.Bind (wx.EVT_MENU, self.OnCloseWindow, item) - mainmenu.Append (menu, "&File") - self.SetMenuBar (mainmenu) - - self.Bind (wx.EVT_CLOSE, self.OnCloseWindow) - self.panel = stdpanel (self, self, flow_graph_maker) - vbox = wx.BoxSizer(wx.VERTICAL) - vbox.Add(self.panel, 1, wx.EXPAND) - self.SetSizer(vbox) - self.SetAutoLayout(True) - vbox.Fit(self) - - def OnCloseWindow (self, event): - self.flow_graph().stop() - self.Destroy () - - def flow_graph (self): - return self.panel.fg - -class stdpanel (wx.Panel): - def __init__ (self, parent, frame, flow_graph_maker): - # print "stdpanel.__init__" - wx.Panel.__init__ (self, parent, -1) - self.frame = frame - - vbox = wx.BoxSizer (wx.VERTICAL) - self.fg = flow_graph_maker (frame, self, vbox, sys.argv) - self.SetSizer (vbox) - self.SetAutoLayout (True) - vbox.Fit (self) - - self.fg.start () - -class gui_flow_graph (gr.flow_graph): - def __init__ (self, *ignore): - gr.flow_graph.__init__ (self) diff --git a/gr-wxgui/src/python/waterfallsink.py b/gr-wxgui/src/python/waterfallsink.py deleted file mode 100755 index 4803cbeb2..000000000 --- a/gr-wxgui/src/python/waterfallsink.py +++ /dev/null @@ -1,475 +0,0 @@ -#!/usr/bin/env python -# -# Copyright 2003,2004,2005,2007 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. -# - -from gnuradio import gr, gru, window -from gnuradio.wxgui import stdgui -import wx -import gnuradio.wxgui.plot as plot -import numpy -import os -import threading -import math - -default_fftsink_size = (640,240) -default_fft_rate = gr.prefs().get_long('wxgui', 'fft_rate', 15) - -class waterfall_sink_base(object): - def __init__(self, input_is_real=False, baseband_freq=0, - sample_rate=1, fft_size=512, - fft_rate=default_fft_rate, - average=False, avg_alpha=None, title=''): - - # initialize common attributes - self.baseband_freq = baseband_freq - self.sample_rate = sample_rate - self.fft_size = fft_size - self.fft_rate = fft_rate - self.average = average - if avg_alpha is None: - self.avg_alpha = 2.0 / fft_rate - else: - self.avg_alpha = avg_alpha - self.title = title - self.input_is_real = input_is_real - self.msgq = gr.msg_queue(2) # queue up to 2 messages - - def set_average(self, average): - self.average = average - if average: - self.avg.set_taps(self.avg_alpha) - else: - self.avg.set_taps(1.0) - - def set_avg_alpha(self, avg_alpha): - self.avg_alpha = avg_alpha - - def set_baseband_freq(self, baseband_freq): - self.baseband_freq = baseband_freq - - def set_sample_rate(self, sample_rate): - self.sample_rate = sample_rate - self._set_n() - - def _set_n(self): - self.one_in_n.set_n(max(1, int(self.sample_rate/self.fft_size/self.fft_rate))) - -class waterfall_sink_f(gr.hier_block, waterfall_sink_base): - def __init__(self, fg, parent, baseband_freq=0, - y_per_div=10, ref_level=50, sample_rate=1, fft_size=512, - fft_rate=default_fft_rate, average=False, avg_alpha=None, - title='', size=default_fftsink_size): - - waterfall_sink_base.__init__(self, input_is_real=True, baseband_freq=baseband_freq, - sample_rate=sample_rate, fft_size=fft_size, - fft_rate=fft_rate, - average=average, avg_alpha=avg_alpha, title=title) - - s2p = gr.serial_to_parallel(gr.sizeof_float, self.fft_size) - self.one_in_n = gr.keep_one_in_n(gr.sizeof_float * self.fft_size, - max(1, int(self.sample_rate/self.fft_size/self.fft_rate))) - mywindow = window.blackmanharris(self.fft_size) - fft = gr.fft_vfc(self.fft_size, True, mywindow) - c2mag = gr.complex_to_mag(self.fft_size) - self.avg = gr.single_pole_iir_filter_ff(1.0, self.fft_size) - log = gr.nlog10_ff(20, self.fft_size, -20*math.log10(self.fft_size)) - sink = gr.message_sink(gr.sizeof_float * self.fft_size, self.msgq, True) - - fg.connect (s2p, self.one_in_n, fft, c2mag, self.avg, log, sink) - gr.hier_block.__init__(self, fg, s2p, sink) - - self.win = waterfall_window(self, parent, size=size) - self.set_average(self.average) - - -class waterfall_sink_c(gr.hier_block, waterfall_sink_base): - def __init__(self, fg, parent, baseband_freq=0, - y_per_div=10, ref_level=50, sample_rate=1, fft_size=512, - fft_rate=default_fft_rate, average=False, avg_alpha=None, - title='', size=default_fftsink_size): - - waterfall_sink_base.__init__(self, input_is_real=False, baseband_freq=baseband_freq, - sample_rate=sample_rate, fft_size=fft_size, - fft_rate=fft_rate, - average=average, avg_alpha=avg_alpha, title=title) - - s2p = gr.serial_to_parallel(gr.sizeof_gr_complex, self.fft_size) - self.one_in_n = gr.keep_one_in_n(gr.sizeof_gr_complex * self.fft_size, - max(1, int(self.sample_rate/self.fft_size/self.fft_rate))) - - mywindow = window.blackmanharris(self.fft_size) - fft = gr.fft_vcc(self.fft_size, True, mywindow) - c2mag = gr.complex_to_mag(self.fft_size) - self.avg = gr.single_pole_iir_filter_ff(1.0, self.fft_size) - log = gr.nlog10_ff(20, self.fft_size, -20*math.log10(self.fft_size)) - sink = gr.message_sink(gr.sizeof_float * self.fft_size, self.msgq, True) - - fg.connect(s2p, self.one_in_n, fft, c2mag, self.avg, log, sink) - gr.hier_block.__init__(self, fg, s2p, sink) - - self.win = waterfall_window(self, parent, size=size) - self.set_average(self.average) - - -# ------------------------------------------------------------------------ - -myDATA_EVENT = wx.NewEventType() -EVT_DATA_EVENT = wx.PyEventBinder (myDATA_EVENT, 0) - - -class DataEvent(wx.PyEvent): - def __init__(self, data): - wx.PyEvent.__init__(self) - self.SetEventType (myDATA_EVENT) - self.data = data - - def Clone (self): - self.__class__ (self.GetId()) - - -class input_watcher (threading.Thread): - def __init__ (self, msgq, fft_size, event_receiver, **kwds): - threading.Thread.__init__ (self, **kwds) - self.setDaemon (1) - self.msgq = msgq - self.fft_size = fft_size - self.event_receiver = event_receiver - self.keep_running = True - self.start () - - def run (self): - while (self.keep_running): - msg = self.msgq.delete_head() # blocking read of message queue - itemsize = int(msg.arg1()) - nitems = int(msg.arg2()) - - s = msg.to_string() # get the body of the msg as a string - - # There may be more than one FFT frame in the message. - # If so, we take only the last one - if nitems > 1: - start = itemsize * (nitems - 1) - s = s[start:start+itemsize] - - complex_data = numpy.fromstring (s, numpy.float32) - de = DataEvent (complex_data) - wx.PostEvent (self.event_receiver, de) - del de - - -class waterfall_window (wx.Panel): - def __init__ (self, fftsink, parent, id = -1, - pos = wx.DefaultPosition, size = wx.DefaultSize, - style = wx.DEFAULT_FRAME_STYLE, name = ""): - wx.Panel.__init__(self, parent, id, pos, size, style, name) - - self.fftsink = fftsink - self.bm = wx.EmptyBitmap(self.fftsink.fft_size, 300, -1) - - self.scale_factor = 5.0 # FIXME should autoscale, or set this - - dc1 = wx.MemoryDC() - dc1.SelectObject(self.bm) - dc1.Clear() - - self.pens = self.make_pens() - - wx.EVT_PAINT( self, self.OnPaint ) - wx.EVT_CLOSE (self, self.on_close_window) - EVT_DATA_EVENT (self, self.set_data) - - self.build_popup_menu() - - wx.EVT_CLOSE (self, self.on_close_window) - self.Bind(wx.EVT_RIGHT_UP, self.on_right_click) - - self.input_watcher = input_watcher(fftsink.msgq, fftsink.fft_size, self) - - - def on_close_window (self, event): - print "waterfall_window: on_close_window" - self.keep_running = False - - def const_list(self,const,len): - return [const] * len - - def make_colormap(self): - r = [] - r.extend(self.const_list(0,96)) - r.extend(range(0,255,4)) - r.extend(self.const_list(255,64)) - r.extend(range(255,128,-4)) - - g = [] - g.extend(self.const_list(0,32)) - g.extend(range(0,255,4)) - g.extend(self.const_list(255,64)) - g.extend(range(255,0,-4)) - g.extend(self.const_list(0,32)) - - b = range(128,255,4) - b.extend(self.const_list(255,64)) - b.extend(range(255,0,-4)) - b.extend(self.const_list(0,96)) - return (r,g,b) - - def make_pens(self): - (r,g,b) = self.make_colormap() - pens = [] - for i in range(0,256): - colour = wx.Colour(r[i], g[i], b[i]) - pens.append( wx.Pen(colour, 2, wx.SOLID)) - return pens - - def OnPaint(self, event): - dc = wx.PaintDC(self) - self.DoDrawing(dc) - - def DoDrawing(self, dc=None): - if dc is None: - dc = wx.ClientDC(self) - dc.DrawBitmap(self.bm, 0, 0, False ) - - - def const_list(self,const,len): - a = [const] - for i in range(1,len): - a.append(const) - return a - - def make_colormap(self): - r = [] - r.extend(self.const_list(0,96)) - r.extend(range(0,255,4)) - r.extend(self.const_list(255,64)) - r.extend(range(255,128,-4)) - - g = [] - g.extend(self.const_list(0,32)) - g.extend(range(0,255,4)) - g.extend(self.const_list(255,64)) - g.extend(range(255,0,-4)) - g.extend(self.const_list(0,32)) - - b = range(128,255,4) - b.extend(self.const_list(255,64)) - b.extend(range(255,0,-4)) - b.extend(self.const_list(0,96)) - return (r,g,b) - - def set_data (self, evt): - dB = evt.data - L = len (dB) - - dc1 = wx.MemoryDC() - dc1.SelectObject(self.bm) - dc1.Blit(0,1,self.fftsink.fft_size,300,dc1,0,0,wx.COPY,False,-1,-1) - - x = max(abs(self.fftsink.sample_rate), abs(self.fftsink.baseband_freq)) - if x >= 1e9: - sf = 1e-9 - units = "GHz" - elif x >= 1e6: - sf = 1e-6 - units = "MHz" - else: - sf = 1e-3 - units = "kHz" - - - if self.fftsink.input_is_real: # only plot 1/2 the points - d_max = L/2 - p_width = 2 - else: - d_max = L/2 - p_width = 1 - - scale_factor = self.scale_factor - if self.fftsink.input_is_real: # real fft - for x_pos in range(0, d_max): - value = int(dB[x_pos] * scale_factor) - value = min(255, max(0, value)) - dc1.SetPen(self.pens[value]) - dc1.DrawRectangle(x_pos*p_width, 0, p_width, 1) - else: # complex fft - for x_pos in range(0, d_max): # positive freqs - value = int(dB[x_pos] * scale_factor) - value = min(255, max(0, value)) - dc1.SetPen(self.pens[value]) - dc1.DrawRectangle(x_pos*p_width + d_max, 0, p_width, 1) - for x_pos in range(0 , d_max): # negative freqs - value = int(dB[x_pos+d_max] * scale_factor) - value = min(255, max(0, value)) - dc1.SetPen(self.pens[value]) - dc1.DrawRectangle(x_pos*p_width, 0, p_width, 1) - - self.DoDrawing (None) - - def on_average(self, evt): - # print "on_average" - self.fftsink.set_average(evt.IsChecked()) - - def on_right_click(self, event): - menu = self.popup_menu - for id, pred in self.checkmarks.items(): - item = menu.FindItemById(id) - item.Check(pred()) - self.PopupMenu(menu, event.GetPosition()) - - - def build_popup_menu(self): - self.id_incr_ref_level = wx.NewId() - self.id_decr_ref_level = wx.NewId() - self.id_incr_y_per_div = wx.NewId() - self.id_decr_y_per_div = wx.NewId() - self.id_y_per_div_1 = wx.NewId() - self.id_y_per_div_2 = wx.NewId() - self.id_y_per_div_5 = wx.NewId() - self.id_y_per_div_10 = wx.NewId() - self.id_y_per_div_20 = wx.NewId() - self.id_average = wx.NewId() - - self.Bind(wx.EVT_MENU, self.on_average, id=self.id_average) - #self.Bind(wx.EVT_MENU, self.on_incr_ref_level, id=self.id_incr_ref_level) - #self.Bind(wx.EVT_MENU, self.on_decr_ref_level, id=self.id_decr_ref_level) - #self.Bind(wx.EVT_MENU, self.on_incr_y_per_div, id=self.id_incr_y_per_div) - #self.Bind(wx.EVT_MENU, self.on_decr_y_per_div, id=self.id_decr_y_per_div) - #self.Bind(wx.EVT_MENU, self.on_y_per_div, id=self.id_y_per_div_1) - #self.Bind(wx.EVT_MENU, self.on_y_per_div, id=self.id_y_per_div_2) - #self.Bind(wx.EVT_MENU, self.on_y_per_div, id=self.id_y_per_div_5) - #self.Bind(wx.EVT_MENU, self.on_y_per_div, id=self.id_y_per_div_10) - #self.Bind(wx.EVT_MENU, self.on_y_per_div, id=self.id_y_per_div_20) - - - # make a menu - menu = wx.Menu() - self.popup_menu = menu - menu.AppendCheckItem(self.id_average, "Average") - # menu.Append(self.id_incr_ref_level, "Incr Ref Level") - # menu.Append(self.id_decr_ref_level, "Decr Ref Level") - # menu.Append(self.id_incr_y_per_div, "Incr dB/div") - # menu.Append(self.id_decr_y_per_div, "Decr dB/div") - # menu.AppendSeparator() - # we'd use RadioItems for these, but they're not supported on Mac - #menu.AppendCheckItem(self.id_y_per_div_1, "1 dB/div") - #menu.AppendCheckItem(self.id_y_per_div_2, "2 dB/div") - #menu.AppendCheckItem(self.id_y_per_div_5, "5 dB/div") - #menu.AppendCheckItem(self.id_y_per_div_10, "10 dB/div") - #menu.AppendCheckItem(self.id_y_per_div_20, "20 dB/div") - - self.checkmarks = { - self.id_average : lambda : self.fftsink.average - #self.id_y_per_div_1 : lambda : self.fftsink.y_per_div == 1, - #self.id_y_per_div_2 : lambda : self.fftsink.y_per_div == 2, - #self.id_y_per_div_5 : lambda : self.fftsink.y_per_div == 5, - #self.id_y_per_div_10 : lambda : self.fftsink.y_per_div == 10, - #self.id_y_per_div_20 : lambda : self.fftsink.y_per_div == 20, - } - - -def next_up(v, seq): - """ - Return the first item in seq that is > v. - """ - for s in seq: - if s > v: - return s - return v - -def next_down(v, seq): - """ - Return the last item in seq that is < v. - """ - rseq = list(seq[:]) - rseq.reverse() - - for s in rseq: - if s < v: - return s - return v - - -# ---------------------------------------------------------------- -# Deprecated interfaces -# ---------------------------------------------------------------- - -# returns (block, win). -# block requires a single input stream of float -# win is a subclass of wxWindow - -def make_waterfall_sink_f(fg, parent, title, fft_size, input_rate): - - block = waterfall_sink_f(fg, parent, title=title, fft_size=fft_size, - sample_rate=input_rate) - return (block, block.win) - -# returns (block, win). -# block requires a single input stream of gr_complex -# win is a subclass of wxWindow - -def make_waterfall_sink_c(fg, parent, title, fft_size, input_rate): - block = waterfall_sink_c(fg, parent, title=title, fft_size=fft_size, - sample_rate=input_rate) - return (block, block.win) - - -# ---------------------------------------------------------------- -# Standalone test app -# ---------------------------------------------------------------- - -class test_app_flow_graph (stdgui.gui_flow_graph): - def __init__(self, frame, panel, vbox, argv): - stdgui.gui_flow_graph.__init__ (self, frame, panel, vbox, argv) - - fft_size = 512 - - # build our flow graph - input_rate = 20.000e3 - - # Generate a complex sinusoid - src1 = gr.sig_source_c (input_rate, gr.GR_SIN_WAVE, 5.75e3, 1000) - #src1 = gr.sig_source_c (input_rate, gr.GR_CONST_WAVE, 5.75e3, 1000) - - # We add these throttle blocks so that this demo doesn't - # suck down all the CPU available. Normally you wouldn't use these. - thr1 = gr.throttle(gr.sizeof_gr_complex, input_rate) - - sink1 = waterfall_sink_c (self, panel, title="Complex Data", fft_size=fft_size, - sample_rate=input_rate, baseband_freq=100e3) - vbox.Add (sink1.win, 1, wx.EXPAND) - self.connect (src1, thr1, sink1) - - # generate a real sinusoid - src2 = gr.sig_source_f (input_rate, gr.GR_SIN_WAVE, 5.75e3, 1000) - #src2 = gr.sig_source_f (input_rate, gr.GR_CONST_WAVE, 5.75e3, 1000) - thr2 = gr.throttle(gr.sizeof_float, input_rate) - sink2 = waterfall_sink_f (self, panel, title="Real Data", fft_size=fft_size, - sample_rate=input_rate, baseband_freq=100e3) - vbox.Add (sink2.win, 1, wx.EXPAND) - self.connect (src2, thr2, sink2) - -def main (): - app = stdgui.stdapp (test_app_flow_graph, - "Waterfall Sink Test App") - app.MainLoop () - -if __name__ == '__main__': - main () |