chrome-mpris2

#!/usr/bin/env python3

import json
import struct
import sys
from functools import partial

from gi.repository import Gio, GLib


def debug(msg):
    print(msg, file=sys.stderr, flush=True)


def make_streams_binary():
    sys.stdin = sys.stdin.detach()
    sys.stdout = sys.stdout.detach()


def escape_object_path(objpath):
    # We basically URI escape but instead of % we use _
    return (GLib.uri_escape_string(objpath, None, False)
            .replace(".", "%2E")
            .replace("-", "%2D")
            .replace("~", "%7E")
            .replace("_", "%5F")
            .replace("%", "_"))


def unescape_object_path(objpath):
    return GLib.uri_unescape_string(objpath.replace("_", "%"))


def encode_msg(msg):
    try:
        text = json.dumps(msg)
    except ValueError:
        return 0

    data = text.encode("utf-8")
    length_bytes = struct.pack("@i", len(data))
    written = sys.stdout.write(length_bytes) + sys.stdout.write(data)

    # We flush to make sure that Chrome gets the message *right now*
    sys.stdout.flush()
    return written


def decode_msg():
    # Read the message length (first 4 bytes).
    length_bytes = sys.stdin.read(4)
    if len(length_bytes) < 4:
        raise ValueError("unexpected end of input")

    # Unpack message length as 4 byte integer.
    length = struct.unpack("@i", length_bytes)[0]

    # Read the text (JSON object) of the message.
    text = sys.stdin.read(length).decode("utf-8")
    return json.loads(text)


def inheritdocstring(cls):
    for base in cls.mro():
        if base.__doc__ is not None:
            cls.__doc__ = base.__doc__
            break
    return cls


class DBusService:
    def __init__(self, conn, name, path, flags=Gio.BusNameOwnerFlags.NONE):
        self._conn = conn
        self._name = name
        self._path = path

        self.__registrations = []
        self._properties = {}
        self._methods = {}

        node_info = Gio.DBusNodeInfo.new_for_xml(self.__doc__)
        for iface in node_info.interfaces:
            self._methods[iface.name] = {
                meth.name: (tuple(a.signature for a in meth.in_args),
                            tuple(a.signature for a in meth.out_args))
                for meth in iface.methods
            }

            self._properties[iface.name] = {
                prop.name: prop.signature
                for prop in iface.properties
                if prop.flags & Gio.DBusPropertyInfoFlags.READABLE
            }

            self.__registrations.append(
                self._conn.register_object(path, iface, self.on_method_call)
            )

        self.__own_id = Gio.bus_own_name_on_connection(self._conn, name, flags)

    def unpublish(self):
        for i in self.__registrations:
            self._conn.unregister_object(i)
        Gio.bus_unown_name(self.__own_id)
        self._conn.close_sync()

    def on_method_call(self, conn, sender, objpath, iface_name, method_name,
                       params, invocation):

        if iface_name == "org.freedesktop.DBus.Properties":
            if method_name == "Get":
                self.get_property(conn, sender, objpath, *params, invocation)
            elif method_name == "GetAll":
                self.get_all_properties(conn, sender, objpath, *params,
                                        invocation)
            elif method_name == "Set":
                self.set_property(conn, sender, objpath, *params, invocation)
            return

        in_args, out_args = self._methods[iface_name][method_name]

        fd_list = invocation.get_message().get_unix_fd_list()
        # FIXME: also handle arrays or dicts with fds
        args = [fd_list.get(a) if sig == "h" else a
                for a, sig in zip(params.unpack(), in_args)]

        try:
            retval = getattr(self, method_name)(*args)

            if not out_args:
                retval = None
            elif len(out_args) == 1:
                retval = GLib.Variant("(%s)" % "".join(out_args), (retval,))
            else:
                retval = GLib.Variant("(%s)" % "".join(out_args), retval)

            invocation.return_value(retval)

        except Exception as e:
            e_type = type(e).__name__
            if "." not in e_type:
                e_type = "org.python." + e_type
            invocation.return_dbus_error(e_type, str(e))

    def get_property(self, conn, sender, objpath, iface_name, prop_name,
                     invocation):
        typ = self._properties[iface_name][prop_name]
        variant = GLib.Variant(typ, getattr(self, prop_name))
        invocation.return_value(GLib.Variant("(v)", (variant,)))

    def get_all_properties(self, conn, sender, objpath, iface_name,
                           invocation):
        all_props = {p: GLib.Variant(t, getattr(self, p))
                     for p, t in self._properties[iface_name].items()}
        invocation.return_value(GLib.Variant("(a{sv})", (all_props,)))

    def set_property(self, conn, sender, objpath, iface_name, prop_name, value,
                     invocation):
        setattr(self, prop_name, value)
        invocation.return_value(None)

    # SIGNALS
    def PropertiesChanged(self, iface_name, changed_props, invalidated_props):
        # we assume readable properties
        changed_props = {
            p: GLib.Variant(self._properties[iface_name][p], v)
            for p, v in changed_props.items()
        }

        self._conn.emit_signal(
            None,
            self._path,
            "org.freedesktop.DBus.Properties",
            "PropertiesChanged",
            GLib.Variant("(sa{sv}as)", (
                iface_name, changed_props, invalidated_props)))


class MediaPlayer2(DBusService):
    """
    <!DOCTYPE node PUBLIC
      "-//freedesktop//DTD D-BUS Object Introspection 1.0//EN"
      "http://www.freedesktop.org/standards/dbus/1.0/introspect.dtd">
    <node>
    <interface name="org.mpris.MediaPlayer2">
        <property name="CanQuit"             type="b"  access="read"/>
        <property name="CanRaise"            type="b"  access="read"/>
        <property name="CanSetFullscreen"    type="b"  access="read">
            <annotation name="org.mpris.MediaPlayer2.property.optional" value="true"/>
        </property>
        <!-- <property name="DesktopEntry"        type="s"  access="read"> -->
        <!--    <annotation name="org.mpris.MediaPlayer2.property.optional" value="true"/> -->
        <!-- </property> -->
        <property name="Fullscreen"          type="b"  access="readwrite">
            <annotation name="org.mpris.MediaPlayer2.property.optional" value="true"/>
        </property>
        <property name="HasTrackList"        type="b"  access="read"/>
        <property name="Identity"            type="s"  access="read"/>
        <property name="SupportedMimeTypes"  type="as" access="read"/>
        <property name="SupportedUriSchemes" type="as" access="read"/>

        <method name="Quit"/>
        <method name="Raise"/>
    </interface>

    <interface name="org.mpris.MediaPlayer2.Player">
        <property name="CanControl"     type="b"     access="read">
            <annotation name="org.freedesktop.DBus.Property.EmitsChangedSignal" value="false"/>
        </property>
        <property name="CanGoNext"      type="b"     access="read"/>
        <property name="CanGoPrevious"  type="b"     access="read"/>
        <property name="CanPause"       type="b"     access="read"/>
        <property name="CanPlay"        type="b"     access="read"/>
        <property name="CanSeek"        type="b"     access="read"/>
        <property name="LoopStatus"     type="s"     access="readwrite">
            <annotation name="org.mpris.MediaPlayer2.property.optional" value="true"/>
        </property>
        <property name="MaximumRate"    type="d"     access="read"/>
        <property name="Metadata"       type="a{sv}" access="read"/>
        <property name="MinimumRate"    type="d"     access="read"/>
        <property name="PlaybackStatus" type="s"     access="read"/>
        <property name="Rate"           type="d"     access="readwrite"/>
        <property name="Position"       type="x"     access="read">
            <annotation name="org.freedesktop.DBus.Property.EmitsChangedSignal" value="false"/>
        </property>
        <property name="Shuffle"        type="b"     access="readwrite">
            <annotation name="org.mpris.MediaPlayer2.property.optional" value="true"/>
        </property>
        <property name="Volume"         type="d"     access="readwrite"/>

        <signal name="Seeked">
            <arg name="Position" type="x"/>
        </signal>

        <method name="Next"/>
        <method name="OpenUri">
            <arg direction="in" type="s" name="Uri"/>
        </method>
        <method name="Pause"/>
        <method name="Play"/>
        <method name="PlayPause"/>
        <method name="Previous"/>
        <method name="Seek">
            <arg direction="in" type="x" name="Offset"/>
        </method>
        <method name="SetPosition">
            <arg direction="in" type="o" name="TrackId"/>
            <arg direction="in" type="x" name="Position"/>
        </method>
        <method name="Stop"/>
    </interface>
    </node>
    """

    def __init__(self, conn, name, tabid):
        self.__tabid = tabid
        self.__position_callbacks = []
        super().__init__(conn, name, "/org/mpris/MediaPlayer2")

    def __getattr__(self, attr):
        if any(attr in iface for iface in self._methods.values()):
            if attr == "SetPosition":
                def SetPosition(trackid, position):
                    if trackid != "/org/mpris/MediaPlayer2/TrackList/NoTrack":
                        self._send_msg("SetPosition",
                                       unescape_object_path(trackid[1:]),
                                       position)
                return SetPosition
            return partial(self._send_msg, attr)
        return AttributeError(attr)

    # PROPERTIES
    def get_property(self, conn, sender, objpath, iface_name, prop_name,
                     invocation):
        if (iface_name == "org.mpris.MediaPlayer2.Player"
                and prop_name == "Position"):

            def callback(pos):
                variant = GLib.Variant("x", pos)
                invocation.return_value(GLib.Variant("(v)", (variant,)))

            self.__position_callbacks.append(callback)
            self._send_msg("Get", "org.mpris.MediaPlayer2.Player", "Position")

        else:
            super().get_property(conn, sender, objpath, iface_name, prop_name,
                                 invocation)

    def get_all_properties(self, conn, sender, objpath, iface_name,
                           invocation):
        if iface_name == "org.mpris.MediaPlayer2.Player":
            non_async = {p: GLib.Variant(t, getattr(self, p))
                         for p, t in self._properties[iface_name].items()
                         if p != "Position"}

            def callback(pos):
                non_async["Position"] = GLib.Variant("x", pos)
                invocation.return_value(GLib.Variant("(a{sv})", (non_async,)))

            self.__position_callbacks.append(callback)
            self._send_msg("Get", "org.mpris.MediaPlayer2.Player", "Position")

        else:
            super().get_all_properties(conn, sender, objpath, iface_name,
                                       invocation)

    def set_property(self, conn, sender, objpath, iface_name, prop_name, value,
                     invocation):
        self._send_msg("Set", iface_name, prop_name, value)
        invocation.return_value(None)

    # SIGNALS
    def Seeked(self, position):
        self._conn.emit_signal(
            None,
            self._path,
            "org.mpris.MediaPlayer2.Player",
            "Seeked",
            GLib.Variant("(x)", (position,)))

    def PropertiesChanged(self, changed_props):
        if "Metadata" in changed_props:
            changed_props["Metadata"] = (
                self._patch_metadata(changed_props["Metadata"]))

        vars(self).update(changed_props)

        # dict.keys() is actually a set
        base_props = (changed_props.keys() &
                      self._properties["org.mpris.MediaPlayer2"])
        player_props = (changed_props.keys() &
                        self._properties["org.mpris.MediaPlayer2.Player"])
        if base_props:
            super().PropertiesChanged("org.mpris.MediaPlayer2", {
                p: changed_props[p] for p in base_props}, [])
        if player_props:
            super().PropertiesChanged("org.mpris.MediaPlayer2.Player", {
                p: changed_props[p] for p in player_props}, [])

    @staticmethod
    def _patch_metadata(metadata):
        metadata_types = {
            "mpris:trackid": "o",
            "mpris:length": "x",
            "mpris:artUrl": "s",
            "xesam:url": "s",
            "xesam:title": "s",
            "xesam:artist": "as",
        }

        if "mpris:trackid" in metadata:
            metadata["mpris:trackid"] = \
                "/" + escape_object_path(metadata["mpris:trackid"])
        return {
            k: GLib.Variant(metadata_types[k], v) for k, v in metadata.items()
            if k in metadata_types and v is not None
        }

    def _send_msg(self, method, *args):
        encode_msg({
            "tabId": self.__tabid,
            "method": method,
            "args": args,
        })

    def _msg_recv(self, msg):
        if msg["type"] == "seeked":
            self.Seeked(*msg["args"])
        elif msg["type"] == "changed":
            self.PropertiesChanged(*msg["args"])

        elif msg["type"] == "return":
            if msg["method"] == "Get":
                # we only ever ask for Position
                for cb in self.__position_callbacks:
                    cb(msg["args"])
                self.position_callbacks = []


@inheritdocstring
class YouTube(MediaPlayer2):
    def __init__(self, conn, name, tabid):
        super().__init__(conn, name, tabid)

        self.CanQuit = False
        self.CanRaise = False
        self.CanSetFullscreen = False
        self.HasTrackList = False
        self.SupportedMimeTypes = []
        self.SupportedUriSchemes = ["http", "https"]

        self.Fullscreen = False
        self.LoopStatus = "None"
        self.Metadata = {}

        self.PlaybackStatus = "Stopped"
        self.Rate = 1
        self.Shuffle = False
        self.Volume = 1

        self.Identity = "youtube"
        self.CanControl = True
        self.CanPlay = True
        self.CanPause = True
        self.CanSeek = True
        self.CanSetFullscreen = False

        self.MinimumRate = 0.25
        self.MaximumRate = 2

        self.CanGoNext = False
        self.CanGoPrevious = False


@inheritdocstring
class SoundCloud(MediaPlayer2):
    def __init__(self, conn, name, tabid):
        super().__init__(conn, name, tabid)

        self.CanQuit = False
        self.CanRaise = False
        self.CanSetFullscreen = False
        self.HasTrackList = False
        self.SupportedMimeTypes = []
        self.SupportedUriSchemes = ["http", "https"]

        self.Fullscreen = False
        self.LoopStatus = "None"
        self.Metadata = {}

        self.PlaybackStatus = "Stopped"
        self.Rate = 1
        self.Shuffle = False
        self.Volume = 1

        self.Identity = "soundcloud"
        self.CanControl = True
        self.CanPlay = True
        self.CanPause = True
        self.CanSeek = False
        self.CanSetFullscreen = False

        self.MinimumRate = 1
        self.MaximumRate = 1

        self.CanGoNext = False
        self.CanGoPrevious = False


players = {
    "youtube": YouTube,
    "soundcloud": SoundCloud,
}


def main():
    mainloop = GLib.MainLoop()

    make_streams_binary()

    active_players = {}

    def message_handler(chan, condition):
        msg = decode_msg()
        # debug(msg)

        if msg["type"] not in ("return", "changed", "seeked", "quit"):
            return True

        tabid = msg["tabId"]

        try:
            player = active_players[tabid]
        except KeyError:
            # Many terrible things can happen to tabs; they can crash or be
            # discarded.  For these cases out background.js should still inform
            # us.  However it may be that that tab has already quit e.g. it may
            # be in YouTube's main page when it crashes.  In these cases we
            # don't want to create a player just to immediately destroy it.
            if msg["type"] == "quit":
                return True

            connflags = (Gio.DBusConnectionFlags.AUTHENTICATION_CLIENT
                         | Gio.DBusConnectionFlags.MESSAGE_BUS_CONNECTION)

            addr = Gio.dbus_address_get_for_bus_sync(Gio.BusType.SESSION)
            conn = Gio.DBusConnection.new_for_address_sync(addr, connflags)
            conn.set_exit_on_close(True)

            name = "org.mpris.MediaPlayer2.chrome"
            if active_players:
                # if we are exposing more than one player we need to give them
                # unique names
                name += ".tab%d" % tabid

            # malformed message
            if "source" not in msg or msg["source"] not in players:
                return True

            player_class = players[msg["source"]]
            player = player_class(conn, name, tabid)
            active_players[tabid] = player

        if msg["type"] == "quit":
            player.unpublish()
            del active_players[tabid]
        else:
            player._msg_recv(msg)

        # otherwise GLib will remove our watch
        return True

    chan = GLib.IOChannel.unix_new(sys.stdin.fileno())
    GLib.io_add_watch(chan, GLib.IOCondition.IN, message_handler)
    GLib.io_add_watch(chan, GLib.IOCondition.HUP, lambda *_: mainloop.quit())

    mainloop.run()

if __name__ == "__main__":
    main()