#!/usr/bin/env python
#
# this module uses code from Michael Hudson's xmms-py modules
# this code is available in its original form here:
# http://www.python.net/crew/mwh/hacks/xmms-py.html
# the original code had this notice on it:
#
# Released by Michael Hudson on 2000-07-01, and again on 2001-04-26
# public domain; no warranty, no restrictions
#
# most of the support for xmms, beep, and audacious comes from
# various pieces of Hudson's modules
#
# licensed under the GNU GPL v2
# a copy of this license can be found:
# http://www.gnu.org/copyleft/gpl.html
#
# Future Plan:
# - allow for customizeable announce string (like dcclogger)
# - more mp3 players (maybe some qt based players)
# - more rigorous testing
import xchat
import sys, struct
import socket, os, pwd
from subprocess import *
__DCOP__ = True
try:
import pcop, pydcop
except:
__DCOP__ = False
__module_name__ = "pymp3"
__module_version__ = "0.4.1.1"
__module_description__ = "mp3 announce/utils"
__debugging__ = True
if __debugging__:
import traceback
def print_debug(string):
global __debugging__
if __debugging__: print "\00302" + str(string) + "\003"
def print_info(string):
print "\00303" + str(string) + "\003"
# grep string, find string
def is_running(gs, fs):
p1 = Popen(["ps", "-ef"], stdout=PIPE)
p2 = Popen(["grep", gs], stdin=p1.stdout, stdout=PIPE)
p3 = Popen(["grep", "-v", "grep"], stdin=p2.stdout, stdout=PIPE)
output = p3.communicate()[0]
if output:
output = output.split('\n')
for line in output:
try:
if line.split()[7] == fs:
return True
except: pass
else:
return False
def audacious_is_running(): return is_running("audacious", "audacious")
def beep_is_running(): return is_running("beep-media-player", "beep-media-player")
def xmms_is_running(): return is_running("xmms", "xmms")
def banshee_is_running(): return is_running("banshee", "banshee.exe")
def juk_is_running(): return is_running("juk", "juk")
# find out which player is running
def which():
funcs = {
'audacious' : audacious_is_running,
'beep' : beep_is_running,
'xmms' : xmms_is_running,
'banshee' : banshee_is_running,
'juk' : juk_is_running,
}
for name,func in funcs.items():
if func(): return name
#FIXME: This code isn't that great; it should probably not rely on 'split' since
# quoted won't work properly. Think of a way to fix this (maybe resort to shell=True)
def command(runstr):
return Popen(runstr.split(), stdout=PIPE).communicate()[0]
# these players use xmms style command socket
SOCKET_PLAYERS = ['audacious', 'beep', 'xmms']
CMD_PLAY = 2 #
CMD_PAUSE = 3 #
CMD_STOP = 4 #
CMD_GET_PLAYLIST_POS = 7 #
# TODO: make socket_next and socket_prev use this
# instead of using next/prev repeatedly
#CMD_SET_PLAYLIST_POS = 8 #
CMD_GET_PLAYLIST_LENGTH = 9 #
CMD_GET_OUTPUT_TIME = 11 #
CMD_GET_PLAYLIST_FILE = 17 #
CMD_GET_PLAYLIST_TITLE = 18 #
CMD_GET_PLAYLIST_TIME = 19 #
CMD_GET_INFO = 20 #
CMD_EJECT = 28 #
CMD_PLAYLIST_PREV = 29 #
CMD_PLAYLIST_NEXT = 30 #
CMD_TOGGLE_REPEAT = 33 #
CMD_TOGGLE_SHUFFLE = 34 #
class ClientPacketHeader:
def __init__(self):
self.version,self.cmd,self.length = 0,0,0
def __repr__(self):
return "<< %s : version: %s cmd: %s length: %s >>"\
%(self.__class__.__name__,self.version,self.cmd,self.length)
def encode(self):
return struct.pack("hhl",self.version,self.cmd,self.length)
"""
I've tried to make the following class a facsimily of a "persistent connection",
but my attempts have led to the following error with xmms:
** WARNING **: ctrl_write_packet(): Failed to send data: Broken pipe
Even manually closing, deleting, and then re-initializing the socket did not avoid
this warning. It seems that only letting the garbage collector snag old Connection
objects makes xmms happy.
There is one aspect here missing from Hudson's original library: sending a custom
send format with the 'args' option. I wasn't using this feature in any requests,
as all of my provided formats were 'l' anyway.
"""
class XmmsConnection:
def __init__(self,session=0):
self.sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
self.sock.connect("/tmp/xmms_%s.%d"%(pwd.getpwuid(os.geteuid())[0],session))
def read_header(self):
head = ClientPacketHeader()
head.version, head.cmd, head.length = struct.unpack("hhl",self.sock.recv(8))
return head
def send(self, cmd, args=''):
data = ""
if isinstance(args, int):
data = struct.pack('l', args)
packet = struct.pack("hhl", 1, cmd, len(data)) + data
self.sock.send(packet)
def get_reply(self, format=''):
header = self.read_header()
if format: reply = struct.unpack(format, self.sock.recv(header.length))
else: reply = self.sock.recv(header.length)
return reply
class AudaciousConnection(XmmsConnection):
def __init__(self,session=0):
self.sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
self.sock.connect("/tmp/audacious_%s.%d"%(pwd.getpwuid(os.geteuid())[0],session))
def cached(cache):
"""A method of some object (assumed to be passed as the first
argument, ie. self) has its return value cached in an attribute
named by 'cache'. If the cache exists, return it; else, run the
function and cache the results for the next call."""
def decorator(func):
def inner(*args, **kwargs):
try:
cc = getattr(args[0], cache)
if cc: return cc
except AttributeError: pass
cc = func(*args, **kwargs)
setattr(args[0], cache, cc)
return cc
return inner
return decorator
class MediaPlayer:
"""A superclass that implements some book-keeping and some convenience functions
for media player objects. These objects print out as an announce string, and
get (and cache) info from the player in a clean, consistent "getitem" API."""
def __init__(self, name):
self.name = name
def __getitem__(self, item):
info = self.get_info()
if info.has_key(item):
return info[item]
else:
raise AttributeError('no info key `%s`' % item)
def _nyi(self, name):
print_debug("%s not yet implemented for `%s`." % name, self.name)
def _empty_dict(self):
"""Return an empty info dictionary with all of the keys set."""
keys = ['player', 'playlist_position', 'playlist_length', 'file',
'display_title', 'elapsed', 'length', 'bitrate', 'frequency',
'title', 'artist', 'album', 'track']
d = {}
for key in keys: d[key] = ''
d['player'] = self.name
return d
def human_bitrate(self, bps):
"""Takes bits per second and returns a string with appropriate
larger units, Ex. 1024bps = 1kbps, 2621440bps = 2.5mbps."""
units = ['bps', 'Kbps', 'Mbps']
# order of magnitude
oom = 0
while bps /(1024.0**(oom+1)) >= 1:
oom += 1
return '%0.1f %s' % (bps/1024.0**oom, units[oom])
def s_to_ms(self, s):
"""Converts seconds to minutes:seconds: mm:ss."""
s = int(s)
sec = s % 60
min = s / 60
return '%2d:%02d' % (min, sec)
def us_to_ms(self, us):
"""Converts miliseconds to minutes:seconds: mm:ss."""
us = int(us)
return self.s_to_ms(us/1000)
def play(self): self._nyi('Play')
def stop(self): self._nyi('Stop')
def pause(self): self._nyi('Pause')
def next(self): self._nyi('Next')
def prev(self): self._nyi('Prev')
def eject(self): self._nyi('eject')
def open(self): self._nyi('open')
def shuffle(self): self._nyi('shuffle')
def repeat(self): self._nyi('repeat')
def get_info(self): return self._empty_dict()
def __str__(self):
"""FIXME: This implements the old announce strings. It's probably easier
to move this to the subclasses, but for now this is fine."""
self.get_info()
if self.name in SOCKET_PLAYERS:
return '%s ~ [%s] of [%s] ~ %s ~ %sHz' % (self['display_title'], \
self['elapsed'], self['length'], self['bitrate'], self['frequency'])
elif self.name == 'juk':
return '%s - [%s] - %s ~ [%s] of [%s] ~ %s' % (self['artist'], \
self['album'], self['title'], self['elapsed'], self['length'], \
self['bitrate'])
elif self.name == 'banshee':
return '%s - [%s] - %s ~ [%s] of [%s]' % (self['artist'], self['album'], \
self['title'], self['elapsed'], self['length'])
def __repr__(self):
return '<MediaPlayer %s ...>' % (self.name)
class Xmms(MediaPlayer):
def __init__(self, name='xmms'):
MediaPlayer.__init__(self, name)
self._ifcache = {}
def _makeConnection(self):
if self.name in ['beep', 'xmms']: return XmmsConnection()
elif self.name == 'audacious': return AudaciousConnection()
return False
def _cmd(self, command, args='', reply_format=''):
connection = self._makeConnection()
connection.send(command, args=args)
return connection.get_reply(format=reply_format)
def play(self): self._cmd(CMD_PLAY)
def stop(self): self._cmd(CMD_STOP)
def pause(self): self._cmd(CMD_PAUSE)
def next(self): self._cmd(CMD_PLAYLIST_NEXT)
def prev(self): self._cmd(CMD_PLAYLIST_PREV)
def eject(self): self._cmd(CMD_EJECT)
def open(self): self._cmd(CMD_EJECT)
def shuffle(self): self._cmd(CMD_TOGGLE_SHUFFLE)
def repeat(self): self._cmd(CMD_TOGGLE_REPEAT)
@cached('_ifcache')
def get_info(self):
d = self._empty_dict()
d['playlist_position'] = self._cmd(CMD_GET_PLAYLIST_POS, reply_format='l')[0]
position = d['playlist_position']
d['playlist_length'] = self._cmd(CMD_GET_PLAYLIST_LENGTH, reply_format='l')[0]
d['file'] = self._cmd(CMD_GET_PLAYLIST_FILE, args=position)[:-1]
d['display_title'] = self._cmd(CMD_GET_PLAYLIST_TITLE, args=position)[:-1]
info = self._cmd(CMD_GET_INFO, reply_format='lll')
utime_elapsed = self._cmd(CMD_GET_OUTPUT_TIME, reply_format='l')[0]
utime_length = self._cmd(CMD_GET_PLAYLIST_TIME, args=position, reply_format='l')[0]
d['elapsed'] = self.us_to_ms(utime_elapsed)
d['length'] = self.us_to_ms(utime_length)
d['bitrate'] = self.human_bitrate(info[0])
d['frequency'] = info[1]
return d
BEEP_FIRST_RUN = True
BEEP_MESSAGE = """beep-media-player has a bug with its control socket and returns
bogus information for bitrate, frequency, and number of channels. Consider the
'audacious' media player, or BMPx, as beep-media-player is no longer in
development.""".replace("\n", ' ')
class Beep(Xmms):
def __init__(self):
global BEEP_FIRST_RUN, BEEP_MESSAGE
if BEEP_FIRST_RUN:
print_info(BEEP_MESSAGE)
BEEP_FIRST_RUN = False
Xmms.__init__(self, 'beep')
class Audacious(Xmms):
def __init__(self):
Xmms.__init__(self, 'audacious')
class Banshee(MediaPlayer):
def __init__(self):
MediaPlayer.__init__(self, 'banshee')
self._ifcache = {}
def play(self): command('banshee --play')
def stop(self): command('banshee --pause')
def pause(self): command('banshee --toggle-playing')
def next(self): command('banshee --next')
def prev(self): command('banshee --previous')
def eject(self): command('banshee --show')
def open(self): command('banshee --show')
# shuffle & repeat not yet implemented
@cached('_ifcache')
def get_info(self):
d = self._empty_dict()
info = command(' '.join(['banshee', '--hide-field', '--query-title',
'--query-artist', '--query-position', '--query-album',
'--query-duration'])).strip()
# duration, artist, album, title, position
# banshee reports things in seconds
info = info.split('\n')
d['length'] = self.s_to_ms(info[0])
d['artist'] = info[1]
d['album'] = info[2]
d['title'] = info[3]
d['elapsed'] = self.s_to_ms(info[4])
return d
JUK_FIRST_RUN = True
DCOP_MESSAGE = """Although juk is supported without them, it is recommended that
you install the python-dcop bindings for increased speed.""".replace("\n", ' ')
class Juk(MediaPlayer):
"""MediaPlayer class for Juk, a Qt/KDE media player. This implementation is
a bit messy because it resolves whether or not to use DCOP statically; after
importing, the comparissons are made and the appropriate functions are used."""
def __init__(self):
global JUK_FIRST_RUN, DCOP_MESSAGE, __DCOP__
if JUK_FIRST_RUN and not __DCOP__:
print_info(DCOP_MESSAGE)
JUK_FIRST_RUN = False
MediaPlayer.__init__(self, 'juk')
self._ifcache = {}
self._use_dcop = __DCOP__
self._functions = ['play', 'stop', 'pause', 'next', 'prev', 'eject', 'open']
if self._use_dcop:
self.juk = pydcop.anyAppCalled("juk")
self.get_property = (lambda x: self.juk.Player.trackProperty(x))
for func in self._functions:
setattr(self, func, getattr(self, '_%s_dcop' % func))
else:
self.get_property = (lambda x: command('dcop juk Player trackProperty %s' % (x)).strip())
for func in self._functions:
setattr(self, func, getattr(self, '_%s_nodcop' % func))
def _play_dcop(self): self.juk.Player.play()
def _stop_dcop(self): self.juk.Player.stop()
def _pause_dcop(self): self.juk.Player.playPause()
def _next_dcop(self): self.juk.Player.forward()
def _prev_dcop(self): self.juk.Player.back()
def _eject_dcop(self):
pcop.dcop_call("juk", "juk-mainwindow#1", "restore", ())
pcop.dcop_call("juk", "juk-mainwindow#1", "raise", ())
def _open_dcop(self): self._eject_dcop()
def _play_nodcop(self): command('dcop juk Player play')
def _stop_nodcop(self): command('dcop juk Player stop')
def _pause_nodcop(self): command('dcop juk Player playPause')
def _next_nodcop(self): command('dcop juk Player forward')
def _prev_nodcop(self): command('dcop juk Player back')
def _eject_nodcop(self):
command('dcop juk juk-mainwindow#1 restore')
command('dcop juk juk-mainwindow#1 raise')
def _open_nodcop(self): self._eject_nodcop()
@cached('_ifcache')
def get_info(self):
d = self._empty_dict()
if self._use_dcop: elapsed = self.juk.Player.currentTime()
else: elapsed = command('dcop juk Player currentTime')
d['elapsed'] = self.s_to_ms(elapsed)
d['title'] = self.get_property('Title')
d['artist'] = self.get_property('Artist')
d['album'] = self.get_property('Album')
d['length'] = self.s_to_ms(self.get_property('Seconds'))
d['bitrate'] = '%s Kbps' % self.get_property('Bitrate')
return d
def current_player():
player = which()
print_debug("detected %s is running" % player)
if not player:
raise Exception("Currently not running a supported media player.")
player_obj = eval("%s()" % player.capitalize())
return player_obj
def help(args):
print "\nCommands:"
print " \002/mp3\002 : announce the currently playing mp3"
print " \002/mp3\002 \00303stop\003 : stop playing"
print " \002/mp3\002 \00303play\003 : start playing"
print " \002/mp3\002 \00303pause\003 : pause playback"
print " \002/mp3\002 \00303next [#]\003 : skip to next (# of) track(s)"
print " \002/mp3\002 \00303prev [#]\003 : skip to prev (# of) track(s)"
print " \002/mp3\002 \00303open\003 : open files"
print ""
def usage():
print "Usage: \002/mp3\002 [cmd]\n\002/mp3\002 \037help\037 for commands."
def announce():
player = current_player()
xchat.command('me is listening to: %s' % (player))
def stop(*args):
player = current_player()
player.stop()
def play(*args):
player = current_player()
player.play()
def pause(*args):
player = current_player()
player.pause()
def open(*args):
player = current_player()
player.open()
def eject(*args):
player = current_player()
player.eject()
def next(argv):
num = 1
if len(argv) == 3:
num = int(argv[2])
player = current_player()
for i in range(num):
player.next()
def prev(argv):
num = 1
if len(argv) == 3:
num = int(argv[2])
player = current_player()
for i in range(num):
player.prev()
def dispatch(argv, arg_to_eol, c):
if len(argv) == 1:
announce()
return xchat.EAT_XCHAT
try:
{
"help" : help,
"stop" : stop,
"play" : play,
"pause" : pause,
"next" : next,
"prev" : prev,
"eject" : eject,
"open" : open,
}[argv[1]](argv)
except:
if __debugging__: traceback.print_exc(sys.stdout)
usage()
return xchat.EAT_XCHAT
__unhook__ = xchat.hook_command("mp3", dispatch, help="/mp3 help for commands.")