# PyNetMony Netmonitor for S60 3rd Edition phones # # Copyright (C) 2007 Carsten Knuetter, Georg Lukas # # email: netmonitor.s80@o2online.de # # This program 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 of the License, or (at your # option) any later version. This program 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 this program; if not, # see . import appuifw import e32 import location import sysinfo import time #import positioning import sys import audio import thread import camera import key_codes from audio import * from graphics import * try: import locationrequestor locreq = True gpsdata = [0, 'GPS disabled'] except: gpsdata = [0, 'Module locationrequestor not installed'] locreq = False SIS_VERSION = '0.3.9' TITLE=u'PyNetMony' VERSION=TITLE + ' ' + SIS_VERSION # temp variables running=True light=True tab=0 draw_rnc = False s = None finder = False # XXX: use absolute path for mp3 etc. # os.path.join(os.getcwd(), u'default_settings.xml') # program config configfile = "c:/system/apps/pynetmony.cfg" defconfig = { 'CellNotify': 0, 'Light': 0, 'Refresh': 0.5, 'Volume': 5, 'Voice': 0, 'GPS': 0 } config = defconfig.copy() # TODO: move these to theme[xxx] font=u"normal" color=(255,255,255) linecol=(255,128,128) #border=(255,0,0) bg=(0,0,60) border=bg headcol=(255,255,255) headbg=(255,255,255) # table for RX level history rxl_log = [] rxl_last = 0 # table for Cell history loc_log = [] logger = None logname = None toggle = 2 usinggps = 0 gpson = 0 # XXX workaround for E90 "losing" link during data traffic offlinetimeout = 90 # global GSM state variables t_last = 0 rxl = None gsmloc = gsmloc_real = None # Spelling for network providers class Voice: offline = u"Offline" band = [u"G S M", u"U M T S"] provider = { 262: { 1: "Timo Beil", 2: "Wo da phon", 3: "E plus", 7: "Oh tuh"} } class Logger: def __init__(self, log_name): self.logfile = log_name def write(self, obj): try: log_file = open(self.logfile, 'a') log_file.write(obj) log_file.close() except IOError, err: appuifw.note(u"Logger: " + unicode(err), "error") def writelines(self, obj): self.write(''.join(list)) def logtab(self, list): # Convert list to unicode and add tabs txt = '\t'.join(map(unicode, list)) + '\n' self.write(txt) def flush(self): pass class Setup( object ): # Config constants SOUND_OFF = 0 SOUND_SOUND = 1 SOUND_VOICE = 2 LIGHT_OFF = 0 LIGHT_CELL = 1 LIGHT_ALWAYS = 2 VOICE_CELL = 0 VOICE_BAND = 1 VOICE_BAND_LAC = 2 VOICE_BAND_LAC_NET = 3 GPS_INTERNAL = 0 GPS_EXTERNAL = 1 ## The constructor. def __init__( self ): global config ## Config options self._iSound = [u'OFF', u'Sound', u'Voice'] self._iLight = [u'OFF', u'Cell change on', u'Always on'] self._iVoice = [u'Cell only', u'Band & Cell', u'Band & Cell & LAC', u'Cell & LAC & NET'] self._iGPS = [u'Internal', u'External'] ## Setup Menu self._iMenu = [(u'Defaults', self.defaultConfig), (u'Reload Config File', self.loadConfig)] ## Form fields self.configToForm() #flags = appuifw.FFormEditModeOnly flags = appuifw.FFormEditModeOnly + appuifw.FFormDoubleSpaced self._iForm = appuifw.Form(self._iFields, flags) self._iForm.save_hook = self.saveForm #self._iForm.flags = appuifw.FFormEditModeOnly self._iForm.menu = self._iMenu def configToForm(self): global config _iFields = [( u'CellNotify', 'combo', ( self._iSound, config['CellNotify'] ) ), ( u'Light', 'combo', ( self._iLight, config['Light'] ) ), ( u'Refresh','float', config['Refresh'] ), ( u'Volume','number', config['Volume'] ), ( u'VoiceText','combo', (self._iVoice, config['Voice']) ), ( u'GPS', 'combo', ( self._iGPS, config['GPS'] ) )] ## XXX: live update doesn't work yet. Need fixing if hasattr(self, '_iFields'): for i in range(len(_iFields)): self._iFields[i] = _iFields[i] else: self._iFields = _iFields def formToConfig(self): global config ## XXX: must find elegant way for conversion Form <-> config config = { 'CellNotify': self._iFields[0][2][1], 'Light': self._iFields[1][2][1], 'Refresh': self._iFields[2][2], 'Volume': self._iFields[3][2], 'Voice': self._iFields[4][2][1], 'GPS': self._iFields[5][2][1] } cfgfile = open(configfile, 'wt') cfgfile.write(repr(config)) cfgfile.close() def defaultConfig(self): global config, defconfig config = defconfig.copy() self.configToForm() appuifw.note(u"Please reopen the Setup dialog.") def loadConfig(self, silent=False): global config config = defconfig.copy() try: cfgfile = open(configfile, 'rt') content = cfgfile.read() cfgfile.close() config.update(eval(content)) if not silent: appuifw.note(u"Please reopen the Setup dialog.") except IOError, error: if not silent: appuifw.note(unicode(error),'error') self.configToForm() ## Displays the form. def execute( self ): self._iForm.execute( ) ## save_hook send True if the form has been saved. def saveForm( self, aConfig ): global config self._iFields = aConfig self.formToConfig() class CellCam: SIZE = (640,480) def __init__(self): global finder self.timeout = 100 self.lock = e32.Ao_lock() finder = True self.oldexit = appuifw.app.exit_key_handler appuifw.app.exit_key_handler = self.abort appuifw.app.body.bind(key_codes.EKeySelect, self.take_picture) camera.start_finder(self.finder_cb) self.lock.wait() def abort(self): camera.stop_finder() self.finish() def finish(self): global finder appuifw.app.body.bind(key_codes.EKeySelect, None) appuifw.app.exit_key_handler = self.oldexit self.lock.signal() finder = False def take_picture(self): global finder, gsmloc fname = rxl_line(gsmloc, fname=True) camera.stop_finder() img = camera.take_photo(size=CellCam.SIZE) w, h = appuifw.app.body.size appuifw.app.body.blit(img,target=(0, 0, w, 0.75 * w), scale = 1) appuifw.app.body.text((10,20), u"Writing Image:", fill=color) appuifw.app.body.text((10,40), fname, fill=color) img.save("E:\\" + fname) self.finish() def finder_cb(self, im): global gsmloc info = rxl_line(gsmloc) im.text((10,20), info, fill=color) #im.text((10,40), u"Feature test: " + unicode(self.timeout), fill=color) appuifw.app.body.blit(im) img=Image.new((800,352)) img_dbl=Image.new((800,352)) def mainmenu_setup(): logmenu = [(u"Start Logging", startlog), (u"Stop Logging", stoplog)] gpsmenu = [(u"Start GPS", startgps), (u"Stop GPS", stopgps)] appuifw.app.menu = [logmenu[logger != None], gpsmenu[gpson], (u"Orientation", rotate), (u"Site Photo...", cellphoto), (u"Settings...", setup)] def menus_setup(): global tab mainmenu_setup() appuifw.app.set_tabs([u"NetMon", u"Graph", u"History", u"GPS", u"About"], set_tab) appuifw.app.activate_tab(tab) def rotate(): if appuifw.app.orientation == 'landscape': appuifw.app.orientation = 'portrait' else: appuifw.app.orientation = 'landscape' def setup(): # Bug in Form(): manually hide Tabs appuifw.app.set_tabs([], None) SetupForm.execute( ) menus_setup() def set_tab(newtab): global tab, gui tab = newtab gui.signal() def startgps(): global gpson, locreq, gpsdata if not locreq: appuifw.note(u"Module Locationrequestor not installed!",'error') return if not gpson: gpsdata = [0, 'Starting GPS...'] gpson = 1 mainmenu_setup() thread.start_new_thread(gps_worker,()) def stopgps(): global gpson if gpson: gpson = 0 mainmenu_setup() def cellphoto(): appuifw.app.set_tabs([], None) c = CellCam() menus_setup() def startlog(): global logger, logname logname = "e:/NetMonLog_"+time.strftime("%Y%m%d_%H%M%S")+".txt" logger = Logger(logname) #sys.stderr = sys.stdout = my_log logger.logtab(("Date (Y/M/D)", "Time (H:M:S)", "CID", "LCID", "LAC", "MCC", "MNC", "RNC", "RXL", "LON", "LAT")) mainmenu_setup() def stoplog(): global logger logger = None mainmenu_setup() def writelog(list): #my_log = Logger(logname) #sys.stderr = sys.stdout = my_log logger.logtab(list) def sound_play(): global s if not s: return try: s.set_volume(config['Volume']) s.play() except Exception, error: appuifw.note(u"Sound Error:" + unicode(error),'error') def decode_cid(mcc, mnc, cid): global draw_rnc #if (mnc == 7) and (mcc == 262) and (cid > 65535): if cid > 65535: cidhex="%Xh" % cid rnc=int(cidhex[2:-4],16) cid=int(cidhex[-4:],16) draw_rnc = True return (cid, rnc) else: return (cid, 'n/a') def log_rxl(t, rxl): global rxl_log, rxl_last if not rxl: rxl = 0 if rxl_last == 0: rxl_last = t - 1 while t > rxl_last: # rescue last 1000 entries, negate for easier drawing rxl_log = rxl_log[-999:] + [-rxl] rxl_last += 1 def dbm(rxl): if not rxl: return u"Offline" else: return unicode(rxl) + u" dBm" def log_loc(t, gsmloc): global loc_log if len(loc_log) > 0 and gsmloc == loc_log[-1:][0][1]: return loc_log += [[t, gsmloc]] def draw_about(): global img size,offset = appuifw.app.layout(appuifw.EMainPane) img.rectangle((0,0)+size,outline=border,fill=bg) img.text((50,20), VERSION,fill=color) img.text((75,40), u'(c) 2007 by',fill=color) img.text((10,60), u'Carsten Knütter, Georg Lukas',fill=color) img.text((10,90), u'netmonitor.s80@o2online.de',fill=color) img.text((5,110), u'This program comes with',fill=color) img.text((5,130), u'ABSOLUTELY NO WARRANTY',fill=color) img.text((5,160), u'This is free software, and you',fill=color) img.text((5,180), u'are welcome to redistribute',fill=color) img.text((5,200), u'it under certain conditions',fill=color) #img.text((2,220), u'RES: ',fill=(255,255,255), font=(None,40)) #img.text((2,210), u'certain conditions',fill=color) #img.text((35,230), u'Press Options to start... ',fill=color) def draw_netmon(lac, mcc, mnc, rnc, rxl, ver, imei, longcid, cidhex, lachex, longcidhex): try: bars = sysinfo.signal_bars() except SystemError: bars = u"n/a" batty = sysinfo.battery() ram = sysinfo.free_ram()/1024 totram = sysinfo.total_ram()/1024 x = appuifw.app.layout(appuifw.EScreen) if longcid == u'n/a': cidtext = u'CID: ('+unicode(cidhex)+u')' band = u' (GSM)'; else: cidtext = u'LCID: '+unicode(longcid)+u' ('+unicode(longcidhex)+u')' band = u' (UMTS)'; img.text((3,40), cidtext,fill=color) img.text((3,60), u'LAC: '+unicode(lac)+u' ('+unicode(lachex)+u')',fill=color) img.text((3,80), u'RNC: '+unicode(rnc),fill=color) img.text((3,100), u'NET: '+unicode(mcc)+u" - "+unicode(mnc)+band,fill=color) img.text((3,120), u'RXL: '+dbm(rxl)+u" ("+unicode(bars)+u")",fill=color) img.text((3,140), u'BAT: '+unicode(batty)+u" %",fill=color) img.text((3,160), u'RAM: '+unicode(ram)+u" / "+unicode(totram)+u" KB",fill=color) img.text((3,180), u'VER: '+unicode(ver),fill=color) img.text((3,200), u'IMEI: '+imei,fill=color) img.text((3,220), u'RES: '+unicode(x),fill=color) #img.text((3,220), u'APP: '+appuifw.app.full_name(),fill=color) def draw_rxl_line(size, pos, text): sz = (size[1]-2, 20) i = Image.new(sz, '1') i.rectangle((0,0)+sz, fill=0x000000) i.text((2,11), text, 0xffffff) irot = i.transpose(ROTATE_90) img.blit(irot, target=pos, mask=irot) #img.text((0,200), unicode(sz)+unicode(pos)+text, color) def headline(size, cid): (w, h) = size # CLK and CID always on top img.rectangle((0,0,w,20),headcol,fill=headbg) img.text((3,17), u'CID: '+unicode(cid),fill=bg) img.text((w-73,17), unicode(time.strftime("%H:%M:%S")),fill=bg) if logger: img.text((w/2-25,17), u'L',fill=bg) if (gpson == 1): if (len(gpsdata) == 15): gpsstat = unicode(gpsdata[14]) else: gpsstat = u"*" img.text((w/2+5,17), u'G'+gpsstat,fill=bg) def rxl_line(log, fname=False): (mcc, mnc, lac, cid) = log longcid = cid (cid, rnc) = decode_cid(mcc, mnc, cid) if mcc is None or mcc == "n/a": return u"Offline" else: if fname: return unicode(mcc)+"-"+unicode(mnc)+u"_"+unicode(lac)+u"-"+unicode(longcid)+".jpg" else: return unicode(mcc)+"-"+unicode(mnc)+u" C:"+unicode(cid)+u" L:"+unicode(lac) def draw_rxlgraph(size): global rxl_log, loc_log if len(rxl_log) == 0: return t = time.time() w = size[0] h = size[1] sublog = rxl_log[-w:] xcoord = range(w-len(sublog)+1, w+1) line = zip(xcoord, sublog) img.line(line, outline=linecol) x = w l = len(loc_log)-1 #img.text((0,h-36),unicode(loc_log),linecol) while x > -20 and l >= 0: offset = t - loc_log[l][0] x = w - offset img.line((x,0,x,h),outline=linecol) draw_rxl_line(size, (x+1, 1), rxl_line(loc_log[l][1])) l-=1 img.text((3, 40), dbm(-rxl_log[-1:][0]), color) def draw_history_line(x, y, tstamp, net, lac, cid, rnc): global draw_rnc img.text((5+x, y), unicode(tstamp), color) img.text((85+x, y), unicode(cid), color) img.text((140+x, y), unicode(lac), color) if draw_rnc: img.text((200+x, y), unicode(rnc), color) img.text((250+x, y), unicode(net), color) else: img.text((200+x, y), unicode(net), color) def draw_history(size): global loc_log y = 64 count = (size[1]-y)/16 if size[0] > 320: count = 2 * count draw_history_line(400, y-16, "Time", "Net", "LAC", "CID", "RNC") #img.text((0, 10), unicode(count), color) draw_history_line(0, y-16, "Time", "Net", "LAC", "CID", "RNC") x = 0 i = 0 for v in loc_log[-count:]: t = time.strftime("%H:%M:%S ", time.localtime(v[0])) (mcc, mnc, lac, cid) = v[1] if mcc is None: cid = u"Offline" net = lac = rnc = u"" else: (cid, rnc) = decode_cid(mcc, mnc, cid) net = unicode(mcc)+"-"+unicode(mnc) if (i == (count/2)) and (size[0] > 320): x = 400 y = 64 draw_history_line(x, y, t, net, lac, cid, rnc) y += 16 i += 1 def draw_gps(size): global gpsdata, usinggps, gpson (w, h) = size if (gpson == 1) and (len(gpsdata) == 15): img.text((10,40), u'Modul: '+gpsdata[6]+u' ('+unicode(usinggps)+u')',fill=color) img.text((10,60), u'Lat: '+unicode("%9.6f"%gpsdata[1]),fill=color) img.text((10,80), u'Lon: '+unicode("%9.6f"%gpsdata[2]),fill=color) img.text((10,100), u'Altitude: '+unicode("%7.1f"%gpsdata[3])+u' m',fill=color) img.text((10,120), u'Accuracy: '+u'H: '+unicode("%5.1f"%gpsdata[4])+u'm'+u' V: '+unicode("%5.1f"%gpsdata[5])+u'm',fill=color) img.text((10,140), u'Speed: '+unicode("%5.1f"%gpsdata[8])+u' km/h',fill=color) img.text((10,160), u'Heading: '+unicode("%5.1f"%gpsdata[10])+u'°',fill=color) img.text((10,180), u'SAT (Fix): '+unicode(gpsdata[13])+u' ('+unicode(gpsdata[14])+u')',fill=color) elif len(gpsdata) >= 2: # Display Error from gpsdata[1] img.text((10,40), u'No data available:',fill=color) img.text((10,60), unicode(gpsdata[1]),fill=color) else: img.text((int(w/2)-30,100), u'GPS OFF',fill=color) def reader(text): try: audio.say(text) except Exception, error: # no GUI from sub-thread! #appuifw.note(u"Sound Error: "+unicode(error),'error') pass def gps_update(args): global gpsdata, gpson gpsdata = args if (not gpson): lockgps.signal() gpsdata = [0, 'GPS stopped'] def gps_find_device(lr, type): count = lr.GetNumModules() for i in range(count): info = lr.GetModuleInfoByIndex(i) if ((info[3] == type) and ((info[2] & locationrequestor.ETechnologyNetwork) == 0)): return info[0] return -1 def gps_worker(): global lockgps, usinggps lockgps = e32.Ao_lock() # create LocationRequestor instance lr = locationrequestor.LocationRequestor() # show default module #print 'Default', lr.GetDefaultModuleId() # show number of modules #print 'Count', count # find internal GPS (if any) if config['GPS'] == Setup.GPS_INTERNAL: probe = [locationrequestor.EDeviceInternal, locationrequestor.EDeviceExternal] else: probe = [locationrequestor.EDeviceExternal, locationrequestor.EDeviceInternal] id = gps_find_device(lr, probe[0]) if id == -1: id = gps_find_device(lr, probe[1]) if id == -1: gpsdata = [0, 'Neither internal nor external GPS available'] return usinggps = unicode(id) # set update options lr.SetUpdateOptions(1, 25, 0, 1) # connect to position module lr.Open(id) # install callback try: lr.InstallPositionCallback(gps_update) except Exception, reason: #appuifw.note(unicode(reason),'error') pass lockgps.wait() lr.Close() def log_worker(): global gsmloc, gsmloc_real, rxl, gui, t_last, running t_old = 0 while running: t_last = time.time() try: rxl = -sysinfo.signal_dbm() if rxl == 0: # Workaround Nokia dumbness: rxl = None except SystemError: rxl = None try: oldloc = gsmloc gsmloc_real = location.gsm_location() #if t_last % 10 < 3: # gsmloc_real = None # Workaround for "no location during data traffic" if gsmloc_real is None and rxl: # and t_old + offlinetimeout < t_last: # We are busy, simulate last cell for history gsmloc = oldloc else: # Update state gsmloc = gsmloc_real t_old = t_last except (ValueError, SystemError): gsmloc = gsmloc_real = None log_rxl(t_last, rxl) if gsmloc is None or not rxl: log_loc(t_last, (None, None, None, None)) else: log_loc(t_last, gsmloc) gui.signal() e32.ao_sleep(1) def netmonitor(): global running, toggle, rxl, gsmloc, img, img_dbl busytext = ['n/a', 'busy'] imei=sysinfo.imei() ver = sysinfo.os_version() oldcid = 0 oldlac = 0 t_log = 0 while running: # Block when Camera is active while finder: e32.ao_sleep(0.5) size,offset = appuifw.app.layout(appuifw.EMainPane) img.rectangle((0,0)+size,outline=border,fill=bg) if gsmloc_real is None and gsmloc is not None: busy = True else: busy = False if gsmloc is None or not rxl or busy: mcc = mnc = lac = lachex = cid = cidhex = longcid = longcidhex = rnc = cid = busytext[busy] else: (mcc, mnc, lac, cid) = gsmloc if cid > 65535: umts = 1 longcid = cid longcidhex = "%Xh"%(longcid) else: umts = 0 longcid = longcidhex = busytext[busy] (cid, rnc) = decode_cid(mcc, mnc, cid) try: lachex = "%Xh"%(lac) except: lachex = busytext[busy] try: cidhex = "%Xh"%(cid) except: cidhex = busytext[busy] if cid != oldcid: oldcid = cid # Cell changed, notify the user if config['Light']==Setup.LIGHT_CELL: e32.reset_inactivity() if config['CellNotify']==Setup.SOUND_SOUND: sound_play() elif config['CellNotify']==Setup.SOUND_VOICE: #sound.set_volume(config['Volume']) voice = u'' if not rxl: voice = Voice.offline elif config['Voice']==Setup.VOICE_CELL: voice = unicode(cid) elif config['Voice']==Setup.VOICE_BAND: voice = Voice.band[umts] + u" " + unicode(cid) elif config['Voice']==Setup.VOICE_BAND_LAC: voice = Voice.band[umts] + u" " if oldlac != lac: voice += unicode(cid)+u'. L A C '+unicode(lac) else: voice += unicode(cid) elif config['Voice']==Setup.VOICE_BAND_LAC_NET: voice = Voice.band[umts] + u" " if oldlac != lac: if mcc in Voice.provider and mnc in Voice.provider[mcc]: voice += Voice.provider[mcc][mnc] else: voice += u'M C C '+unicode(mcc)+u' M N C '+unicode(mnc) voice += u". " + unicode(cid) + u'. L A C '+unicode(lac) else: voice += unicode(cid) thread.start_new_thread(reader,(voice,)) oldlac = lac if tab==0: draw_netmon(lac, mcc, mnc, rnc, rxl, ver, imei, longcid, cidhex, lachex, longcidhex) elif tab==1: draw_rxlgraph(size) elif tab==2: draw_history(size) elif tab==3: draw_gps(size) elif tab==4: draw_about() # Headline for all but "About" if tab <= 3: headline(size, cid) tmp = img_dbl img_dbl = img img = tmp handle_redraw(()) if logger: if t_log != t_last: # Only log coords when GPS is there and values are not NaN if (gpson == 1) and (len(gpsdata) == 15) and str(gpsdata[1]) != 'NaN': # Convert lat, lon to formatted strings (lon, lat) = map(lambda x: "%2.6f" % x, gpsdata[1:3]) else: lon = lat = 'n/a' if rxl: trxl = rxl else: trxl = 'n/a' writelog((time.strftime("%Y/%m/%d"), time.strftime("%H:%M:%S"), cid, longcid, lac, mcc, mnc, rnc, trxl, lon, lat)) t_log = t_last if config['Light']==Setup.LIGHT_ALWAYS: e32.reset_inactivity() #e32.ao_yield() gui.wait() def handle_redraw(rect): global img_dbl, canvas canvas.blit(img_dbl) def standalone(): # XXX: use appuifw.app.uid() # TITLE in full_name() means standalone return appuifw.app.full_name().find(TITLE) != -1 def exit_key_handler(): if appuifw.query(u'Do you really want to exit?', 'query'): finish() def finish(): global script_lock, running, s script_lock.signal() gui.signal() #appuifw.app.set_exit() running = False if s != None: s.close() appuifw.app.set_tabs([], None) if standalone(): appuifw.app.set_exit() def main(): global script_lock, gui, s, SetupForm, canvas canvas = appuifw.Canvas(event_callback=None, redraw_callback=handle_redraw) #appuifw.app.screen='large' appuifw.app.screen='normal' appuifw.app.title = TITLE appuifw.app.body=canvas script_lock = e32.Ao_lock() gui = e32.Ao_lock() thread.start_new_thread(log_worker,()) SetupForm = Setup( ) SetupForm.loadConfig(silent=True) menus_setup() appuifw.app.exit_key_handler = exit_key_handler appuifw.app.body.bind(key_codes.EKeyNo, exit_key_handler) #if locreq and config['GPS']: # startgps() try: file = u"e:\\fanfare3.mp3" s = Sound.open(file) except Exception, error: appuifw.note(u"Sound Error: "+unicode(error),'error') #audio.say(u'test 123456789') netmonitor() script_lock.wait() if standalone(): # Standalone: Display exception as note try: main() except Exception, error: appuifw.note(u"Fatal Error: "+unicode(error),'error') else: # Python Script Shell: log exception to stdout main()