| #!/usr/bin/python |
| # SPDX-License-Identifier: LGPL-2.1-or-later |
| |
| from __future__ import print_function |
| |
| 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 |
| import bluezutils |
| |
| PLAYER_IFACE = 'org.mpris.MediaPlayer2.Player' |
| DBUS_OM_IFACE = 'org.freedesktop.DBus.ObjectManager' |
| DBUS_PROP_IFACE = 'org.freedesktop.DBus.Properties' |
| |
| class InvalidArgsException(dbus.exceptions.DBusException): |
| _dbus_error_name = 'org.freedesktop.DBus.Error.InvalidArgs' |
| |
| class Player(dbus.service.Object): |
| def __init__(self, bus, path, obj): |
| self.path = path |
| dbus.service.Object.__init__(self, bus, self.path) |
| |
| if obj != None: |
| mp = dbus.Interface(bus.get_object("org.bluez", obj), |
| "org.bluez.MediaPlayer1") |
| prop = dbus.Interface(bus.get_object("org.bluez", obj), |
| "org.freedesktop.DBus.Properties") |
| |
| self.properties = prop.GetAll("org.bluez.MediaPlayer1") |
| |
| bus.add_signal_receiver(self.properties_changed, path = obj, |
| dbus_interface = "org.freedesktop.DBus.Properties", |
| signal_name = "PropertiesChanged") |
| else: |
| self.track = dbus.Dictionary({"xesam:title" : "Title", |
| "xesam:artist" : ["Artist"], |
| "xesam:album" : "Album", |
| "xesam:genre" : ["Genre"], |
| "xesam:trackNumber" : dbus.Int32(1), |
| "mpris:length" : dbus.Int64(10000) }, |
| signature="sv") |
| |
| self.properties = dbus.Dictionary({"PlaybackStatus" : "playing", |
| "Identity" : "SimplePlayer", |
| "LoopStatus" : "None", |
| "Rate" : dbus.Double(1.0), |
| "Shuffle" : dbus.Boolean(False), |
| "Metadata" : self.track, |
| "Volume" : dbus.Double(1.0), |
| "Position" : dbus.Int64(0), |
| "MinimumRate" : dbus.Double(1.0), |
| "MaximumRate" : dbus.Double(1.0), |
| "CanGoNext" : dbus.Boolean(False), |
| "CanGoPrevious" : dbus.Boolean(False), |
| "CanPlay" : dbus.Boolean(False), |
| "CanSeek" : dbus.Boolean(False), |
| "CanControl" : dbus.Boolean(False), |
| }, |
| signature="sv") |
| |
| print('Register media player with:\n\tProperties: %s' \ |
| % (self.properties)) |
| handler = InputHandler(self) |
| GObject.io_add_watch(sys.stdin, GObject.IO_IN, handler.handle) |
| |
| @dbus.service.method("org.freedesktop.DBus.Properties", |
| in_signature="ssv", out_signature="") |
| def Set(self, interface, key, value): |
| print("Set (%s, %s)" % (key, value), file=sys.stderr) |
| return |
| |
| def get_properties(self): |
| return self.properties |
| |
| def get_path(self): |
| return dbus.ObjectPath(self.path) |
| |
| @dbus.service.method("org.freedesktop.DBus.Properties", |
| in_signature='s', out_signature='a{sv}') |
| def GetAll(self, interface): |
| if interface != PLAYER_IFACE: |
| raise InvalidArgsException() |
| |
| return self.get_properties() |
| |
| @dbus.service.signal("org.freedesktop.DBus.Properties", |
| signature="sa{sv}as") |
| def PropertiesChanged(self, interface, properties, |
| invalidated = dbus.Array()): |
| """PropertiesChanged(interface, properties, invalidated) |
| |
| Send a PropertiesChanged signal. 'properties' is a dictionary |
| containing string parameters as specified in doc/media-api.txt. |
| """ |
| pass |
| |
| def help(self, func): |
| help(self.__class__.__dict__[func]) |
| |
| def properties_changed(self, interface, properties, invalidated): |
| print("properties_changed(%s, %s)" % (properties, invalidated)) |
| |
| self.PropertiesChanged(interface, properties, invalidated) |
| |
| class InputHandler: |
| commands = { 'PropertiesChanged': '(interface, properties)', |
| 'help': '(cmd)' } |
| def __init__(self, player): |
| self.player = player |
| print('\n\nAvailable commands:') |
| for cmd in self.commands: |
| print('\t', cmd, self.commands[cmd], sep='') |
| |
| print("\nUse python syntax to pass arguments to available methods.\n" \ |
| "E.g.: PropertiesChanged({'Metadata' : {'Title': 'My title', \ |
| 'Album': 'my album' }})") |
| self.prompt() |
| |
| def prompt(self): |
| print('\n>>> ', end='') |
| sys.stdout.flush() |
| |
| def handle(self, fd, condition): |
| s = os.read(fd.fileno(), 1024).strip() |
| try: |
| cmd = s[:s.find('(')] |
| if not cmd in self.commands: |
| print("Unknown command ", cmd) |
| except ValueError: |
| print("Malformed command") |
| return True |
| try: |
| exec "self.player.%s" % s |
| except Exception as e: |
| print(e) |
| pass |
| self.prompt() |
| return True |
| |
| class Application(dbus.service.Object): |
| def __init__(self, bus, path, obj): |
| self.path = '/' |
| self.players = [] |
| dbus.service.Object.__init__(self, bus, self.path) |
| self.add_player(Player(bus, path, obj)) |
| |
| def get_path(self): |
| return dbus.ObjectPath(self.path) |
| |
| def add_player(self, player): |
| self.players.append(player) |
| |
| @dbus.service.method(DBUS_OM_IFACE, out_signature='a{oa{sa{sv}}}') |
| def GetManagedObjects(self): |
| response = {} |
| print('GetManagedObjects') |
| |
| for player in self.players: |
| response[player.get_path()] = { PLAYER_IFACE: |
| player.get_properties() } |
| |
| return response |
| |
| def register_app_cb(): |
| print('Media application registered') |
| |
| |
| def register_app_error_cb(error): |
| print('Failed to register application: ' + str(error)) |
| mainloop.quit() |
| |
| if __name__ == '__main__': |
| dbus.mainloop.glib.DBusGMainLoop(set_as_default=True) |
| |
| bus = dbus.SystemBus() |
| |
| if len(sys.argv) > 1: |
| path = bluezutils.find_adapter(sys.argv[1]).object_path |
| else: |
| path = bluezutils.find_adapter().object_path |
| |
| media = dbus.Interface(bus.get_object("org.bluez", path), |
| "org.bluez.Media1") |
| |
| path = "/test/player" |
| |
| if len(sys.argv) > 2: |
| app = Application(bus, path, sys.argv[2]) |
| else: |
| app = Application(bus, path, None) |
| |
| mainloop = GObject.MainLoop() |
| |
| media.RegisterApplication(app.get_path(), {}, |
| reply_handler=register_app_cb, |
| error_handler=register_app_error_cb) |
| |
| mainloop.run() |