src/sat-monitor

   1 #!/usr/bin/python
   2 #
   3 # Satellite monitor for Gypsy, based on the zhone satellite monitor
   4 #
   5 # Copyright (C) 2009  Enrico Zini <enrico@enricozini.org>
   6 #
   7 # This program is free software; you can redistribute it and/or modify
   8 # it under the terms of the GNU General Public License as published by
   9 # the Free Software Foundation; either version 2 of the License, or
  10 # (at your option) any later version.
  11 #
  12 # This program is distributed in the hope that it will be useful,
  13 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  14 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  15 # GNU General Public License for more details.
  16 #
  17 # You should have received a copy of the GNU General Public License
  18 # along with this program; if not, write to the Free Software
  19 # Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
  20
  21 import sys
  22 import os
  23 import os.path
  24 import dbus
  25 import dbus.mainloop.glib
  26 import gtk
  27 from gettext import gettext as _
  28
  29 SAT_QI_NAMES = {
  30     0: _("idle"),
  31     1: _("searching"),
  32     2: _("signal acquired"),
  33     3: _("signal unusable"),
  34     4: _("code lock"),
  35     5: _("code&carrier lock"),
  36     6: _("code&carrier lock"),
  37     7: _("receiving data")
  38 }
  39
  40
  41 class GPSMonitor:
  42     def __init__(self):
  43         self.bus = dbus.SystemBus()
  44
  45         # see mdbus -s org.freesmartphone.ousaged /org/freesmartphone/Usage
  46         self.usage = self.bus.get_object('org.freesmartphone.ousaged', '/org/freesmartphone/Usage')
  47         self.usage = dbus.Interface(self.usage, "org.freesmartphone.Usage")
  48
  49         # see mdbus -s org.freesmartphone.ogpsd /org/freedesktop/Gypsy
  50         gps = self.bus.get_object('org.freesmartphone.ogpsd', '/org/freedesktop/Gypsy')
  51         self.gps = dbus.Interface(gps, "org.freedesktop.Gypsy.Device")
  52         self.gps_ubx = dbus.Interface(gps, 'org.freesmartphone.GPS.UBX')
  53
  54         # This piece of machinery is taken from Zhone
  55         self.debug_busy = None
  56         self.debug_want = set( ["NAV-STATUS", "NAV-SVINFO"] )
  57         self.debug_have = set()
  58         self.debug_error = set()
  59
  60         self.callback = None
  61
  62     def debug_update(self):
  63         if self.debug_busy is None:
  64             pending = self.debug_want - self.debug_have - self.debug_error
  65             if pending:
  66                 self.debug_busy = pending.pop()
  67                 self.gps_ubx.SetDebugFilter(
  68                     self.debug_busy,
  69                     True,
  70                     reply_handler=self.on_debug_reply,
  71                     error_handler=self.on_debug_error,
  72                 )
  73
  74     def debug_request(self):
  75         self.debug_have = set()
  76         self.debug_update()
  77
  78     def on_debug_reply(self):
  79         self.debug_have.add(self.debug_busy)
  80         self.debug_busy = None
  81         self.debug_update()
  82
  83     def on_debug_error(self, e):
  84         zavai.info(e, "error while requesting debug packet %s" % self.debug_busy)
  85         self.debug_error.add(self.debug_busy)
  86         self.debug_busy = None
  87         self.debug_update()
  88
  89     def on_satellites_changed(self, satellites):
  90         self.debug_request()
  91
  92     def on_ubxdebug_packet(self, clid, length, data):
  93         self.callback(clid, length, data)
  94
  95     def start(self):
  96         self.usage.RequestResource('GPS')
  97
  98         # TODO: find out how come sometimes these events are not sent
  99         self.bus.add_signal_receiver(
 100             self.on_satellites_changed, 'SatellitesChanged', 'org.freedesktop.Gypsy.Satellite',
 101             'org.freesmartphone.ogpsd', '/org/freedesktop/Gypsy')
 102         self.bus.add_signal_receiver(
 103             self.on_ubxdebug_packet, 'DebugPacket', 'org.freesmartphone.GPS.UBX',
 104             'org.freesmartphone.ogpsd', '/org/freedesktop/Gypsy')
 105         self.debug_request()
 106
 107     def stop(self):
 108         self.bus.remove_signal_receiver(
 109             self.on_satellites_changed, 'SatellitesChanged', 'org.freedesktop.Gypsy.Satellite',
 110             'org.freesmartphone.ogpsd', '/org/freedesktop/Gypsy')
 111         self.bus.remove_signal_receiver(
 112             self.on_ubxdebug_packet, 'DebugPacket', 'org.freesmartphone.GPS.UBX',
 113             'org.freesmartphone.ogpsd', '/org/freedesktop/Gypsy')
 114         self.usage.ReleaseResource('GPS')
 115
 116 class SatelliteMonitor(gtk.Window):
 117     def __init__(self, **kw):
 118         super(SatelliteMonitor, self).__init__()
 119
 120         self.set_title(_("Satellite monitor"))
 121
 122         self.gps = GPSMonitor()
 123         self.gps.callback = self.on_ubxdebug_packet
 124
 125         self.store = gtk.ListStore(str, str, str, str, str, str, str, str, str, str, str)
 126         self.view = gtk.TreeView(self.store)
 127         renderer = gtk.CellRendererText()
 128         for idx, name in enumerate((_("CH"), _("ID"), _("SN"), _("ELE"), _("AZI"),
 129                      _("Used"), _("Diff"), _("Alm"), _("Eph"),
 130                      _("Bad"), _("Status"))):
 131             col = gtk.TreeViewColumn(name)
 132             col.pack_start(renderer, False)
 133             col.add_attribute(renderer, "text", idx)
 134             self.view.append_column(col)
 135         self.add(self.view)
 136
 137         self.connect("destroy", gtk.main_quit)
 138         self.set_size_request(100, 200)
 139         self.show_all()
 140
 141     def start(self, *args):
 142         self.gps.start()
 143
 144     def stop(self, *args):
 145         self.gps.stop()
 146
 147     def on_ubxdebug_packet(self, clid, length, data):
 148         # In zhone it is cbUBXDebugPacket
 149         #if clid == "NAV-STATUS" and data:
 150         #    i = ["%s: %d" % (k, data[0][k]) for k in sorted(data[0].keys())]
 151         #    zavai.info("Status:", " ".join(i))
 152         ##    if data[0]['TTFF']:
 153         ##        zavai.info("TTFF: %f", data[0]['TTFF']/1000.0)
 154         if clid == "NAV-SVINFO":
 155             self.handle_ubx_sat_data(data[1:])
 156         #else:
 157         #    zavai.info("gps got ubxdebug packet", clid)
 158         #    zavai.info("DATA:", data)
 159
 160     def handle_ubx_sat_data(self, ubxinfo):
 161         #zavai.info("CH ID SN ELE AZI Used Diff Alm Eph Bad Status")
 162         self.store.clear()
 163         for sv in ubxinfo:
 164             if sv["CNO"] == 0: continue
 165             svid = sv["SVID"]
 166             used = sv["Flags"] & 0x01
 167             diff = sv["Flags"] & 0x02
 168             almoreph = sv["Flags"] & 0x04
 169             eph = sv["Flags"] & 0x08
 170             bad = sv["Flags"] & 0x10
 171             qi = ("%i: " % sv["QI"]) + SAT_QI_NAMES.get(sv["QI"], _("Unknown"))
 172             self.store.append([
 173                 "%2d" % sv["chn"],
 174                 "%2d" % sv["SVID"],
 175                 "%2d" % sv["CNO"],
 176                 "%3d" % sv["Elev"],
 177                 "%3d" % sv["Azim"],
 178                 used and "used" or "",
 179                 diff and "diff" or "",
 180                 almoreph and "alm" or "",
 181                 eph and "eph" or "",
 182                 bad and "bad" or "",
 183                 qi])
 184
 185 if __name__ == "__main__":
 186     dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
 187
 188     app = SatelliteMonitor()
 189     app.start()
 190     gtk.main()
 191
 192