| #!/usr/bin/python | |
| from __future__ import absolute_import, print_function, unicode_literals | |
| import optparse | |
| import os | |
| import sys | |
| import dbus | |
| import dbus.service | |
| import dbus.mainloop.glib | |
| try: | |
| from gi.repository import GObject | |
| except ImportError: | |
| import gobject as GObject | |
| BUS_NAME = 'org.bluez' | |
| ALERT_INTERFACE = 'org.bluez.Alert1' | |
| ALERT_AGENT_INTERFACE = 'org.bluez.AlertAgent1' | |
| BLUEZ_OBJECT_PATH = '/org/bluez' | |
| TEST_OBJECT_PATH = '/org/bluez/test' | |
| class AlertAgent(dbus.service.Object): | |
| def __init__(self, bus, object_path, alert, mainloop): | |
| dbus.service.Object.__init__(self, bus, object_path) | |
| self.alert = alert | |
| self.mainloop = mainloop | |
| @dbus.service.method(ALERT_AGENT_INTERFACE, in_signature='', | |
| out_signature='') | |
| def MuteOnce(self): | |
| print('method MuteOnce() was called') | |
| self.alert.NewAlert('ringer', 1, 'not active') | |
| @dbus.service.method(ALERT_AGENT_INTERFACE, in_signature='s', | |
| out_signature='') | |
| def SetRinger(self, mode): | |
| print('method SetRinger(%s) was called' % mode) | |
| self.alert.NewAlert('ringer', 1, mode) | |
| @dbus.service.method(ALERT_AGENT_INTERFACE, in_signature='', | |
| out_signature='') | |
| def Release(self): | |
| print('method Release() was called') | |
| self.mainloop.quit() | |
| def print_command_line(options): | |
| if not options.verbose: | |
| return False | |
| print('-w: ' + str(options.wait)) | |
| if options.times: | |
| print('-t: ' + str(options.times)) | |
| if options.register: | |
| print('-r: ' + options.register) | |
| else: | |
| print('-r: ' + str(None)) | |
| if options.new_alert: | |
| print('-n:') | |
| for i in options.new_alert: | |
| print(' ' + i[0] + ', ' + i[1] + ', ' + i[2]) | |
| else: | |
| print('-n: ' + str(None)) | |
| if options.unread_alert: | |
| print('-u:') | |
| for i in options.unread_alert: | |
| print(' ' + i[0] + ', ' + i[1]) | |
| else: | |
| print('-u: ' + str(None)) | |
| print() | |
| return True | |
| def read_count(param): | |
| try: | |
| return int(param) | |
| except ValueError: | |
| print('<count> must be integer, not \"%s\"' % param) | |
| sys.exit(1) | |
| def new_alert(alert, params): | |
| if not params: | |
| return False | |
| for param in params: | |
| category = param[0] | |
| count = read_count(param[1]) | |
| description = param[2] | |
| alert.NewAlert(category, count, description) | |
| def unread_alert(alert, params): | |
| if not params: | |
| return False | |
| for param in params: | |
| category = param[0] | |
| count = read_count(param[1]) | |
| alert.UnreadAlert(category, count) | |
| option_list = [ | |
| optparse.make_option('-v', None, | |
| action = 'store_true', | |
| default = False, | |
| dest = 'verbose', | |
| help = 'verbose'), | |
| optparse.make_option('-w', None, | |
| action = 'store_true', | |
| default = False, | |
| dest = 'wait', | |
| help = 'wait for dbus events'), | |
| optparse.make_option('-t', None, | |
| action = 'store', | |
| default = 1, | |
| type = "int", | |
| dest = 'times', | |
| help = 'repeat UnreadAlert/NewAlert <times> times', | |
| metavar = '<times>'), | |
| optparse.make_option('-r', None, | |
| action = 'store', | |
| dest = 'register', | |
| type = 'string', | |
| metavar = '<category>', | |
| help = 'register alert'), | |
| optparse.make_option('-n', None, | |
| action = 'append', | |
| dest = 'new_alert', | |
| type = 'string', | |
| nargs = 3, | |
| metavar = '<category> <count> <description>', | |
| help = 'send new alert'), | |
| optparse.make_option('-u', None, | |
| action = 'append', | |
| dest = 'unread_alert', | |
| type = 'string', | |
| nargs = 2, | |
| metavar = '<category> <count>', | |
| help = 'send unread alert'), | |
| ] | |
| parser = optparse.OptionParser(option_list=option_list) | |
| parser.disable_interspersed_args() | |
| (options, args) = parser.parse_args() | |
| dbus.mainloop.glib.DBusGMainLoop(set_as_default=True) | |
| bus = dbus.SystemBus() | |
| mainloop = GObject.MainLoop() | |
| alert = dbus.Interface(bus.get_object(BUS_NAME, BLUEZ_OBJECT_PATH), | |
| ALERT_INTERFACE) | |
| alert_agent = AlertAgent(bus, TEST_OBJECT_PATH, alert, mainloop) | |
| print_command_line(options) | |
| if not (options.register or options.new_alert or options.unread_alert or | |
| options.wait): | |
| parser.print_usage() | |
| sys.exit(1) | |
| if options.register: | |
| alert.RegisterAlert(options.register, TEST_OBJECT_PATH) | |
| times = 0 | |
| while times < options.times: | |
| times += 1 | |
| new_alert(alert, options.new_alert) | |
| unread_alert(alert, options.unread_alert) | |
| if not options.wait: | |
| sys.exit(0) | |
| try: | |
| mainloop.run() | |
| except: | |
| pass |