git.gag.com Git - debian/gnuradio/blob - gr-wxgui/src/python/common.py

   1 #
   2 # Copyright 2008 Free Software Foundation, Inc.
   3 #
   4 # This file is part of GNU Radio
   5 #
   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)
   9 # any later version.
  10 #
  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.
  15 #
  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.
  20 #
  21
  22 ##################################################
  23 # conditional disconnections of wx flow graph
  24 ##################################################
  25 import wx
  26 from gnuradio import gr
  27
  28 class wxgui_hb(object):
  29         """
  30         The wxgui hier block helper/wrapper class:
  31         A hier block should inherit from this class to make use of the wxgui connect method.
  32         To use, call wxgui_connect in place of regular connect; self.win must be defined.
  33         The implementation will conditionally connect or disconnect the self (source) of the hb.
  34         This condition depends on weather or not the window is visible with the parent notebooks.
  35         This condition will be re-checked on every ui update event.
  36         """
  37
  38         def wxgui_connect(self, *points):
  39                 """
  40                 Use wxgui connect when the first point is the self source of the hb.
  41                 The win property of this object should be set to the wx window.
  42                 When this method tries to connect self to the next point,
  43                 it will conditionally make this connection based on the visibility state.
  44                 All other points will be connected normally.
  45                 """
  46                 try:
  47                         assert points[0] == self or points[0][0] == self
  48                         self._conditional_connect(points[0], points[1])
  49                         if len(points[1:]) > 1: self.connect(*points[1:])
  50                 except (AssertionError, IndexError): self.connect(*points)
  51
  52         def _conditional_connect(self, source, sink):
  53                 """
  54                 Create a handler for visibility changes.
  55                 Initially call the handler to setup the fg.
  56                 Bind the handler to the visibility meta event.
  57                 """
  58                 handler = self._conditional_connect_handler_factory(
  59                         source=source, sink=sink, win=self.win, hb=self,
  60                         size=self._hb.input_signature().sizeof_stream_item(0),
  61                 )
  62                 handler(False, init=True) #initially connect
  63                 self._bind_to_visible_event(win=self.win, handler=handler)
  64
  65         @staticmethod
  66         def _conditional_connect_handler_factory(source, sink, hb, win, size):
  67                 """
  68                 Create a function that will handle the re-connections based on a flag.
  69                 The current state of the connection is stored in the namespace.
  70                 """
  71                 nulls = list()
  72                 cache = [None]
  73                 def callback(visible, init=False):
  74                         if visible == cache[0]: return
  75                         cache[0] = visible
  76                         if not init: hb.lock()
  77                         #print 'visible', visible, source, sink
  78                         if visible:
  79                                 if not init:
  80                                         hb.disconnect(source, nulls[0])
  81                                         hb.disconnect(nulls[1], nulls[2])
  82                                         hb.disconnect(nulls[2], sink)
  83                                         while nulls: nulls.pop()
  84                                 hb.connect(source, sink)
  85                         else:
  86                                 if not init: hb.disconnect(source, sink)
  87                                 nulls.extend([gr.null_sink(size), gr.null_source(size), gr.head(size, 0)])
  88                                 hb.connect(source, nulls[0])
  89                                 hb.connect(nulls[1], nulls[2], sink)
  90                         if not init: hb.unlock()
  91                 return callback
  92
  93         @staticmethod
  94         def _bind_to_visible_event(win, handler):
  95                 """
  96                 Bind a handler to a window when its visibility changes.
  97                 Specifically, call the handler when the window visibility changes.
  98                 This condition is checked on every update ui event.
  99                 @param win the wx window
 100                 @param handler a function of 1 param
 101                 """
 102                 #is the window visible in the hierarchy
 103                 def is_wx_window_visible(my_win):
 104                         while True:
 105                                 parent = my_win.GetParent()
 106                                 if not parent: return True #reached the top of the hierarchy
 107                                 #if we are hidden, then finish, otherwise keep traversing up
 108                                 if isinstance(parent, wx.Notebook) and parent.GetCurrentPage() != my_win: return False
 109                                 my_win = parent
 110                 #call the handler, the arg is shown or not
 111                 def handler_factory(my_win, my_handler):
 112                         return lambda *args: my_handler(is_wx_window_visible(my_win))
 113                 handler = handler_factory(win, handler)
 114                 #bind the handler to all the parent notebooks
 115                 win.Bind(wx.EVT_UPDATE_UI, handler)
 116
 117 ##################################################
 118 # Helpful Functions
 119 ##################################################
 120
 121 #A macro to apply an index to a key
 122 index_key = lambda key, i: "%s_%d"%(key, i+1)
 123
 124 def _register_access_method(destination, controller, key):
 125         """
 126         Helper function for register access methods.
 127         This helper creates distinct set and get methods for each key
 128         and adds them to the destination object.
 129         """
 130         def set(value): controller[key] = value
 131         setattr(destination, 'set_'+key, set)
 132         def get(): return controller[key]
 133         setattr(destination, 'get_'+key, get)
 134
 135 def register_access_methods(destination, controller):
 136         """
 137         Register setter and getter functions in the destination object for all keys in the controller.
 138         @param destination the object to get new setter and getter methods
 139         @param controller the pubsub controller
 140         """
 141         for key in controller.keys(): _register_access_method(destination, controller, key)
 142
 143 ##################################################
 144 # Input Watcher Thread
 145 ##################################################
 146 from gnuradio import gru
 147
 148 class input_watcher(gru.msgq_runner):
 149         """
 150         Input watcher thread runs forever.
 151         Read messages from the message queue.
 152         Forward messages to the message handler.
 153         """
 154         def __init__ (self, msgq, controller, msg_key, arg1_key='', arg2_key=''):
 155                 self._controller = controller
 156                 self._msg_key = msg_key
 157                 self._arg1_key = arg1_key
 158                 self._arg2_key = arg2_key
 159                 gru.msgq_runner.__init__(self, msgq, self.handle_msg)
 160
 161         def handle_msg(self, msg):
 162                 if self._arg1_key: self._controller[self._arg1_key] = msg.arg1()
 163                 if self._arg2_key: self._controller[self._arg2_key] = msg.arg2()
 164                 self._controller[self._msg_key] = msg.to_string()
 165
 166
 167 ##################################################
 168 # Shared Functions
 169 ##################################################
 170 import numpy
 171 import math
 172
 173 def get_exp(num):
 174         """
 175         Get the exponent of the number in base 10.
 176         @param num the floating point number
 177         @return the exponent as an integer
 178         """
 179         if num == 0: return 0
 180         return int(math.floor(math.log10(abs(num))))
 181
 182 def get_clean_num(num):
 183         """
 184         Get the closest clean number match to num with bases 1, 2, 5.
 185         @param num the number
 186         @return the closest number
 187         """
 188         if num == 0: return 0
 189         sign = num > 0 and 1 or -1
 190         exp = get_exp(num)
 191         nums = numpy.array((1, 2, 5, 10))*(10**exp)
 192         return sign*nums[numpy.argmin(numpy.abs(nums - abs(num)))]
 193
 194 def get_clean_incr(num):
 195         """
 196         Get the next higher clean number with bases 1, 2, 5.
 197         @param num the number
 198         @return the next higher number
 199         """
 200         num = get_clean_num(num)
 201         exp = get_exp(num)
 202         coeff = int(round(num/10**exp))
 203         return {
 204                 -5: -2,
 205                 -2: -1,
 206                 -1: -.5,
 207                 1: 2,
 208                 2: 5,
 209                 5: 10,
 210         }[coeff]*(10**exp)
 211
 212 def get_clean_decr(num):
 213         """
 214         Get the next lower clean number with bases 1, 2, 5.
 215         @param num the number
 216         @return the next lower number
 217         """
 218         num = get_clean_num(num)
 219         exp = get_exp(num)
 220         coeff = int(round(num/10**exp))
 221         return {
 222                 -5: -10,
 223                 -2: -5,
 224                 -1: -2,
 225                 1: .5,
 226                 2: 1,
 227                 5: 2,
 228         }[coeff]*(10**exp)
 229
 230 def get_min_max(samples):
 231         """
 232         Get the minimum and maximum bounds for an array of samples.
 233         @param samples the array of real values
 234         @return a tuple of min, max
 235         """
 236         scale_factor = 3
 237         mean = numpy.average(samples)
 238         rms = numpy.max([scale_factor*((numpy.sum((samples-mean)**2)/len(samples))**.5), .1])
 239         min_val = mean - rms
 240         max_val = mean + rms
 241         return min_val, max_val
 242
 243 def get_min_max_fft(fft_samps):
 244         """
 245         Get the minimum and maximum bounds for an array of fft samples.
 246         @param samples the array of real values
 247         @return a tuple of min, max
 248         """
 249         #get the peak level (max of the samples)
 250         peak_level = numpy.max(fft_samps)
 251         #separate noise samples
 252         noise_samps = numpy.sort(fft_samps)[:len(fft_samps)/2]
 253         #get the noise floor
 254         noise_floor = numpy.average(noise_samps)
 255         #get the noise deviation
 256         noise_dev = numpy.std(noise_samps)
 257         #determine the maximum and minimum levels
 258         max_level = peak_level
 259         min_level = noise_floor - abs(2*noise_dev)
 260         return min_level, max_level