2 # Copyright 2008 Free Software Foundation, Inc.
4 # This file is part of GNU Radio
6 # GNU Radio is free software; you can redistribute it and/or modify
7 # it under the terms of the GNU General Public License as published by
8 # the Free Software Foundation; either version 3, or (at your option)
11 # GNU Radio is distributed in the hope that it will be useful,
12 # but WITHOUT ANY WARRANTY; without even the implied warranty of
13 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
14 # GNU General Public License for more details.
16 # You should have received a copy of the GNU General Public License
17 # along with GNU Radio; see the file COPYING. If not, write to
18 # the Free Software Foundation, Inc., 51 Franklin Street,
19 # Boston, MA 02110-1301, USA.
22 ##################################################
23 # conditional disconnections of wx flow graph
24 ##################################################
27 def bind_to_visible_event(win, callback):
29 Bind a callback to a window when its visibility changes.
30 Specifically, callback when the window changes visibility
31 when a notebook tab event in one of the parents occurs.
32 @param win the wx window
33 @param callback a 1 param function
35 #is the window visible in the hierarchy
36 def is_wx_window_visible(my_win):
38 parent = my_win.GetParent()
39 if not parent: return True #reached the top of the hierarchy
40 #if we are hidden, then finish, otherwise keep traversing up
41 if isinstance(parent, wx.Notebook) and parent.GetCurrentPage() != my_win: return False
43 #call the callback, the arg is shown or not
44 def callback_factory(my_win, my_callback):
45 return lambda *args: my_callback(is_wx_window_visible(my_win))
46 handler = callback_factory(win, callback)
47 #bind the handler to all the parent notebooks
49 if isinstance(win, wx.Notebook):
50 win.Bind(wx.EVT_NOTEBOOK_PAGE_CHANGED, handler)
51 if not win.GetParent():
52 win.Bind(wx.EVT_ACTIVATE, handler)
55 from gnuradio import gr
57 def conditional_connect(source, sink, hb, win, size):
60 def callback(visible, init=False):
61 if visible == cache[0]: return
63 if not init: hb.lock()
64 print 'visible', visible, source, sink
67 hb.disconnect(source, nulls[0])
68 hb.disconnect(nulls[1], nulls[2])
69 hb.disconnect(nulls[2], sink)
70 while nulls: nulls.pop()
71 hb.connect(source, sink)
73 if not init: hb.disconnect(source, sink)
74 nulls.extend([gr.null_sink(size), gr.null_source(size), gr.head(size, 0)])
75 hb.connect(source, nulls[0])
76 hb.connect(nulls[1], nulls[2], sink)
77 if not init: hb.unlock()
78 callback(False, init=True) #initially connect
79 bind_to_visible_event(win, callback)
81 class wxgui_hb(object):
82 def wxgui_connect(self, *points):
84 Use wxgui connect when the first point is the self source of the hb.
85 The win property of this object should be set to the wx window.
86 When this method tries to connect self to the next point,
87 it will conditionally make this connection based on the visibility state.
90 assert points[0] == self or points[0][0] == self
93 win=self.win, hb=self,
94 size=self._hb.input_signature().sizeof_stream_item(0),
96 if len(points[1:]) > 1: self.connect(*points[1:])
97 except (AssertionError, IndexError): self.connect(*points)
99 #A macro to apply an index to a key
100 index_key = lambda key, i: "%s_%d"%(key, i+1)
102 def _register_access_method(destination, controller, key):
104 Helper function for register access methods.
105 This helper creates distinct set and get methods for each key
106 and adds them to the destination object.
108 def set(value): controller[key] = value
109 setattr(destination, 'set_'+key, set)
110 def get(): return controller[key]
111 setattr(destination, 'get_'+key, get)
113 def register_access_methods(destination, controller):
115 Register setter and getter functions in the destination object for all keys in the controller.
116 @param destination the object to get new setter and getter methods
117 @param controller the pubsub controller
119 for key in controller.keys(): _register_access_method(destination, controller, key)
121 ##################################################
122 # Input Watcher Thread
123 ##################################################
124 from gnuradio import gru
126 class input_watcher(gru.msgq_runner):
128 Input watcher thread runs forever.
129 Read messages from the message queue.
130 Forward messages to the message handler.
132 def __init__ (self, msgq, controller, msg_key, arg1_key='', arg2_key=''):
133 self._controller = controller
134 self._msg_key = msg_key
135 self._arg1_key = arg1_key
136 self._arg2_key = arg2_key
137 gru.msgq_runner.__init__(self, msgq, self.handle_msg)
139 def handle_msg(self, msg):
140 if self._arg1_key: self._controller[self._arg1_key] = msg.arg1()
141 if self._arg2_key: self._controller[self._arg2_key] = msg.arg2()
142 self._controller[self._msg_key] = msg.to_string()
145 ##################################################
147 ##################################################
153 Get the exponent of the number in base 10.
154 @param num the floating point number
155 @return the exponent as an integer
157 if num == 0: return 0
158 return int(math.floor(math.log10(abs(num))))
160 def get_clean_num(num):
162 Get the closest clean number match to num with bases 1, 2, 5.
163 @param num the number
164 @return the closest number
166 if num == 0: return 0
167 sign = num > 0 and 1 or -1
169 nums = numpy.array((1, 2, 5, 10))*(10**exp)
170 return sign*nums[numpy.argmin(numpy.abs(nums - abs(num)))]
172 def get_clean_incr(num):
174 Get the next higher clean number with bases 1, 2, 5.
175 @param num the number
176 @return the next higher number
178 num = get_clean_num(num)
180 coeff = int(round(num/10**exp))
190 def get_clean_decr(num):
192 Get the next lower clean number with bases 1, 2, 5.
193 @param num the number
194 @return the next lower number
196 num = get_clean_num(num)
198 coeff = int(round(num/10**exp))
208 def get_min_max(samples):
210 Get the minimum and maximum bounds for an array of samples.
211 @param samples the array of real values
212 @return a tuple of min, max
215 mean = numpy.average(samples)
216 rms = numpy.max([scale_factor*((numpy.sum((samples-mean)**2)/len(samples))**.5), .1])
219 return min_val, max_val
221 def get_min_max_fft(fft_samps):
223 Get the minimum and maximum bounds for an array of fft samples.
224 @param samples the array of real values
225 @return a tuple of min, max
227 #get the peak level (max of the samples)
228 peak_level = numpy.max(fft_samps)
229 #separate noise samples
230 noise_samps = numpy.sort(fft_samps)[:len(fft_samps)/2]
232 noise_floor = numpy.average(noise_samps)
233 #get the noise deviation
234 noise_dev = numpy.std(noise_samps)
235 #determine the maximum and minimum levels
236 max_level = peak_level
237 min_level = noise_floor - abs(2*noise_dev)
238 return min_level, max_level