API tools faq
paste
Advertisement
Guest User

Untitled

a guest
Feb 23rd, 2019
74
0
Never
Add comment
Not a member of Pastebin yet? Sign Up, it unlocks many cool features!
Python 16.88 KB | None | 0 0
raw download clone embed print report
  1. #!/usr/bin/env python3
  2.  
  3. import sys
  4. import dbus
  5. from operator import itemgetter
  6. import argparse
  7. import re
  8. from urllib.parse import unquote
  9. import time
  10. from dbus.mainloop.glib import DBusGMainLoop
  11. from gi.repository import GLib
  12. DBusGMainLoop(set_as_default=True)
  13.  
  14.  
  15. FORMAT_STRING = '{icon} {artist} - {title}'
  16. FORMAT_REGEX = re.compile(r'(\{:(?P<tag>.*?)(:(?P<format>[wt])(?P<formatlen>\d+))?:(?P<text>.*?):\})', re.I)
  17. FORMAT_TAG_REGEX = re.compile(r'(?P<format>[wt])(?P<formatlen>\d+)')
  18. SAFE_TAG_REGEX = re.compile(r'[{}]')
  19.  
  20. class PlayerManager:
  21.     def __init__(self, blacklist = [], connect = True):
  22.         self.blacklist = blacklist
  23.         self._connect = connect
  24.         self._session_bus = dbus.SessionBus()
  25.         self._last_status = ''
  26.         self.players = {}
  27.         self.refreshPlayerList()
  28.  
  29.         if self._connect:
  30.             self.connect()
  31.             loop = GLib.MainLoop()
  32.             try:
  33.                 loop.run()
  34.             except KeyboardInterrupt:
  35.                 print("interrupt received, stopping…")
  36.    
  37.     def connect(self):
  38.         self._session_bus.add_signal_receiver(self.onOwnerChangedName, 'NameOwnerChanged')
  39.    
  40.     def onOwnerChangedName(self, bus_name, old_owner, new_owner):
  41.         if self.busNameIsAPlayer(bus_name):
  42.             if new_owner and not old_owner:
  43.                 self.addPlayer(bus_name, new_owner)
  44.             elif old_owner and not new_owner:
  45.                 self.removePlayer(old_owner)
  46.             else:
  47.                 self.changePlayerOwner(bus_name, old_owner, new_owner)
  48.  
  49.     def busNameIsAPlayer(self, bus_name):
  50.         return bus_name.startswith('org.mpris.MediaPlayer2') and bus_name.split('.')[-1] not in self.blacklist
  51.  
  52.     def refreshPlayerList(self):
  53.         player_bus_names = [ bus_name for bus_name in self._session_bus.list_names() if self.busNameIsAPlayer(bus_name) ]
  54.         for player_bus_name in player_bus_names:
  55.             self.addPlayer(player_bus_name)
  56.  
  57.     def addPlayer(self, bus_name, owner = None):
  58.         player = Player(self._session_bus, bus_name, owner = owner, connect = self._connect)
  59.         self.players[player.owner] = player
  60.        
  61.     def removePlayer(self, owner):
  62.         self.players[owner].disconnect()
  63.         del self.players[owner]
  64.         if len(self.players) == 0:
  65.             _printFlush(ICON_NONE)
  66.  
  67.     def changePlayerOwner(self, bus_name, old_owner, new_owner):
  68.         player = Player(self._session_bus, bus_name, owner = new_owner, connect = self._connect)
  69.         self.players[new_owner] = player
  70.         del self.players[old_owner]
  71.    
  72.     # Get a list of player owners sorted by current status and age
  73.     def getSortedPlayerOwnerList(self):
  74.         players = [
  75.             {
  76.                 'number': int(owner.split('.')[-1]),
  77.                 'status': 2 if player.status == 'playing' else 1 if player.status == 'paused' else 0,
  78.                 'owner': owner
  79.             }
  80.             for owner, player in self.players.items()
  81.         ]
  82.         return [ info['owner'] for info in reversed(sorted(players, key=itemgetter('status', 'number'))) ]
  83.  
  84.     # Get latest player that's currently playing
  85.     def getCurrentPlayer(self):
  86.         playing_players = [
  87.             player_owner for player_owner in self.getSortedPlayerOwnerList()
  88.             if
  89.                 self.players[player_owner].status == 'playing' or
  90.                 self.players[player_owner].status == 'paused'
  91.         ]
  92.         return self.players[playing_players[0]] if playing_players else None
  93.  
  94.  
  95. class Player:
  96.     def __init__(self, session_bus, bus_name, owner = None, connect = True):
  97.         self._session_bus = session_bus
  98.         self.bus_name = bus_name
  99.         self._disconnecting = False
  100.  
  101.         self.metadata = {
  102.             'artist' : '',
  103.             'album'  : '',
  104.             'title'  : '',
  105.             'track'  : 0
  106.         }
  107.         self._metadata = None
  108.         self.status = 'stopped'
  109.         self.icon = ICON_NONE
  110.         self.icon_reversed = ICON_PLAYING
  111.         if owner is not None:
  112.             self.owner = owner
  113.         else:
  114.             self.owner = self._session_bus.get_name_owner(bus_name)
  115.         self._obj = self._session_bus.get_object(self.bus_name, '/org/mpris/MediaPlayer2')
  116.         self._properties_interface = dbus.Interface(self._obj, dbus_interface='org.freedesktop.DBus.Properties')
  117.         self._introspect_interface = dbus.Interface(self._obj, dbus_interface='org.freedesktop.DBus.Introspectable')
  118.         self._media_interface      = dbus.Interface(self._obj, dbus_interface='org.mpris.MediaPlayer2')
  119.         self._player_interface     = dbus.Interface(self._obj, dbus_interface='org.mpris.MediaPlayer2.Player')
  120.         self._introspect = self._introspect_interface.get_dbus_method('Introspect', dbus_interface=None)
  121.         self._getProperty = self._properties_interface.get_dbus_method('Get', dbus_interface=None)
  122.         self._playerPlay      = self._player_interface.get_dbus_method('Play', dbus_interface=None)
  123.         self._playerPause     = self._player_interface.get_dbus_method('Pause', dbus_interface=None)
  124.         self._playerPlayPause = self._player_interface.get_dbus_method('PlayPause', dbus_interface=None)
  125.         self._playerStop      = self._player_interface.get_dbus_method('Stop', dbus_interface=None)
  126.         self._playerPrevious  = self._player_interface.get_dbus_method('Previous', dbus_interface=None)
  127.         self._playerNext      = self._player_interface.get_dbus_method('Next', dbus_interface=None)
  128.         self._playerRaise     = self._media_interface.get_dbus_method('Raise', dbus_interface=None)
  129.         self._signals = {}
  130.  
  131.         self.refreshStatus()
  132.         self.refreshMetadata()
  133.  
  134.         if connect:
  135.             self.printStatus()
  136.             self.connect()
  137.    
  138.     def play(self):
  139.         self._playerPlay()
  140.     def pause(self):
  141.         self._playerPause()
  142.     def playpause(self):
  143.         self._playerPlayPause()
  144.     def stop(self):
  145.         self._playerStop()
  146.     def previous(self):
  147.         self._playerPrevious()
  148.     def next(self):
  149.         self._playerNext()
  150.     def raisePlayer(self):
  151.         self._playerRaise()
  152.        
  153.     def connect(self):
  154.         if self._disconnecting is not True:
  155.             introspect_xml = self._introspect(self.bus_name, '/')
  156.             if 'TrackMetadataChanged' in introspect_xml:
  157.                 self._signals['track_metadata_changed'] = self._session_bus.add_signal_receiver(self.onMetadataChanged, 'TrackMetadataChanged', self.bus_name)
  158.             self._signals['properties_changed'] = self._properties_interface.connect_to_signal('PropertiesChanged', self.onPropertiesChanged)
  159.    
  160.     def disconnect(self):
  161.         self._disconnecting = True
  162.         for signal_name, signal_handler in list(self._signals.items()):
  163.             signal_handler.remove()
  164.             del self._signals[signal_name]
  165.    
  166.     def refreshStatus(self):
  167.         # Some clients (VLC) will momentarily create a new player before removing it again
  168.         # so we can't be sure the interface still exists
  169.         try:
  170.             self.status = str(self._getProperty('org.mpris.MediaPlayer2.Player', 'PlaybackStatus')).lower()
  171.             self.updateIcon()
  172.         except dbus.exceptions.DBusException:
  173.             self.disconnect()
  174.  
  175.     def refreshMetadata(self):
  176.         # Some clients (VLC) will momentarily create a new player before removing it again
  177.         # so we can't be sure the interface still exists
  178.         try:
  179.             self._metadata = self._getProperty('org.mpris.MediaPlayer2.Player', 'Metadata')
  180.             self._parseMetadata()
  181.         except dbus.exceptions.DBusException:
  182.             self.disconnect()
  183.    
  184.     def updateIcon(self):
  185.         self.icon = (
  186.             ICON_PLAYING if self.status == 'playing' else
  187.             ICON_PAUSED if self.status == 'paused' else
  188.             ICON_STOPPED if self.status == 'stopped' else
  189.             ICON_NONE
  190.         )
  191.         self.icon_reversed = (
  192.             ICON_PAUSED if self.status == 'playing' else
  193.             ICON_PLAYING
  194.         )
  195.  
  196.     def _parseMetadata(self):
  197.         if self._metadata != None:
  198.             artist = _getProperty(self._metadata, 'xesam:artist', [''])
  199.             if len(artist):
  200.                 self.metadata['artist'] = re.sub(SAFE_TAG_REGEX, """\1\1""", artist[0])
  201.             else:
  202.                 self.metadata['artist'] = '';
  203.             self.metadata['album']  = re.sub(SAFE_TAG_REGEX, """\1\1""", _getProperty(self._metadata, 'xesam:album', ''))
  204.             self.metadata['title']  = re.sub(SAFE_TAG_REGEX, """\1\1""", _getProperty(self._metadata, 'xesam:title', ''))
  205.             self.metadata['track']  = _getProperty(self._metadata, 'xesam:trackNumber', '')
  206.             length = str(_getProperty(self._metadata, 'xesam:length', ''))
  207.             if not len(length):
  208.                 length = str(_getProperty(self._metadata, 'mpris:length', ''))
  209.             if len(length):
  210.                 self.metadata['length'] = int(length)
  211.             else:
  212.                 self.metadata['length'] = 0
  213.             self.metadata['genre']  = _getProperty(self._metadata, 'xesam:genre', '')
  214.             self.metadata['disc']   = _getProperty(self._metadata, 'xesam:discNumber', '')
  215.             self.metadata['date']   = re.sub(SAFE_TAG_REGEX, """\1\1""", _getProperty(self._metadata, 'xesam:contentCreated', ''))
  216.             self.metadata['year']   = re.sub(SAFE_TAG_REGEX, """\1\1""", self.metadata['date'][0:4])
  217.             cover = _getProperty(self._metadata, 'xesam:artUrl', '')
  218.             if not len(cover):
  219.                 cover = _getProperty(self._metadata, 'mpris:artUrl', '')
  220.             if len(cover):
  221.                 self.metadata['cover'] = re.sub(SAFE_TAG_REGEX, """\1\1""", cover)
  222.             else:
  223.                 self.metadata['cover'] = ''
  224.  
  225.             self.metadata['duration'] = _getDuration(self.metadata['length'])
  226.    
  227.     def onMetadataChanged(self, track_id, metadata):
  228.         self.refreshMetadata()
  229.         self.printStatus()
  230.    
  231.     def onPropertiesChanged(self, interface, properties, signature):
  232.         updated = False
  233.         if dbus.String('Metadata') in properties:
  234.             _metadata = properties[dbus.String('Metadata')]
  235.             if _metadata != self._metadata:
  236.                 self._metadata = _metadata
  237.                 self._parseMetadata()
  238.                 updated = True
  239.         if dbus.String('PlaybackStatus') in properties:
  240.             status = str(properties[dbus.String('PlaybackStatus')]).lower()
  241.             if status != self.status:
  242.                 self.status = status
  243.                 self.updateIcon()
  244.                 updated = True
  245.        
  246.         if updated:
  247.             self.printStatus()
  248.  
  249.     def _statusReplace(self, match, metadata):
  250.         tag = match.group('tag')
  251.         format = match.group('format')
  252.         formatlen = match.group('formatlen')
  253.         text = match.group('text')
  254.         tag_found = False
  255.         if format is None:
  256.             tag_is_format_match = re.match(FORMAT_TAG_REGEX, tag)
  257.             if tag_is_format_match:
  258.                 format = tag_is_format_match.group('format')
  259.                 formatlen = tag_is_format_match.group('formatlen')
  260.                 tag_found = True
  261.         if format is not None:
  262.             text = text.format_map(CleanSafeDict(**metadata))
  263.             if format == 'w':
  264.                 formatlen = int(formatlen)
  265.                 text = text[:formatlen]
  266.             elif format == 't':
  267.                 formatlen = int(formatlen)
  268.                 if len(text) > formatlen:
  269.                     text = text[:max(formatlen - len(TRUNCATE_STRING), 0)] + TRUNCATE_STRING
  270.         if tag_found is False and tag in metadata and len(metadata[tag]):
  271.             tag_found = True
  272.  
  273.         if tag_found:
  274.             return text
  275.         else:
  276.             return ''
  277.  
  278.     def printStatus(self):
  279.         if self.status in [ 'playing', 'paused' ]:
  280.             if self.metadata['title']:
  281.                 metadata = { **self.metadata, 'icon': self.icon, 'icon-reversed': self.icon_reversed }
  282.                 # replace metadata tags in text
  283.                 text = re.sub(FORMAT_REGEX, lambda match: self._statusReplace(match, metadata), FORMAT_STRING)
  284.                 # restore polybar tag formatting and replace any remaining metadata tags after that
  285.                 try:
  286.                     text = re.sub(r'􏿿p􏿿(.*?)􏿿p􏿿(.*?)􏿿p􏿿(.*?)􏿿p􏿿', r'%{\1}\2%{\3}', text.format_map(CleanSafeDict(**metadata)))
  287.                 except:
  288.                     print("Invalid format string")
  289.                 _printFlush(text)
  290.             return
  291.         _printFlush(ICON_STOPPED)
  292.  
  293.  
  294. def _dbusValueToPython(value):
  295.     if isinstance(value, dbus.Dictionary):
  296.         return {_dbusValueToPython(key): _dbusValueToPython(value) for key, value in value.items()}
  297.     elif isinstance(value, dbus.Array):
  298.         return [ _dbusValueToPython(item) for item in value ]
  299.     elif isinstance(value, dbus.Boolean):
  300.         return int(value) == 1
  301.     elif (
  302.         isinstance(value, dbus.Byte) or
  303.         isinstance(value, dbus.Int16) or
  304.         isinstance(value, dbus.UInt16) or
  305.         isinstance(value, dbus.Int32) or
  306.         isinstance(value, dbus.UInt32) or
  307.         isinstance(value, dbus.Int64) or
  308.         isinstance(value, dbus.UInt64)
  309.     ):
  310.         return int(value)
  311.     elif isinstance(value, dbus.Double):
  312.         return float(value)
  313.     elif (
  314.         isinstance(value, dbus.ObjectPath) or
  315.         isinstance(value, dbus.Signature) or
  316.         isinstance(value, dbus.String)
  317.     ):
  318.         return unquote(str(value))
  319.  
  320. def _getProperty(properties, property, default = None):
  321.     value = default
  322.     if not isinstance(property, dbus.String):
  323.         property = dbus.String(property)
  324.     if property in properties:
  325.         value = properties[property]
  326.         return _dbusValueToPython(value)
  327.     else:
  328.         return value
  329.  
  330. def _getDuration(t: int):
  331.         seconds = t / 1000000
  332.         return time.strftime("%M:%S", time.gmtime(seconds))
  333.  
  334.  
  335. class CleanSafeDict(dict):
  336.     def __missing__(self, key):
  337.         return '{{{}}}'.format(key)
  338.  
  339.  
  340. """
  341. Seems to assure print() actually prints when no terminal is connected
  342. """
  343.  
  344. _last_status = ''
  345. def _printFlush(status, **kwargs):
  346.     global _last_status
  347.     if status != _last_status:
  348.         print(status, **kwargs)
  349.         sys.stdout.flush()
  350.         _last_status = status
  351.  
  352.  
  353.  
  354. parser = argparse.ArgumentParser()
  355. parser.add_argument('command', help="send the given command to the active player",
  356.                     choices=[ 'play', 'pause', 'play-pause', 'stop', 'previous', 'next', 'status', 'list', 'current', 'metadata', 'raise' ],
  357.                     default=None,
  358.                     nargs='?')
  359. parser.add_argument('-b', '--blacklist', help="ignore a player by it's bus name. Can be be given multiple times (e.g. -b vlc -b audacious)",
  360.                     action='append',
  361.                     metavar="BUS_NAME",
  362.                     default=[])
  363. parser.add_argument('-f', '--format', default='{icon} {artist} - {title}')
  364. parser.add_argument('--truncate-text', default='…')
  365. parser.add_argument('--icon-playing', default='⏵')
  366. parser.add_argument('--icon-paused', default='⏸')
  367. parser.add_argument('--icon-stopped', default='⏹')
  368. parser.add_argument('--icon-none', default='')
  369. args = parser.parse_args()
  370.  
  371. FORMAT_STRING = re.sub(r'%\{(.*?)\}(.*?)%\{(.*?)\}', r'􏿿p􏿿\1􏿿p􏿿\2􏿿p􏿿\3􏿿p􏿿', args.format)
  372. TRUNCATE_STRING = args.truncate_text
  373. ICON_PLAYING = args.icon_playing
  374. ICON_PAUSED = args.icon_paused
  375. ICON_STOPPED = args.icon_stopped
  376. ICON_NONE = args.icon_none
  377.  
  378. if args.command is None:
  379.     PlayerManager(blacklist = args.blacklist)
  380. else:
  381.     player_manager = PlayerManager(blacklist = args.blacklist, connect = False)
  382.     current_player = player_manager.getCurrentPlayer()
  383.     if args.command == 'play' and current_player:
  384.         current_player.play()
  385.     elif args.command == 'pause' and current_player:
  386.         current_player.pause()
  387.     elif args.command == 'play-pause' and current_player:
  388.         current_player.playpause()
  389.     elif args.command == 'stop' and current_player:
  390.         current_player.stop()
  391.     elif args.command == 'previous' and current_player:
  392.         current_player.previous()
  393.     elif args.command == 'next' and current_player:
  394.         current_player.next()
  395.     elif args.command == 'status' and current_player:
  396.         current_player.printStatus()
  397.     elif args.command == 'list':
  398.         print("\n".join(sorted([
  399.             "{} : {}".format(player.bus_name.split('.')[-1], player.status)
  400.             for player in player_manager.players.values() ])))
  401.     elif args.command == 'current' and current_player:
  402.         print("{} : {}".format(current_player.bus_name.split('.')[-1], current_player.status))
  403.     elif args.command == 'metadata' and current_player:
  404.         print(_dbusValueToPython(current_player._metadata))
  405.     elif args.command == 'raise' and current_player:
  406.         current_player.raisePlayer()
Advertisement
Add Comment
Please, Sign In to add comment
Advertisement
Public Pastes
Advertisement
We use cookies for various purposes including analytics. By continuing to use Pastebin, you agree to our use of cookies as described in the Cookies Policy.  OK, I Understand
Not a member of Pastebin yet?
Sign Up, it unlocks many cool features!