PyGObject MPRIS

mpris_dbus.py

#
# Copyright (C) 2017 Jason Gray <[email protected]>
#
# This program is free software: you can redistribute it and/or modify it
# under the terms of the GNU General Public License version 3, as published
# by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranties of
# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
# PURPOSE.  See the GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program.  If not, see <http://www.gnu.org/licenses/>.
# END LICENSE

import gi
gi.require_version('Gtk', '3.0')
from gi.repository import Gtk, GLib, GObject, Gio

GObject.threads_init()

NAME_KEY = 'name'
OWNER_KEY = 'owner'

GLIB_VAR_STRING_TUPLE_SIG = '(s)'

DBUS_CALL_TIMEOUT = -1
RESULT_INDEX = 0

DBUS_IFACE = 'org.freedesktop.DBus'
DBUS_NAME = 'org.freedesktop.DBus'
DBUS_PATH = '/org/freedesktop/DBus'

PROPS_IFACE = 'org.freedesktop.DBus.Properties'
PROPS_PATH = '/org/mpris/MediaPlayer2'

MPRIS_IFACE = 'org.mpris.MediaPlayer2'
MPRIS_PATH = '/org/mpris/MediaPlayer2'
MPRIS_PREFIX = 'org.mpris.'

MPRIS_PLAYER_IFACE = 'org.mpris.MediaPlayer2.Player'
MPRIS_PLAYER_PATH = '/org/mpris/MediaPlayer2'
MPRIS_PLAYER_BASE_NAME = 'org.mpris.MediaPlayer2'

DBUS_METHOD_GET_NAME_OWNER = 'GetNameOwner'
DBUS_METHOD_LIST_NAMES = 'ListNames'

DBUS_SIGNAL_NAME_OWNER_CHANGED = 'NameOwnerChanged'

PROPS_SIGNAL_PROPERTIES_CHANGED = 'PropertiesChanged'

MPRIS_METHOD_RAISE = 'Raise'

MPRIS_PROP_CAN_RAISE = 'CanRaise'
MPRIS_PROP_DESKTOP_ENTRY = 'DesktopEntry'

MPRIS_PLAYER_METHOD_NEXT = 'Next'
MPRIS_PLAYER_METHOD_PREVIOUS = 'Previous'
MPRIS_PLAYER_METHOD_PLAYPAUSE = 'PlayPause'

MPRIS_PLAYER_PROP_PLAYBACK_STATUS = 'PlaybackStatus'
MPRIS_PLAYER_PROP_METADATA = 'Metadata'
MPRIS_PLAYER_PROP_CAN_GO_NEXT = 'CanGoNext'
MPRIS_PLAYER_PROP_CAN_GO_PREVIOUS = 'CanGoPrevious'
MPRIS_PLAYER_PROP_CAN_PLAY = 'CanPlay'
MPRIS_PLAYER_PROP_CAN_PAUSE = 'CanPause'

PLAYBACK_STATUS_PLAYING = 'Playing'
PLAYBACK_STATUS_PAUSED = 'Paused'
PLAYBACK_STATUS_STOPPED = 'Stopped'

DBUS_PROXY_SIGNAL_PLAYER_ADDED = 'player-added'
DBUS_PROXY_SIGNAL_PLAYER_REMOVED = 'player-removed'
DBUS_PROXY_SIGNAL_CHANGE_OWNER = 'change-owner'

class DBusProxy(Gio.DBusProxy):
    __gtype_name__ = 'DBusProxy'
    __gsignals__ = {
        DBUS_PROXY_SIGNAL_PLAYER_ADDED: (GObject.SignalFlags.RUN_FIRST, None, (int, GObject.TYPE_PYOBJECT,)),
        DBUS_PROXY_SIGNAL_PLAYER_REMOVED: (GObject.SignalFlags.RUN_FIRST, None, (int, GObject.TYPE_PYOBJECT)),
        DBUS_PROXY_SIGNAL_CHANGE_OWNER: (GObject.SignalFlags.RUN_FIRST, None, (int, GObject.TYPE_PYOBJECT,)),
    }

    def __init__(self, **kwargs):
        super().__init__(
            g_bus_type=Gio.BusType.SESSION,
            g_interface_name=DBUS_IFACE,
            g_name=DBUS_NAME,
            g_object_path=DBUS_PATH,
            **kwargs
        )

        self._player_list = []

    @classmethod
    def async_init(cls, callback):
        def on_init_finish(self, result, data):
            try:
                self.init_finish(result)
            except GLib.Error as e:
                pass
            else:
                if not self.get_name_owner():
                    pass
                else:
                    callback(self)

        self = cls()
        self.init_async(GLib.PRIORITY_DEFAULT, None, on_init_finish, None)

    def refresh_players(self):
        def on_list_names_finish(self, result, data):
            try:
                names = self.call_finish(result).unpack()[RESULT_INDEX]
            except GLib.Error as e:
                pass
            else:
                names = [name for name in names if name.startswith(MPRIS_PREFIX)]
                names.sort()
                for name in names:
                    self.call(
                        DBUS_METHOD_GET_NAME_OWNER,
                        GLib.Variant(GLIB_VAR_STRING_TUPLE_SIG, (name,)),
                        Gio.DBusCallFlags.NONE,
                        DBUS_CALL_TIMEOUT,
                        None,
                        on_get_name_owner_finish,
                        name,
                    )

        def on_get_name_owner_finish(self, result, name):
            try:
                owner = self.call_finish(result).unpack()[RESULT_INDEX]
            except GLib.Error as e:
                pass
            else:
                self._add_player(name, owner)

        self.call(
            DBUS_METHOD_LIST_NAMES,
            None,
            Gio.DBusCallFlags.NONE,
            DBUS_CALL_TIMEOUT,
            None,
            on_list_names_finish,
            None,
        )

    def _add_player(self, name, owner):
        name_owner = {NAME_KEY: name, OWNER_KEY: owner}
        if name_owner not in self._player_list:
            index = len(self._player_list)
            self._player_list.append(name_owner)
            self.emit(DBUS_PROXY_SIGNAL_PLAYER_ADDED, index, name_owner)

    def _remove_player(self, name):
        for index, item in enumerate(self._player_list):
            if item[NAME_KEY] == name:
                self._player_list.remove(item)
                self.emit(DBUS_PROXY_SIGNAL_PLAYER_REMOVED, index, item)
                break

    def _change_player_owner(self, name, owner):
        for index, item in enumerate(self._player_list):
            if item[NAME_KEY] == name:
                name_owner = {NAME_KEY: name, OWNER_KEY: owner}
                self._player_list[index] = name_owner
                self.emit(DBUS_PROXY_SIGNAL_CHANGE_OWNER, index, name_owner)
                break

    def do_g_signal(self, sender_name, signal_name, parameters):
        if signal_name != DBUS_SIGNAL_NAME_OWNER_CHANGED:
            return
        try:
            name, old_owner, new_owner = parameters.unpack()
        except ValueError:
            pass
        else:
            if not name.startswith(MPRIS_PREFIX):
                return
            elif new_owner and not old_owner:
                self._add_player(name, new_owner)
            elif old_owner and not new_owner:
                self._remove_player(name)
            else:
                self._change_player_owner(name, new_owner)

    def __getattr__(self, name):
        # PyGObject ships an override that breaks our usage.
        return object.__getattr__(self, name)


class PlayerManager(GObject.Object):
    __gtype_name__ = 'Test'

    def __init__(self):
        super().__init__()
        DBusProxy.async_init(self._on_dbus_proxy_init_finish)
    
    def _on_dbus_proxy_init_finish(self, dbus_proxy):
        self.dbus_proxy = dbus_proxy
        self.dbus_proxy.connect(DBUS_PROXY_SIGNAL_PLAYER_ADDED, self._on_player_added)
        self.dbus_proxy.connect(DBUS_PROXY_SIGNAL_PLAYER_REMOVED, self._on_player_removed)
        self.dbus_proxy.connect(DBUS_PROXY_SIGNAL_CHANGE_OWNER, self._on_change_owner)
        self.dbus_proxy.refresh_players()

    def _on_player_added(self, dbus_proxy, index, player):
        print('{} added at index: {}'.format(player, index))

    def _on_player_removed(self, dbus_proxy, index, player):
        print('{} removed at index: {}'.format(player, index))

    def _on_change_owner(self, dbus_proxy, index, player):
        print('{} changed owner at index: {}'.format(player, index))        

test = PlayerManager() 
Gtk.main()