diff options
Diffstat (limited to 'gnuradio-runtime/python/gnuradio/ctrlport/gr-ctrlport-curses')
| -rwxr-xr-x | gnuradio-runtime/python/gnuradio/ctrlport/gr-ctrlport-curses | 268 |
1 files changed, 268 insertions, 0 deletions
diff --git a/gnuradio-runtime/python/gnuradio/ctrlport/gr-ctrlport-curses b/gnuradio-runtime/python/gnuradio/ctrlport/gr-ctrlport-curses new file mode 100755 index 0000000000..1bee3b1a1e --- /dev/null +++ b/gnuradio-runtime/python/gnuradio/ctrlport/gr-ctrlport-curses @@ -0,0 +1,268 @@ +#!/usr/bin/env python +# +# Copyright 2012 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. +# + +import threading +import curses +import os, sys, time +from optparse import OptionParser + +import Ice +from gnuradio.ctrlport import GNURadio + +ENTER = chr(10) +UP_ARROW = chr(65) +DOWN_ARROW = chr(66) + +class modem_monitor(threading.Thread): + def __init__(self, cb_live, cb_exit, interface): + threading.Thread.__init__(self) + self.cb_live = cb_live + self.cb_exit = cb_exit + + self.running = True + + def __del__(self): + rx.close() + + def run(self): + while self.running: + time.sleep(0.5) + + def shutdown(self): + self.running = False + self.rx.close() + + def cb(self,contents): + (op, sep, payload) = contents.partition(":") + if(op == "live"): + print "live" + self.cb_live(payload) + elif(op == "exit"): + self.cb_exit(payload) + else: + print "unknown op arrived! garbage on multicast adx?" + +class modem_window(threading.Thread): + def __init__(self, locator): + threading.Thread.__init__(self) + self.locator = locator + + # curses init + self.win = curses.newwin(30,100,4,4) + + # Ice/GRCP init + self.comm = Ice.initialize() + proxy = self.comm.stringToProxy(locator) + self.radio = GNURadio.ControlPortPrx.checkedCast(proxy) + self.updateKnobs() + + # GUI init + self.running = True + self.ssel = 0 + self.start() + #self.updateGUI() + + # input loop + while(self.running): + self.getInput() + + # wait for update thread exit + self.join() + + def updateKnobs(self): + self.knobs = self.radio.get([]) + + def getInput(self): + a = self.win.getch() + if(a <= 256): + a = chr(a) + if(a == 'q'): + self.running = False + elif(a == UP_ARROW): + self.ssel = max(self.ssel-1, 0) + self.updateGUI() + elif(a == DOWN_ARROW): + self.ssel = max(min(self.ssel+1, len(self.knobs.keys())-1),0) + self.updateGUI() + self.updateGUI() + + def updateGUI(self): + self.win.clear() + self.win.border(0) + self.win.addstr(1, 2, "Modem Statistics :: %s"%(self.locator)) + self.win.addstr(2, 2, "---------------------------------------------------") + + maxnb = 0 + maxk = 0 + for k in self.knobs.keys(): + (nb,k) = k.split("::", 2) + maxnb = max(maxnb,len(nb)) + maxk = max(maxk,len(k)) + + offset = 3 + keys = self.knobs.keys() + keys.sort() + for k in keys: + (nb,sk) = k.split("::", 2) + v = self.knobs[k].value + sv = str(v) + if(len(sv) > 20): + sv = sv[0:20] + props = 0 + if(self.ssel == offset-3): + props = props | curses.A_REVERSE + self.win.addstr(offset, 2, "%s %s %s" % \ + (nb.rjust(maxnb," "), sk.ljust(maxk," "), sv),props) + offset = offset + 1 + self.win.refresh() + + def run(self): + while(self.running): + self.updateKnobs() + self.updateGUI() + time.sleep(1) + +class monitor_gui: + def __init__(self, interfaces, options): + + locator = None + + # Extract options into a locator + if(options.host and options.port): + locator = "{0} -t:{1} -h {2} -p {3}".format( + options.app, options.protocol, + options.host, options.port) + + # Set up GUI + self.locators = {} + + self.mode = 0 # modem index screen (modal keyboard input) + self.lsel = 0 # selected locator + self.scr = curses.initscr() + self.updateGUI() + + # Kick off initial monitors + self.monitors = [] + for i in interfaces: + self.monitors.append( modem_monitor(self.addModem, self.delModem, i) ) + self.monitors[-1].start() + + if not ((locator == None) or (locator == "")): + self.addModem(locator) + + # wait for user input + while(True): + self.getInput() + + def addModem(self, locator): + if(not self.locators.has_key(locator)): + self.locators[locator] = {} + self.locators[locator]["update_time"] = time.time() + self.locators[locator]["status"] = "live" + + self.updateGUI(); + + def delModem(self, locator): + #if(self.locators.has_key(locator)): + if(not self.locators.has_key(locator)): + self.locators[locator] = {} + self.locators[locator]["update_time"] = time.time() + self.locators[locator]["status"] = "exit" + + self.updateGUI() + + def updateGUI(self): + if(self.mode == 0): #redraw locators + self.scr.clear() + self.scr.border(0) + self.scr.addstr(1, 2, " GRCP-Curses Modem Monitor :: (A)dd (R)efresh, (Q)uit, ...") + for i in range(len(self.locators.keys())): + locator = self.locators.keys()[i] + lhash = self.locators[locator] + #self.scr.addstr(3+i, 5, locator + str(lhash)) + props = 0 + if(self.lsel == i): + props = props | curses.A_REVERSE + self.scr.addstr(3+i, 5, locator + str(lhash), props) + self.scr.refresh() + + def connectGUI(self): + self.scr.clear() + self.scr.addstr(1, 1, "Connect to radio:") + locator = self.scr.getstr(200) + self.addModem(locator) + self.updateGUI() + + def getInput(self): + a = self.scr.getch() + self.scr.addstr(20, 2, "got key (%d) " % (int(a))) + if(a <= 256): + a = chr(a) + if(a =='r'): + self.updateGUI() + elif(a == 'q'): + self.shutdown() + elif(a == 'a'): + self.connectGUI() + elif(a == UP_ARROW): + self.lsel = max(self.lsel-1, 0) + self.updateGUI() + elif(a == DOWN_ARROW): + self.lsel = max(min(self.lsel+1, len(self.locators.keys())-1),0) + self.updateGUI() + elif(a == ENTER): + try: + locator = self.locators.keys()[self.lsel] + self.mode = 1 + mwin = modem_window(locator) + self.mode = 0 + # pop up a new modem display ... + self.updateGUI() + except: + pass + + def shutdown(self): + curses.endwin() + os._exit(0) + +if __name__ == "__main__": + parser = OptionParser() + parser.add_option("-H", "--host", type="string", + help="Hostname of ControlPort server.") + parser.add_option("-p", "--port", type="int", + help="Port number of host's ControlPort endpoint.") + parser.add_option("-i", "--interfaces", type="string", + action="append", default=["lo"], + help="Interfaces to use. [default=%default]") + parser.add_option("-P", "--protocol", type="string", default="tcp", + help="Type of protocol to use (usually tcp). [default=%default]") + parser.add_option("-a", "--app", type="string", default="gnuradio", + help="Name of application [default=%default]") + (options, args) = parser.parse_args() + + if((options.host == None) ^ (options.port == None)): + print "Please set both a hostname and a port number.\n" + parser.print_help() + sys.exit(1) + + mg = monitor_gui(options.interfaces, options) + |