#!/usr/bin/env python # MIT License # # Copyright (c) 2018 Andreas Backx # # Permission is hereby granted, free of charge, to any person obtaining a copy # of this software and associated documentation files (the "Software"), to deal # in the Software without restriction, including without limitation the rights # to use, copy, modify, merge, publish, distribute, sublicense, and/or sell # copies of the Software, and to permit persons to whom the Software is # furnished to do so, subject to the following conditions: # # The above copyright notice and this permission notice shall be included in all # copies or substantial portions of the Software. # # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR # IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, # FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE # AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER # LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE # SOFTWARE. """Spotify DBus listener""" import os import socket import logging from concurrent.futures import ThreadPoolExecutor import click import dbus # type: ignore import dbus.mainloop.glib # type: ignore from dbus.mainloop.glib import DBusGMainLoop from gi.repository import GLib # type: ignore #from spotipy import SpotifyException #from spotipy.oauth2 import SpotifyClientCredentials INACTIVE_COLOR = '%{F#6E6E6E}' ACTIVE_COLOR = '%{F#CECECE}' DEFAULT_COLOR = '%{F-}' SERVER_ADDRESS = '/tmp/playercl-socket' LOG_FILE = '/tmp/playercl.log' class Playercl: """MediaPlayer DBus Listener""" # pylint: disable=too-many-instance-attributes MEDIA_PLAYER_PREFIX = 'org.mpris.MediaPlayer2' SPOTIFY_BUS = 'org.mpris.MediaPlayer2.spotify' SPOTIFYD_BUS = 'org.mpris.MediaPlayer2.spotifyd' MPRIS_OBJECT_PATH = '/org/mpris/MediaPlayer2' PLAYER_INTERFACE = 'org.mpris.MediaPlayer2.Player' PROPERTIES_INTERFACE = 'org.freedesktop.DBus.Properties' SAVE_REMOVE = b'save' logging.basicConfig(filename=LOG_FILE, filemode='w', level=logging.DEBUG) logger = logging.getLogger("playercl") def __init__(self): DBusGMainLoop(set_as_default=True) self.session_bus = dbus.SessionBus() self.last_output = '' self.player_objects = dict() self.current_player = None # Last shown metadata self.last_title = None # Whether the current song is added to the library self.saved_track = False # Whether to ignore the update self.ignore = False # DBus session object self.freedesktop = None def monitor(self): """ Monitor """ self.logger.info("monitor") self.locate_player_services() self.freedesktop = self.session_bus.get_object( "org.freedesktop.DBus", "/org/freedesktop/DBus" ) self.freedesktop.connect_to_signal( "NameOwnerChanged", self.on_name_owner_changed ) executor = ThreadPoolExecutor(max_workers=2) executor.submit(self._start_glib_loop) executor.submit(self._start_server) @staticmethod def _start_glib_loop(): """ Start Glib loop """ loop = GLib.MainLoop() loop.run() @staticmethod def _start_server(): try: os.unlink(SERVER_ADDRESS) except OSError: if os.path.exists(SERVER_ADDRESS): raise sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) sock.bind(SERVER_ADDRESS) sock.listen(5) @property def empty_output(self): """ is output empty""" return not self.last_output @property def metadata_status(self): """ Get song status """ properties = self.get_current_player_properties() metadata = properties.Get( Playercl.PLAYER_INTERFACE, 'Metadata' ) playback_status = properties.Get( Playercl.PLAYER_INTERFACE, 'PlaybackStatus' ) return metadata, playback_status def output(self, line): """ Output for polybar """ if line != self.last_output: print(line, flush=True) self.last_output = line def locate_player_services(self): """ Get all players """ self.logger.info("Find services") for service in self.session_bus.list_names(): if service.startswith(self.MEDIA_PLAYER_PREFIX): self.logger.info("service: %s", service) self.add_player_service(service) def output_playback_status(self, data, retry=False): """ output current song """ if self.ignore: return if 'Metadata' not in data or 'xesam:artist' not in data: self.output('') metadata = data['Metadata'] artists = metadata['xesam:artist'] artist = artists[0] if artists else None artist_string = f'{artist} -' if artist else '' title = metadata['xesam:title'] playback_status = data['PlaybackStatus'] same_song = title == self.last_title color = ACTIVE_COLOR if playback_status == 'Playing' else INACTIVE_COLOR # divider = '+' if same_song and self.saved_track else '-' self.output(f'{color}{artist_string}{title}{DEFAULT_COLOR}') if not same_song: self.last_title = title def on_properties_changed(self, interface, data, *args, **kwargs): """On name properties changed event""" self.logger.info("properties_changed: %s, %s, %s, %s", interface, data, args, kwargs) self.output_playback_status(data) def set_current_player(self, name): """ set the current player by getting what's currently playing""" self.current_player = name def get_current_player_properties(self): """ return the player intefrace for the current player """ property_interface = dbus.Interface( self.player_objects[self.current_player], dbus_interface=self.PROPERTIES_INTERFACE) return property_interface def add_player_service(self, service, path=None): """ Add a service to our known list of players """ self.logger.info("add player service: %s", service) if not path: path = self.MPRIS_OBJECT_PATH try: player = self.session_bus.get_object(service, path) self.player_objects[service] = player player.connect_to_signal( "PropertiesChanged", self.on_properties_changed, sender_keyword='sender', path_keyword='path') self.set_current_player(service) if self.empty_output: metadata, playback_status = self.metadata_status self.output_playback_status( data={ 'Metadata': metadata, 'PlaybackStatus': playback_status, } ) except dbus.DBusException as error: self.logger.warning("Exception: %s", error) def on_name_owner_changed(self, name, old_owner, new_owner): """On name owner changed event""" if name.startswith(self.MEDIA_PLAYER_PREFIX): self.logger.info("name owner changed: name: %s old: %s new: %s", name, old_owner, new_owner or "Removed") if not old_owner: # new object self.add_player_service(name) elif not new_owner: del self.player_objects[name] # if old_owner is None then new object # if new_owner is None then deleted # check name for MPRIS prefix then add to self.player_objects if # not in list @click.group() def cli(): """Script for listening to Spotify over dbus and adding tracks to your library.""" @cli.command() def status(): """Follow the status of the currently playing song on Spotify.""" player = Playercl() player.monitor() if __name__ == '__main__': cli()