#!/usr/bin/env python # # Copyright 2003,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 2, 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 from gnuradio.wxgui import stdgui,peakfind import wx import wx.lib.plot as plot import Numeric import os import threading # FIXME this should be rewritten to use hierarchical modules (when they're ready) # ======================================================================== # returns (block, win). # block requires a single input stream of float # win is a subclass of wxWindow def make_fft_sink_f (fg, parent, label, fft_size, input_rate,show_peak_info=False,show_peak_markers=False,navg_frames=100,peak_searchwidth=-1): #show_peak_info (default=False): if set to true the fft_window will show the frequency and level of every big peak found in the spectrum #show_peak_markers (default=False): if set to true the fft_window will show a marker at every big peak found in the spectrum #navg_frames (default=100): number of fft frames to average before showing it. If set to 100 or higher you get a much quiter spectrum (less noisy and not changing very much every frame) #peak_search_width(-1 gets you the default of 10 for a fft_size of 1024): number of neigbours the peakfinding algorithm will look at for determing if something is a peak. # A higher value means you get less peaks detected #Default the peakfinding algorithm determines a minimum signalstrength threshold below which no peaks are detected # you can override this with set_autopeakthreshold(False) and set_peakthreshold(mythreshold) to use your own threshold (r_fd, w_fd) = os.pipe () fft_rate = 20 if(navg_frames>1): fft_rate = fft_rate*navg_frames s2p = gr.serial_to_parallel (gr.sizeof_float, fft_size) decimate=int (input_rate / fft_size / fft_rate) if(decimate<1): decimate=1 #fft_rate=input_rate/fft_size/decimate one_in_n = gr.keep_one_in_n (gr.sizeof_float * fft_size, decimate) fft = gr.fft_vfc (fft_size, True, True) dst = gr.file_descriptor_sink (gr.sizeof_gr_complex * fft_size, w_fd) fg.connect (s2p, one_in_n) fg.connect (one_in_n, fft) fg.connect (fft, dst) block = s2p # head of pipeline win = fft_window (fft_info (r_fd, fft_size, input_rate, True, label,show_peak_info,show_peak_markers,navg_frames,peak_searchwidth), parent) return (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, label, fft_size, input_rate,show_peak_info=False,show_peak_markers=False,navg_frames=100,peak_searchwidth=-1): (r_fd, w_fd) = os.pipe () fft_rate = 20 if(navg_frames>1): fft_rate = fft_rate*navg_frames decimate=int (input_rate / fft_size / fft_rate) if(decimate<1): decimate=1 s2p = gr.serial_to_parallel (gr.sizeof_gr_complex, fft_size) one_in_n = gr.keep_one_in_n (gr.sizeof_gr_complex * fft_size, decimate) fft = gr.fft_vcc (fft_size, True, True) dst = gr.file_descriptor_sink (gr.sizeof_gr_complex * fft_size, w_fd) fg.connect (s2p, one_in_n) fg.connect (one_in_n, fft) fg.connect (fft, dst) block = s2p # head of pipeline win = fft_window (fft_info (r_fd, fft_size, input_rate, False, label,show_peak_info,show_peak_markers,navg_frames,peak_searchwidth), parent) return (block, win) # ------------------------------------------------------------------------ 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 fft_info: def __init__ (self, file_descriptor, fft_size, sample_rate, input_is_real, title = "fft",show_peak_info=False,show_peak_markers=False,navg_frames=100,peak_searchwidth=-1): self.file_descriptor = file_descriptor self.fft_size = fft_size self.sample_rate = sample_rate self.input_is_real = input_is_real self.title = title; self.show_peak_info=show_peak_info self.show_peak_markers=show_peak_markers self.navg_frames = navg_frames self.peak_searchwidth = peak_searchwidth class input_watcher (threading.Thread): def __init__ (self, file_descriptor, fft_size, navg_frames,event_receiver, **kwds): threading.Thread.__init__ (self, **kwds) self.setDaemon (1) self.file_descriptor = file_descriptor self.fft_size = fft_size self.event_receiver = event_receiver self.keep_running = True self.navg_frames = navg_frames self.start () def run (self): # print "input_watcher: pid = ", os.getpid () while (self.keep_running): s = os.read (self.file_descriptor, gr.sizeof_gr_complex * self.fft_size) if not s: self.keep_running = False break mag_data = abs(Numeric.fromstring (s, Numeric.Complex32)) if(self.navg_frames>1): for i in range(1,self.navg_frames): s = os.read (self.file_descriptor, gr.sizeof_gr_complex * self.fft_size) if not s: self.keep_running = False break mag_data = mag_data + abs(Numeric.fromstring (s, Numeric.Complex32)) de = DataEvent (mag_data) wx.PostEvent (self.event_receiver, de) # print "run: len(complex_data) = ", len(complex_data) del de class fft_window (plot.PlotCanvas): def __init__ (self, info, 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.SetEnableGrid (True) self.SetEnableZoom (True) # self.SetBackgroundColour ('black') self.info = info; self.y_range = None self.avg_y_min = None self.avg_y_max = None self.navg_frames=info.navg_frames #number of fft's to average over EVT_DATA_EVENT (self, self.set_data) wx.EVT_CLOSE (self, self.on_close_window) self.input_watcher = input_watcher (info.file_descriptor, info.fft_size , self.navg_frames, self) self.peakfinder=peakfind.SimplePeakFinder() self.peakthreshold=0.0 self.peak_searchwidth=info.peak_searchwidth #searchspace in x-direction of frequencies that will be checked to be lower than the peak. Can be 1 to fftsize if(self.peak_searchwidth<=0): self.peak_searchwidth=10*1024/info.fft_size #default to inspect 10 neigbours at fft_size 1024, scale accordingly if other fft_size is used info.peak_searchwidth=self.peak_searchwidth self.show_peak_markers=info.show_peak_markers self.show_peak_info=info.show_peak_info self.autopeakthreshold=True #default if finds its own lower threshold for peaks #self.SetFontSizeLegend(1) def set_peak_searchwidth(peak_searchwidth): self.peak_searchwidth=peak_searchwidth def get_peak_searchwidth(): return self.peak_searchwidth def set_autopeakthreshold(autopeakthreshold): self.autopeakthreshold=autopeakthreshold def get_autopeakthreshold(): return self.autopeakthreshold def set_peakthreshold(peakthreshold): self.autopeakthreshold=False self.peakthreshold=peakthreshold def get_peakthreshold(): return peakthreshold def on_close_window (self, event): print "fft_window:on_close_window" self.keep_running = False def set_data (self, evt): data = evt.data #dB = 20 * Numeric.log10 (abs(data) + 1e-8) dB = 20 * Numeric.log10 (data + 1e-8) - 20*Numeric.log10(self.navg_frames) l = len (dB) if self.info.sample_rate >= 1e6: sf = 1e-6 units = "MHz" else: sf = 1e-3 units = "kHz" plot_objects=[] if self.info.input_is_real: # only plot 1/2 the points x_vals = Numeric.arrayrange (l/2) * (self.info.sample_rate * sf / l) y_vals = dB[0:l/2] else: # the "negative freqs" are in the second half of the array x_vals = Numeric.arrayrange (-l/2+1, l/2) * (self.info.sample_rate * sf / l) y_vals = Numeric.concatenate ((dB[l/2:], dB[0:l/2])) points = zip (x_vals, y_vals) lines = plot.PolyLine (points, colour='LIME GREEN') self.current_lines=lines plot_objects.append(lines) if(self.show_peak_info | self.show_peak_markers): peaks = self.peakfinder.getPeaks(self.peak_searchwidth,y_vals,self.peakthreshold,False) peak_points=[] for i in peaks: peak_points.append([x_vals[i],y_vals[i]]) peaktext_x_offset=0 if(self.show_peak_markers): plot_objects.append(plot.PolyMarker(peak_points,legend='peak')) peaktext_x_offset=10 graphics = plot.PlotGraphics (plot_objects, title=self.info.title, xLabel = units, yLabel = "dB") self.Draw (graphics, xAxis=None, yAxis=self.y_range) mydc=wx.ClientDC(self) h = mydc.GetCharHeight() if(self.show_peak_info): for pt in peak_points: label='%5.4f ' % pt[0] + units + ' %4.1f dB' % pt[1] ptscaled=Numeric.array(pt)*self._pointScale+self._pointShift mydc.DrawText(label,ptscaled[0]+peaktext_x_offset,ptscaled[1]-0.5*h) self.update_y_range () def update_y_range (self): alpha = 1.0/25 graphics = self.last_draw[0] #p1, p2 = graphics.boundingBox () # min, max points of graphics p1,p2 = self.current_lines.boundingBox() # min, max points of graphics if self.avg_y_min: 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: self.avg_y_min = p1[1] self.avg_y_max = p2[1] if(self.autopeakthreshold): self.peakthreshold=(8*self.avg_y_min+2*self.avg_y_max)/10 self.y_range = self._axisInterval ('auto', self.avg_y_min, self.avg_y_max) # ---------------------------------------------------------------- # 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) # build our flow graph input_rate = 20.0001234e6 src = gr.sig_source_c (input_rate, gr.GR_SIN_WAVE, 5.75e6, 10e3) block, fft_win = make_fft_sink_c (self, panel, "Secret Data", 512, input_rate,True,False,100) self.connect (src, block) vbox.Add (fft_win, 1, wx.EXPAND) def main (): app = stdgui.stdapp (test_app_flow_graph, "FFT Sink Test App") app.MainLoop () if __name__ == '__main__': main () # ----------------------------------------------------------------