kaylee/kayleevc/plugins/mpris.py

# This is part of Kaylee
# -- this code is licensed GPLv3
# Copyright 2015-2017 Clayton G. Hobbs
# Portions Copyright 2013 Jezra

"""MPRIS media player control plugin

This plugin allows control of a media player supporting the Media
Player Remote Interfacing Specification (MPRIS) D-Bus interface.  To
use this plugin, you must first install `pydbus
<https://github.com/LEW21/pydbus>`__ for Python 3.  This plugin takes
no configuration, so load it as follows::

    ".mpris": {}

Note: if your microphone can hear the audio produced by your speakers,
it is recommended to load PulseAudio's module-echo-cancel when using
this plugin.  Recognition while music is playing may still be mediocre,
but without module-echo-cancel it has almost no chance of working.
"""

from gi.repository import GLib
from pydbus import SessionBus

from . import PluginBase, Handler, HandlerFailure


class Plugin(PluginBase):
    """Main class for the MPRIS plugin"""

    def __init__(self, config, name):
        """Initialize the MPRIS plugin"""
        super().__init__(config, name)

        # Get the D-Bus proxy object
        self._bus = SessionBus()
        self._dbus_proxy = self._bus.get('.DBus')

        self.commands = {
            'next song': MPRISNextHandler,
            'next video': MPRISNextHandler,
            'pause music': MPRISPauseHandler,
            'pause video': MPRISPauseHandler,
            'play music': MPRISPlayHandler,
            'play video': MPRISPlayHandler,
            'previous song': MPRISPreviousHandler,
            'previous video': MPRISPreviousHandler,
            'whats playing': MPRISTrackInfoHandler
        }

        self.corpus_strings.update(self.commands)

    def _mpris_proxy(self):
        """Get a proxy for the first MPRIS media player, or None"""
        try:
            # Get the bus name of the first MPRIS media player
            bus_name = [name for name in self._dbus_proxy.ListNames()
                        if name.startswith('org.mpris.MediaPlayer2')][0]
            # Get the proxy
            return self._bus.get(bus_name, '/org/mpris/MediaPlayer2')
        except IndexError:
            return None

    def get_handler(self, text):
        """Return a handler if a recognized command is heard"""
        if text not in self.corpus_strings:
            return None
        p = self._mpris_proxy()
        if p is None:
            return None
        # Make a handler for the command we heard
        return self.commands[text](1, p)


class MPRISHandler(Handler):
    """Base class for MPRIS plugin handlers"""

    def __init__(self, confidence, proxy):
        """Store the MPRIS proxy object"""
        super().__init__(confidence)
        self.proxy = proxy


class MPRISNextHandler(MPRISHandler):
    """Handler for the MPRIS Next method"""

    def __call__(self, tts):
        """Call the MPRIS Next method"""
        try:
            self.proxy.Next()
        except GLib.Error:
            raise HandlerFailure()


class MPRISPauseHandler(MPRISHandler):
    """Handler for the MPRIS Pause method"""

    def __call__(self, tts):
        """Call the MPRIS Pause method"""
        try:
            self.proxy.Pause()
        except GLib.Error:
            raise HandlerFailure()


class MPRISPlayHandler(MPRISHandler):
    """Handler for the MPRIS Play method"""

    def __call__(self, tts):
        """Call the MPRIS Play method"""
        try:
            self.proxy.Play()
        except GLib.Error:
            raise HandlerFailure()


class MPRISPreviousHandler(MPRISHandler):
    """Handler for the MPRIS Previous method"""

    def __call__(self, tts):
        """Call the MPRIS Previous method"""
        try:
            self.proxy.Previous()
        except GLib.Error:
            raise HandlerFailure()


class MPRISTrackInfoHandler(MPRISHandler):
    """Handler for getting info about the currently playing track from MPRIS"""

    def __call__(self, tts):
        """Get and speak info about the currently playing track"""
        # Get metadata about the player
        try:
            metadata = self.proxy.Metadata
        except GLib.Error:
            raise HandlerFailure()

        # Get the title
        try:
            title = metadata['xesam:title']
        except KeyError:
            raise HandlerFailure()

        # Get the artist list (optional)
        try:
            artist_list = metadata['xesam:artist']
        except KeyError:
            artist_text = ''
        else:
            # Get the artists in a speakable format
            artist_text = ' by ' + ', '.join(
                    artist_list[:-2] + [', and '.join(artist_list[-2:])])


        # Speak the information
        tts(f"{title}{artist_text}")