#!/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()