#!/usr/bin/env python
import dbus
import dbus.decorators
import dbus.glib
import gobject
import sys
import getopt
from signal import *
mgr_cmds = [ "InterfaceVersion", "ListAdapters", "DefaultAdapter" ]
mgr_signals = [ "AdapterAdded", "AdapterRemoved" ]
dev_cmds = [ "GetAddress",
"GetVersion",
"GetRevision",
"GetManufacturer",
"GetCompany",
"GetMode",
"SetMode",
"GetDiscoverableTimeout",
"SetDiscoverableTimeout",
"IsConnectable",
"IsDiscoverable",
"IsConnected",
"ListConnections",
"GetMajorClass",
"ListAvailableMinorClasses",
"GetMinorClass",
"SetMinorClass",
"GetServiceClasses",
"GetName",
"SetName",
"GetRemoteVersion",
"GetRemoteRevision",
"GetRemoteManufacturer",
"GetRemoteCompany",
"GetRemoteMajorClass",
"GetRemoteMinorClass",
"GetRemoteServiceClasses",
"GetRemoteClass",
"GetRemoteName",
"GetRemoteAlias",
"SetRemoteAlias",
"ClearRemoteAlias",
"LastSeen",
"LastUsed",
"DisconnectRemoteDevice",
"CreateBonding",
"CancelBondingProcess",
"RemoveBonding",
"HasBonding",
"ListBondings",
"GetPinCodeLength",
"GetEncryptionKeySize",
"DiscoverDevices",
"DiscoverDevicesWithoutNameResolving",
"CancelDiscovery",
"ListRemoteDevices",
"ListRecentRemoteDevices" ]
dev_signals = [ "ModeChanged",
"NameChanged",
"MinorClassChanged",
"DiscoveryStarted",
"DiscoveryCompleted",
"RemoteDeviceFound",
"RemoteNameUpdated",
"RemoteNameFailed",
"RemoteAliasChanged"
"RemoteAliasCleared",
"RemoteDeviceConnected",
"RemoteDeviceDisconnectRequested",
"RemoteDeviceDisconnected",
"BondingCreated",
"BondingRemoved" ]
dev_signals_filter = [ "/org/bluez/hci0", "/org/bluez/hci1",
"/org/bluez/hci2", "/org/bluez/hci3",
"/org/bluez/hci4", "/org/bluez/hci5",
"/org/bluez/hci6", "/org/bluez/hci7" ]
class Tester:
exit_events = []
dev_path = None
need_dev = False
listen = False
at_interrupt = None
def __init__(self, argv):
self.name = argv[0]
self.parse_args(argv[1:])
try:
self.dbus_setup()
except dbus.DBusException, e:
print 'Failed to do D-Bus setup: %s' % e
sys.exit(1)
def parse_args(self, argv):
try:
opts, args = getopt.getopt(argv, "hli:")
except getopt.GetoptError:
self.usage()
sys.exit(1)
for o, a in opts:
if o == "-h":
self.usage()
sys.exit()
elif o == "-l":
self.listen = True
elif o == "-i":
if a[0] == '/':
self.dev_path = a
else:
self.dev_path = '/org/bluez/%s' % a
if not (args or self.listen):
self.usage()
sys.exit(1)
if args:
self.cmd = args[0]
self.cmd_args = args[1:]
def dbus_dev_setup(self):
if not self.dev_path:
try:
self.dbus_mgr_setup()
self.dev_path = self.manager.DefaultAdapter()
except dbus.DBusException, e:
print 'Failed to get default device: %s' % e
sys.exit(1)
try:
obj = self.bus.get_object('org.bluez', self.dev_path)
self.device = dbus.Interface(obj, 'org.bluez.Adapter')
except dbus.DBusException, e:
print 'Failed to setup device path: %s' % e
sys.exit(1)
def dbus_dev_sig_setup(self):
try:
for signal in dev_signals:
for path in dev_signals_filter:
self.bus.add_signal_receiver(self.dev_signal_handler,
signal, 'org.bluez.Adapter',
'org.bluez', path,
message_keyword='dbus_message')
except dbus.DBusException, e:
print 'Failed to setup signal handler for device path: %s' % e
sys.exit(1)
def dbus_mgr_sig_setup(self):
try:
for signal in mgr_signals:
self.bus.add_signal_receiver(self.mgr_signal_handler,
signal,'org.bluez.Manager',
'org.bluez', '/org/bluez')
except dbus.DBusException, e:
print 'Failed to setup signal handler for manager path: %s' % e
sys.exit(1)
def dbus_mgr_setup(self):
self.manager_obj = self.bus.get_object('org.bluez', '/org/bluez')
self.manager = dbus.Interface(self.manager_obj, 'org.bluez.Manager')
def dbus_setup(self):
self.bus = dbus.SystemBus()
def usage(self):
print 'Usage: %s [-i <dev>] [-l] [-h] <cmd> [arg1..]' % self.name
print ' -i <dev> Specify device (e.g. "hci0" or "/org/bluez/hci0")'
print ' -l Listen for events (no command required)'
print ' -h Show this help'
print 'Manager commands:'
for cmd in mgr_cmds:
print '\t%s' % cmd
print 'Adapter commands:'
for cmd in dev_cmds:
print '\t%s' % cmd
#@dbus.decorators.explicitly_pass_message
def dev_signal_handler(*args, **keywords):
dbus_message = keywords["dbus_message"]
print '%s - %s: ' % (dbus_message.get_member(), dbus_message.get_path()),
for arg in args[1:]:
print '%s ' % arg,
print
#@dbus.decorators.explicitly_pass_message
def mgr_signal_handler(*args, **keywords):
dbus_message = keywords["dbus_message"]
print '%s: ' % dbus_message.get_member()
for arg in args[1:]:
print '%s ' % arg,
print
def signal_cb(self, sig, frame):
print 'Caught signal, exiting'
if self.at_interrupt:
self.at_interrupt()
self.main_loop.quit()
def call_mgr_dbus_func(self):
if self.cmd == 'InterfaceVersion':
try:
print self.manager.InterfaceVersion()
except dbus.DBusException, e:
print 'Sending %s failed: %s' % (self.cmd, e)
if self.cmd == 'ListAdapters':
try:
devices = self.manager.ListAdapters()
except dbus.DBusException, e:
print 'Sending %s failed: %s' % (self.cmd, e)
sys.exit(1)
for device in devices:
print device
elif self.cmd == 'DefaultAdapter':
try:
print self.manager.DefaultAdapter()
except dbus.DBusException, e:
print 'Sending %s failed: %s' % (self.cmd, e)
sys.exit(1)
def call_dev_dbus_func(self):
try:
if self.cmd == 'GetAddress':
print self.device.GetAddress()
elif self.cmd == 'GetManufacturer':
print self.device.GetManufacturer()
elif self.cmd == 'GetVersion':
print self.device.GetVersion()
elif self.cmd == 'GetRevision':
print self.device.GetRevision()
elif self.cmd == 'GetCompany':
print self.device.GetCompany()
elif self.cmd == 'GetMode':
print self.device.GetMode()
elif self.cmd == 'SetMode':
if len(self.cmd_args) == 1:
self.device.SetMode(self.cmd_args[0])
else:
print 'Usage: %s -i <dev> SetMode scan_mode' % self.name
elif self.cmd == 'GetDiscoverableTimeout':
print '%u' % (self.device.GetDiscoverableTimeout())
elif self.cmd == 'SetDiscoverableTimeout':
if len(self.cmd_args) == 1:
self.device.SetDiscoverableTimeout(dbus.UInt32(self.cmd_args[0]))
else:
print 'Usage: %s -i <dev> SetDiscoverableTimeout timeout' % self.name
elif self.cmd == 'IsConnectable':
print self.device.IsConnectable()
elif self.cmd == 'IsDiscoverable':
print self.device.IsDiscoverable()
elif self.cmd == 'IsConnected':
if len(self.cmd_args) == 1:
print self.device.IsConnected(self.cmd_args[0])
else:
print 'Usage: %s -i <dev> IsConnected address' % self.name
elif self.cmd == 'ListConnections':
print self.device.ListConnections()
elif self.cmd == 'GetMajorClass':
print self.device.GetMajorClass()
elif self.cmd == 'ListAvailableMinorClasses':
print self.device.ListAvailableMinorClasses()
elif self.cmd == 'GetMinorClass':
print self.device.GetMinorClass()
elif self.cmd == 'SetMinorClass':
if len(self.cmd_args) == 1:
self.device.SetMinorClass(self.cmd_args[0])
else:
print 'Usage: %s -i <dev> SetMinorClass minor' % self.name
elif self.cmd == 'GetServiceClasses':
classes = self.device.GetServiceClasses()
for clas in classes:
print clas,
elif self.cmd == 'GetName':
print self.device.GetName()
elif self.cmd == 'SetName':
if len(self.cmd_args) == 1:
self.device.SetName(self.cmd_args[0])
else:
print 'Usage: %s -i <dev> SetName newname' % self.name
elif self.cmd == 'GetRemoteName':
if len(self.cmd_args) == 1:
print self.device.GetRemoteName(self.cmd_args[0])
else:
print 'Usage: %s -i <dev> GetRemoteName address' % self.name
elif self.cmd == 'GetRemoteVersion':
if len(self.cmd_args) == 1:
print self.device.GetRemoteVersion(self.cmd_args[0])
else:
print 'Usage: %s -i <dev> GetRemoteVersion address' % self.name
elif self.cmd == 'GetRemoteRevision':
if len(self.cmd_args) == 1:
print self.device.GetRemoteRevision(self.cmd_args[0])
else:
print 'Usage: %s -i <dev> GetRemoteRevision address' % self.name
elif self.cmd == 'GetRemoteManufacturer':
if len(self.cmd_args) == 1:
print self.device.GetRemoteManufacturer(self.cmd_args[0])
else:
print 'Usage: %s -i <dev> GetRemoteManufacturer address' % self.name
elif self.cmd == 'GetRemoteCompany':
if len(self.cmd_args) == 1:
print self.device.GetRemoteCompany(self.cmd_args[0])
else:
print 'Usage: %s -i <dev> GetRemoteCompany address' % self.name
elif self.cmd == 'GetRemoteAlias':
if len(self.cmd_args) == 1:
print self.device.GetRemoteAlias(self.cmd_args[0])
else:
print 'Usage: %s -i <dev> GetRemoteAlias address' % self.name
elif self.cmd == 'GetRemoteMajorClass':
if len(self.cmd_args) == 1:
print self.device.GetRemoteMajorClass(self.cmd_args[0])
else:
print 'Usage: %s -i <dev> GetRemoteMajorClass address' % self.name
elif self.cmd == 'GetRemoteMinorClass':
if len(self.cmd_args) == 1:
print self.device.GetRemoteMinorClass(self.cmd_args[0])
else:
print 'Usage: %s -i <dev> GetRemoteMinorClass address' % self.name
elif self.cmd == 'GetRemoteServiceClasses':
if len(self.cmd_args) == 1:
print self.device.GetRemoteServiceClasses(self.cmd_args[0])
else:
print 'Usage: %s -i <dev> GetRemoteServiceClasses address' % self.name
elif self.cmd == 'SetRemoteAlias':
if len(self.cmd_args) == 2:
self.device.SetRemoteAlias(self.cmd_args[0], self.cmd_args[1])
else:
print 'Usage: %s -i <dev> SetRemoteAlias address alias' % self.name
elif self.cmd == 'ClearRemoteAlias':
if len(self.cmd_args) == 1:
print self.device.ClearRemoteAlias(self.cmd_args[0])
else:
print 'Usage: %s -i <dev> ClearRemoteAlias address' % self.name
elif self.cmd == 'LastSeen':
if len(self.cmd_args) == 1:
print self.device.LastSeen(self.cmd_args[0])
else:
print 'Usage: %s -i <dev> LastSeen address' % self.name
elif self.cmd == 'LastUsed':
if len(self.cmd_args) == 1:
print self.device.LastUsed(self.cmd_args[0])
else:
print 'Usage: %s -i <dev> LastUsed address' % self.name
elif self.cmd == 'DisconnectRemoteDevice':
if len(self.cmd_args) == 1:
print self.device.LastUsed(self.cmd_args[0])
else:
print 'Usage: %s -i <dev> DisconnectRemoteDevice address' % self.name
elif self.cmd == 'CreateBonding':
if len(self.cmd_args) == 1:
print self.device.CreateBonding(self.cmd_args[0])
else:
print 'Usage: %s -i <dev> CreateBonding address' % self.name
elif self.cmd == 'RemoveBonding':
if len(self.cmd_args) == 1:
print self.device.RemoveBonding(self.cmd_args[0])
else:
print 'Usage: %s -i <dev> RemoveBonding address' % self.name
elif self.cmd == 'CancelBondingProcess':
if len(self.cmd_args) == 1:
print self.device.CancelBondingProcess(self.cmd_args[0])
else:
print 'Usage: %s -i <dev> CancelBondingProcess address' % self.name
elif self.cmd == 'HasBonding':
if len(self.cmd_args) == 1:
print self.device.HasBonding(self.cmd_args[0])
else:
print 'Usage: %s -i <dev> HasBonding address' % self.name
elif self.cmd == 'ListBondings':
bondings = self.device.ListBondings()
for bond in bondings:
print bond,
elif self.cmd == 'GetPinCodeLength':
if len(self.cmd_args) == 1:
print self.device.GetPinCodeLength(self.cmd_args[0])
else:
print 'Usage: %s -i <dev> GetPinCodeLength address' % self.name
elif self.cmd == 'GetEncryptionKeySize':
if len(self.cmd_args) == 1:
print self.device.GetEncryptionKeySize(self.cmd_args[0])
else:
print 'Usage: %s -i <dev> GetEncryptionKeySize address' % self.name
elif self.cmd == 'DiscoverDevices':
print self.device.DiscoverDevices()
elif self.cmd == 'DiscoverDevicesWithoutNameResolving':
print self.device.DiscoverDevicesWithoutNameResolving()
elif self.cmd == 'ListRemoteDevices':
devices = self.device.ListRemoteDevices()
for device in devices:
print device,
elif self.cmd == 'ListRecentRemoteDevices':
if len(self.cmd_args) == 1:
devices = self.device.ListRecentRemoteDevices(self.cmd_args[0])
for device in devices:
print device,
else:
print 'Usage: %s -i <dev> ListRecentRemoteDevices date' % self.name
else:
# FIXME: remove at future version
print 'Script Error: Method %s not found. Maybe a mispelled word.' % (self.cmd_args)
except dbus.DBusException, e:
print '%s failed: %s' % (self.cmd, e)
sys.exit(1)
def run(self):
# Manager methods
if self.listen:
self.dbus_mgr_sig_setup()
self.dbus_dev_sig_setup()
print 'Listening for events...'
if self.cmd in mgr_cmds:
try:
self.dbus_mgr_setup()
except dbus.DBusException, e:
print 'Failed to setup manager interface: %s' % e
sys.exit(1)
self.call_mgr_dbus_func()
elif self.cmd in dev_cmds:
try:
self.dbus_dev_setup()
except dbus.DBusException, e:
print 'Failed to setup device interface: %s' % e
sys.exit(1)
self.call_dev_dbus_func()
elif not self.listen:
print 'Unknown command: %s' % self.cmd
self.usage()
sys.exit(1)
if self.listen:
signal(SIGINT, self.signal_cb)
signal(SIGTERM, self.signal_cb)
self.main_loop = gobject.MainLoop()
self.main_loop.run()
if __name__ == '__main__':
gobject.threads_init()
dbus.glib.init_threads()
tester = Tester(sys.argv)
tester.run()