scopesink2.py

这是用python语言写的一个数字广播的信号处理工具包。利用它

        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_top_block (stdgui2.std_top_block):
    def __init__(self, frame, panel, vbox, argv):
        stdgui2.std_top_block.__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
        self.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...
        self.thr = gr.throttle(gr.sizeof_gr_complex, input_rate)

        scope = scope_sink_c (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)

        # Ultimately this will be
        # self.connect("src0 throttle scope")
	self.connect(self.src0, self.thr, scope) 

def main ():
    app = stdgui2.stdapp (test_top_block, "O'Scope Test App")
    app.MainLoop ()

if __name__ == '__main__':
    main ()

# ----------------------------------------------------------------