path: root/mpris_to_text.py

#!/usr/bin/env python

import sys
import re
import threading
import argparse
import signal

from gi.repository import GLib
from pydbus import SessionBus
from blessed import Terminal

class MetaWriter:
    filename = ""
    output_format = ""
    output_format_artist = ""
    output_format_title = ""
    output_format_album = ""
    last_output = ""

    def __init__(self, filename, output_format, output_format_artist, output_format_title, output_format_album):
        self.filename = filename
        self.output_format = output_format
        self.output_format_album = output_format_album
        self.output_format_artist = output_format_artist
        self.output_format_title = output_format_title

    def write_meta(self, metadata):
        artist = metadata["xesam:artist"][0] if "xesam:artist" in metadata else ""
        title = metadata["xesam:title"] if "xesam:title" in metadata else ""
        album = metadata["xesam:album"] if "xesam:album" in metadata else ""

        self.write(self.output_format.format(
            artist=self.output_format_artist.format(
                artist) if artist != "" else "",
            title=self.output_format_title.format(
                title) if title != "" else "",
            album=self.output_format_album.format(album) if album != "" else ""
        ))

    def write(self, text):
        self.last_output = text
        f = open(self.filename, "w")
        f.write(text)
        f.close()


class PlayerSelector(threading.Thread):
    service_regex = re.compile("^org\.mpris\.MediaPlayer2\.")
    bus = None
    loop = None
    signal_receiver = None
    players = {}
    players_indexes = []
    active_player = ""
    menu = None
    meta_writer = None

    def __init__(self, meta_writer, menu=None):
        super().__init__()

        self.bus = SessionBus()
        self.loop = GLib.MainLoop()
        self.meta_writer = meta_writer
        self.menu = menu

    def set_menu(self, menu):
        self.menu = menu

    def get_players(self):
        self.players = {}
        self.players_indexes = []

        bus_ctrl = self.bus.get(".DBus")

        for service in bus_ctrl.ListNames():
            if self.service_regex.match(service):
                split_service = service.split(".")
                name_start_index = len(
                    split_service[0]) + len(split_service[1]) + len(split_service[2]) + 3

                player = self.bus.get(service, "/org/mpris/MediaPlayer2")
                player_name = ""
                
                try:
                    player_name = player.Identity
                except dbus.exceptions.DBusException:
                    player_name = service[name_start_index:]

                self.players[service] = player_name
                self.players_indexes.append(service)

        self.set_active_player(self.active_player)

    def set_active_player_index(self, player_index):
        if player_index < len(self.players_indexes):
            self.set_active_player(self.players_indexes[player_index])

    def set_active_player(self, player_id):
        if player_id in self.players:
            self.active_player = player_id
        elif len(self.players) != 0:
            self.active_player = next(iter(self.players.keys()))
        else:
            self.active_player = ""

        if self.signal_receiver is not None:
            self.signal_receiver.disconnect()

        if self.active_player != "":
            player = self.bus.get(self.active_player, "/org/mpris/MediaPlayer2")

            self.signal_receiver = self.bus.subscribe(
                sender=self.active_player,
                object="/org/mpris/MediaPlayer2",
                signal="PropertiesChanged",
                signal_fired=self.playing_song_changed
            )

            self.meta_writer.write_meta(player.Metadata)
        else:
            self.meta_writer.write("")

    def run(self):
        bus_ctrl = self.bus.get(".DBus")
        bus_ctrl.NameOwnerChanged.connect(self.dbus_name_owner_changed)

        self.get_players()
        self.menu.refresh()
        self.loop.run()

    def quit(self):
        self.loop.quit()

    def dbus_name_owner_changed(self, name, old_owner, new_owner):
        if self.service_regex.match(name):
            self.get_players()
            self.menu.refresh()

    def playing_song_changed(self, sender, object, iface, signal, params):
        if "Metadata" in params[1]:
            self.meta_writer.write_meta(params[1]["Metadata"])
            self.menu.refresh()


class Menu(threading.Thread):
    refresh_cond = threading.Condition()
    refresh_flag = True
    exit_flag = False
    player_selector = None
    meta_writer = None

    def __init__(self, term, player_selector, meta_writer):
        super().__init__()

        self.player_selector = player_selector
        self.meta_writer = meta_writer

    def refresh(self, exit_flag=False):
        with self.refresh_cond:
            self.refresh_flag = True
            self.exit_flag = self.exit_flag or exit_flag
            self.refresh_cond.notify()

    def run(self):
        with term.fullscreen(), term.hidden_cursor():
            while not self.exit_flag:
                with self.refresh_cond:
                    while not self.refresh_flag and not self.exit_flag:
                        self.refresh_cond.wait()

                    self.refresh_flag = False

                    print(term.clear)
                    print(term.move_xy(0, 0) + term.bold_bright_white_on_red(
                        ("{0:<{width}}").format("MPRIS To Text", width=term.width)) + "\n")
                    print(term.move_x(2) + term.bold("Player: ") + term.move_up())

                    for i, (id, name) in enumerate(self.player_selector.players.items()):
                        output = "%d: %s" % (i, name)

                        if id == self.player_selector.active_player:
                            print(term.move_x(10) + term.standout(output))
                        else:
                            print(term.move_x(10) + output)

                    print(term.move_x(2) + term.bold("File:   ") +
                          self.meta_writer.filename)
                    print(term.move_x(2) + term.bold("Output: ") + "\n".join(term.wrap(
                        self.meta_writer.last_output, width=term.width - 10, subsequent_indent=" " * 10)))

                    print(term.move_x(
                        0) + "\nEnter number to select player or q to exit." + term.move_up())

        print(term.move_xy(0, 0) + "Exiting...")


class Input(threading.Thread):
    player_selector = None
    meta_writer = None
    menu = None

    def __init__(self, term, player_selector, meta_writer, menu):
        super().__init__()

        self.player_selector = player_selector
        self.meta_writer = meta_writer
        self.menu = menu

    def run(self):
        with term.cbreak():
            val = None

            while val not in (u'q', u'Q',):
                val = term.inkey(timeout=5)

                if not val or val.is_sequence or not val.isnumeric():
                    continue

                if int(val) < len(self.player_selector.players):
                    self.player_selector.set_active_player_index(int(val))
                    self.menu.refresh()

            self.menu.refresh(True)
            self.player_selector.quit()


def on_resize(*args):
    global menu_thread

    menu_thread.refresh()


def create_writer():
    parser = argparse.ArgumentParser(
        description="Write metadata from MPRIS-compliant media players into a text file.")
    parser.add_argument("--file",
                        type=str,
                        dest="filename",
                        default="/tmp/mpris_info.txt",
                        help="Full path to file (default: \"/tmp/mpris_info.txt\")"
                        )
    parser.add_argument("--format-artist",
                        type=str,
                        dest="format_artist",
                        default="   by   {}",
                        help="Format string for the artist part (default: \"   by   {}\")"
                        )
    parser.add_argument("--format-title",
                        type=str,
                        dest="format_title",
                        default="\"{}\"",
                        help="Format string for the title part (default: \"\"{}\"\")"
                        )
    parser.add_argument("--format-album",
                        type=str,
                        dest="format_album",
                        default="   from   \"{}\"",
                        help="Format string for the album part (default: \"   from   \"{}\"\")"
                        )
    parser.add_argument("--format",
                        type=str,
                        dest="format",
                        default="{title}{album}{artist}            ",
                        help="Full format string (default: \"{title}{album}{artist}            \")"
                        )

    args = parser.parse_args()

    return MetaWriter(
        filename=args.filename,
        output_format=args.format,
        output_format_artist=args.format_artist,
        output_format_title=args.format_title,
        output_format_album=args.format_album
    )


term = Terminal()
meta_writer = create_writer()

signal.signal(signal.SIGWINCH, on_resize)

player_selector = PlayerSelector(meta_writer)
menu_thread = Menu(term, player_selector, meta_writer)
player_selector.set_menu(menu_thread)
input_thread = Input(term, player_selector, meta_writer, menu_thread)

player_selector.start()
menu_thread.start()
input_thread.start()