fso-frameworkd 0.9.5.9+git20110512-4ubuntu1 / usr / bin / cli-framework

#!/usr/bin/python
# -*- coding: UTF-8 -*-
"""
Interactive console for the freesmartphone.org framework

(C) 2008-2010 Michael 'Mickey' Lauer <mlauer@vanille-media.de>
(C) 2008 Jan 'Shoragan' Lübbe <jluebbe@lasnet.de>
(C) 2008 Openmoko, Inc.

GPLv2 or later
"""

__version__ = "0.9.4"

import dbus.service
import dbus.mainloop.glib
from gobject import MainLoop
import os, sys, code, pprint, types, atexit

pp = pprint.PrettyPrinter( indent=4 )

INTROSPECT = False

#----------------------------------------------------------------------------#
def dbus_to_python( v ):
#----------------------------------------------------------------------------#
    class ObjectPath( object ):
        def __init__( self, path ):
            self.path = str( path )

        def __repr__( self ):
            return "op%s" % repr(self.path)

    if isinstance(v, dbus.Byte) \
        or isinstance(v, dbus.Int64) \
        or isinstance(v, dbus.UInt64) \
        or isinstance(v, dbus.Int32) \
        or isinstance(v, dbus.UInt32) \
        or isinstance(v, dbus.Int16) \
        or isinstance(v, dbus.UInt16) \
        or type(v) == types.IntType:
        return int(v)
    elif isinstance(v, dbus.Double) or type(v) == types.FloatType:
        return float(v)
    elif isinstance(v, dbus.String) or type(v) == types.StringType:
        return str(v)
    elif isinstance(v, dbus.Dictionary) or type(v) == types.DictType:
        return dict( (dbus_to_python(k), dbus_to_python(v)) for k,v in v.iteritems() )
    elif isinstance(v, dbus.Array) or type(v) == types.ListType:
        return [dbus_to_python(x) for x in v]
    elif isinstance(v, dbus.Struct) or type(v) == types.TupleType:
        return tuple(dbus_to_python(x) for x in v)
    elif isinstance(v, dbus.Boolean) or type(v) == types.BooleanType:
        return bool(v)
    elif isinstance(v, dbus.ObjectPath) or type(v) == ObjectPath:
        return ObjectPath(v)
    else:
        return v
        raise TypeError("Can't convert type %s to python object" % type(v))

#----------------------------------------------------------------------------#
def prettyPrint( expression ):
#----------------------------------------------------------------------------#
    if expression is not None:
        pp.pprint( dbus_to_python(expression) )

#----------------------------------------------------------------------------#
def handler( *args, **kwargs ):
#----------------------------------------------------------------------------#
    if kwargs["path"].startswith( "/org/freesmartphone" ):
        print "[SIGNAL]   %s.%s    from " % ( kwargs["interface"], kwargs["member"] ),
        print "%s via %s" % ( kwargs["path"], kwargs["sender"] )
        for arg in args[:-1]:
            print "%s, " % arg,
        print "%s" % args[-1]
        print ">>> ",
        sys.stdout.flush()

#----------------------------------------------------------------------------#
def getObject( busname, objectpath ):
#----------------------------------------------------------------------------#
    return bus.get_object( busname, objectpath, follow_name_owner_changes=True, introspect=INTROSPECT )

#----------------------------------------------------------------------------#
def getInterface( busname, objectpath, interface ):
#----------------------------------------------------------------------------#
    try:
        proxy = getObject( busname, objectpath )
        iface = dbus.Interface( proxy, interface )
    except dbus.DBusException:
        print "DBus Exception. Can't get %s@%s:%s" % ( interface, busname, objectpath )
        return None
    else:
        return iface

#----------------------------------------------------------------------------#
def getObjectsForInterface( busname, interface ):
#----------------------------------------------------------------------------#
    try:
        paths = dbus_hlid.ListObjectsByInterface( busname, interface, dbus_interface="org.freesmartphone.DBus" )
    except dbus.DBusException, e:
        print "DBus Exception. dbus-hlid not installed"
        paths = []
    result = {}
    for path in paths:
        result[str(path)] = getInterface( busname, path, interface )
    return result

dbus.mainloop.glib.DBusGMainLoop( set_as_default=True )
mainloop = MainLoop()
bus = dbus.SystemBus()

# FIXME use introspection

# dbus_hlid object
dbus_hlid = getObject( "org.freesmartphone.DBus", "/org/freesmartphone/DBus" )

# framework object
framework = getObject( "org.freesmartphone.frameworkd", "/org/freesmartphone/Framework" )

# ogsmd device object
gsm = getObject( "org.freesmartphone.ogsmd", "/org/freesmartphone/GSM/Device" )

# ogps device object
gps = getObject( "org.freesmartphone.ogpsd", "/org/freedesktop/Gypsy" )

# usage device object
usage = getObject( "org.freesmartphone.ousaged", "/org/freesmartphone/Usage" )

# network
network = getInterface( \
    "org.freesmartphone.onetworkd",
    "/org/freesmartphone/Network",
    "org.freesmartphone.Network" )

# testing
testing = getInterface( \
    "org.freesmartphone.testing",
    "/org/freesmartphone/Testing",
    "org.freesmartphone.Testing" )

# framework
frameworkiface = getInterface( \
    "org.freesmartphone.frameworkd",
    "/org/freesmartphone/Framework",
    "org.freesmartphone.Framework" )

# odeviced objects
devaudio = getInterface( \
    "org.freesmartphone.odeviced",
    "/org/freesmartphone/Device/Audio",
    "org.freesmartphone.Device.Audio" )

devidle = getInterface( \
    "org.freesmartphone.odeviced",
    "/org/freesmartphone/Device/IdleNotifier/0",
    "org.freesmartphone.Device.IdleNotifier" )

devrtc = getInterface( \
    "org.freesmartphone.odeviced",
    "/org/freesmartphone/Device/RTC/0",
    "org.freesmartphone.Device.RealtimeClock" )
    
# otimed objects
timealarm = getInterface( \
    "org.freesmartphone.otimed",
    "/org/freesmartphone/Time/Alarm",
    "org.freesmartphone.Time.Alarm" )

try:
    display = getObjectsForInterface( "org.freesmartphone.odeviced", "org.freesmartphone.Device.Display" ).values()[0]
except IndexError:
    pass
try:
    bat = getObjectsForInterface( "org.freesmartphone.odeviced", "org.freesmartphone.Device.PowerSupply" ).values()[0]
except IndexError:
    pass
leds = getObjectsForInterface( "org.freesmartphone.odeviced", "org.freesmartphone.Device.LED" ).values()

# omuxerd object
muxer = getInterface( \
    "org.freesmartphone.omuxerd",
    "/org/freesmartphone/GSM/Muxer",
    "org.freesmartphone.GSM.MUX" )

# ogsmd device objects
gsmdevice = getInterface( \
    "org.freesmartphone.ogsmd",
    "/org/freesmartphone/GSM/Device",
    "org.freesmartphone.GSM.Device" )
gsmrtc = getInterface( \
    "org.freesmartphone.ogsmd",
    "/org/freesmartphone/GSM/Device",
    "org.freesmartphone.Device.RealtimeClock" )
gsmsim = getInterface( \
    "org.freesmartphone.ogsmd",
    "/org/freesmartphone/GSM/Device",
    "org.freesmartphone.GSM.SIM" )
gsmsms = getInterface( \
    "org.freesmartphone.ogsmd",
    "/org/freesmartphone/GSM/Device",
    "org.freesmartphone.GSM.SMS" )
gsmnetwork = getInterface( \
    "org.freesmartphone.ogsmd",
    "/org/freesmartphone/GSM/Device",
    "org.freesmartphone.GSM.Network" )
gsmcall = getInterface( \
    "org.freesmartphone.ogsmd",
    "/org/freesmartphone/GSM/Device",
    "org.freesmartphone.GSM.Call" )
gsmcb = getInterface( \
    "org.freesmartphone.ogsmd",
    "/org/freesmartphone/GSM/Device",
    "org.freesmartphone.GSM.CB" )
gsmpdp = getInterface( \
    "org.freesmartphone.ogsmd",
    "/org/freesmartphone/GSM/Device",
    "org.freesmartphone.GSM.PDP" )
gsmmonitor = getInterface( \
    "org.freesmartphone.ogsmd",
    "/org/freesmartphone/GSM/Device",
    "org.freesmartphone.GSM.Monitor" )
gsmtest = getInterface( \
    "org.freesmartphone.ogsmd",
    "/org/freesmartphone/GSM/Device",
    "org.freesmartphone.GSM.Test" )
gsmdebug = getInterface( \
    "org.freesmartphone.ogsmd",
    "/org/freesmartphone/GSM/Device",
    "org.freesmartphone.GSM.Debug" )

# ogsmd server objects
gsmdata = getInterface( \
    "org.freesmartphone.ogsmd",
    "/org/freesmartphone/GSM/Server",
    "org.freesmartphone.GSM.Data" )
gsmhz = getInterface( \
    "org.freesmartphone.ogsmd",
    "/org/freesmartphone/GSM/Server",
    "org.freesmartphone.GSM.HZ" )
gsmphone = getInterface( \
    "org.freesmartphone.ogsmd",
    "/org/freesmartphone/GSM/Server",
    "org.freesmartphone.GSM.Phone" )

# ogps
gpsaccuracy = getInterface( \
    "org.freesmartphone.ogpsd",
    "/org/freedesktop/Gypsy",
    "org.freedesktop.Gypsy.Accuracy" )

gpsposition = getInterface( \
    "org.freesmartphone.ogpsd",
    "/org/freedesktop/Gypsy",
    "org.freedesktop.Gypsy.Position" )

gpssatellite = getInterface( \
    "org.freesmartphone.ogpsd",
    "/org/freedesktop/Gypsy",
    "org.freedesktop.Gypsy.Satellite" )

gpstime = getInterface( \
    "org.freesmartphone.ogpsd",
    "/org/freedesktop/Gypsy",
    "org.freedesktop.Gypsy.Time" )

ubxdebug = getInterface( \
    "org.freesmartphone.ogpsd",
    "/org/freedesktop/Gypsy",
    "org.freesmartphone.GPS.UBX" )

# usage
usageiface = getInterface( \
    "org.freesmartphone.ousaged",
    "/org/freesmartphone/Usage",
    "org.freesmartphone.Usage" )

# phone
phone = getInterface( \
    "org.freesmartphone.ophoned",
    "/org/freesmartphone/Phone",
    "org.freesmartphone.Phone" )

# preferences
preferences = getInterface( \
    "org.freesmartphone.opreferencesd",
    "/org/freesmartphone/Preferences",
    "org.freesmartphone.Preferences" )

# events
events = getInterface( \
    "org.freesmartphone.oeventsd",
    "/org/freesmartphone/Events",
    "org.freesmartphone.Events" )


# pim
pim = getInterface( \
    "org.freesmartphone.opimd",
    "/org/freesmartphone/PIM",
    "org.freesmartphone.PIM" )

# pimsources
pims = getInterface( \
    "org.freesmartphone.opimd",
    "/org/freesmartphone/PIM/Sources",
    "org.freesmartphone.PIM.Sources" )

# pimcontacts
pimc = getInterface( \
    "org.freesmartphone.opimd",
    "/org/freesmartphone/PIM/Contacts",
    "org.freesmartphone.PIM.Contacts" )

# pimcontacts
pimcq = getInterface( \
    "org.freesmartphone.opimd",
    "/org/freesmartphone/PIM/Contacts/Queries",
    "org.freesmartphone.PIM.ContactQuery" )

# pimmessages
pimm = getInterface( \
    "org.freesmartphone.opimd",
    "/org/freesmartphone/PIM/Messages",
    "org.freesmartphone.PIM.Messages" )

# pimmessagequeries
pimmq = getInterface( \
    "org.freesmartphone.opimd",
    "/org/freesmartphone/PIM/Messages/Queries",
    "org.freesmartphone.PIM.MessageQuery" )

# bluez
try:
    bluez_manager = getInterface( \
        "org.bluez",
        "/",
        "org.bluez.Manager" )

    bluez = getInterface( \
        "org.bluez",
        bluez_manager.DefaultAdapter(),
        "org.bluez.Adapter" )
except:
    print "failed to connect to bluez"

bus.add_signal_receiver( handler, None, None, None, None,
    sender_keyword = "sender",
    destination_keyword = "destination",
    interface_keyword = "interface",
    member_keyword = "member",
    path_keyword = "path" )

import rlcompleter, readline, atexit
readline.parse_and_bind( "tab: complete" )
readline.set_history_length(1000)
try:
    readline.read_history_file( os.path.expanduser( "~/.framework-history" ) )
except IOError:
    pass
atexit.register( readline.write_history_file, os.path.expanduser( "~/.framework-history" ) )

def runmainloop():
    try:
        mainloop.run()
    except KeyboardInterrupt:
        mainloop.quit()
        sys.exit( 0 )

sys.displayhook = prettyPrint
console = code.InteractiveConsole( locals() )
try:
    console.interact( "freesmartphone.org interactive command line" )
except KeyboardInterrupt:
    sys.exit( 0 )