from collections import defaultdict
import socket
from os.path import basename
from math import floor
from i3pystatus import IntervalModule, formatp
from i3pystatus.core.util import TimeWrapper
[docs]class MPD(IntervalModule):
"""
Displays various information from MPD (the music player daemon)
.. rubric:: Available formatters (uses :ref:`formatp`)
* `{title}` — (the title of the current song)
* `{album}` — (the album of the current song, can be an empty string \
(e.g. for online streams))
* `{artist}` — (can be empty, too)
* `{album_artist}` — (can be empty)
* `{filename}` — (file name with out extension and path; empty unless \
title is empty)
* `{song_elapsed}` — (Position in the currently playing song, uses \
:ref:`TimeWrapper`, default is `%m:%S`)
* `{song_length}` — (Length of the current song, same as song_elapsed)
* `{pos}` — (Position of current song in playlist, one-based)
* `{len}` — (Songs in playlist)
* `{status}` — (play, pause, stop mapped through the `status` dictionary)
* `{bitrate}` — (Current bitrate in kilobit/s)
* `{volume}` — (Volume set in MPD)
.. rubric:: Available callbacks
* ``switch_playpause`` — Plays if paused or stopped, otherwise pauses. \
Emulates ``mpc toggle``.
* ``stop`` — Stops playback. Emulates ``mpc stop``.
* ``next_song`` — Goes to next track in the playlist. Emulates ``mpc \
next``.
* ``previous_song`` — Goes to previous track in the playlist. Emulates \
``mpc prev``.
* ``mpd_command`` — Send a command directly to MPD's socket. The command \
is the second element of the list. Documentation for available commands can \
be found at https://www.musicpd.org/doc/protocol/command_reference.html
Example module registration with callbacks:
::
status.register("mpd",
on_leftclick="switch_playpause",
on_rightclick=["mpd_command", "stop"],
on_middleclick=["mpd_command", "shuffle"],
on_upscroll=["mpd_command", "seekcur -10"],
on_downscroll=["mpd_command", "seekcur +10"])
Note that ``next_song`` and ``previous_song``, and their ``mpd_command`` \
equivalents, are ignored while mpd is stopped.
"""
interval = 1
settings = (
("host"),
("port", "MPD port. If set to 0, host will we interpreted as a Unix \
socket."),
("format", "formatp string"),
("status", "Dictionary mapping pause, play and stop to output"),
("color", "The color of the text"),
("color_map", "The mapping from state to color of the text"),
("max_field_len", "Defines max length for in truncate_fields defined \
fields, if truncated, ellipsis are appended as indicator. It's applied \
*before* max_len. Value of 0 disables this."),
("max_len", "Defines max length for the hole string, if exceeding \
fields specefied in truncate_fields are truncated equaly. If truncated, \
ellipsis are appended as indicator. It's applied *after* max_field_len. Value \
of 0 disables this."),
("time_format", "format string for 'pos' and 'len' fields"),
("truncate_fields", "fields that will be truncated if exceeding \
max_field_len or max_len."),
("hide_inactive", "Hides status information when MPD is not running"),
("password", "A password for access to MPD. (This is sent in \
cleartext to the server.)"),
)
host = "localhost"
port = 6600
password = None
s = None
format = "{title} {status}"
status = {
"pause": "▷",
"play": "▶",
"stop": "◾",
}
color = "#FFFFFF"
color_map = {}
max_field_len = 25
max_len = 100
time_format = "%m:%S"
truncate_fields = ("title", "album", "artist", "album_artist")
hide_inactive = False
on_leftclick = "switch_playpause"
on_rightclick = "next_song"
on_upscroll = on_rightclick
on_downscroll = "previous_song"
def _mpd_command(self, sock, command):
try:
sock.send((command + "\n").encode("utf-8"))
except Exception as e:
if self.port != 0:
self.s = socket.create_connection((self.host, self.port))
else:
self.s = socket.socket(family=socket.AF_UNIX)
self.s.connect(self.host)
sock = self.s
sock.recv(8192)
if self.password is not None:
sock.send('password "{}"\n'.format(self.password).
encode("utf-8"))
sock.recv(8192)
sock.send((command + "\n").encode("utf-8"))
try:
reply = sock.recv(16384).decode("utf-8", "replace")
replylines = reply.split("\n")[:-2]
return dict(
(line.split(": ", 1)) for line in replylines
)
except Exception as e:
return None
def run(self):
try:
status = self._mpd_command(self.s, "status")
playback_state = status["state"]
if playback_state == "stop":
currentsong = {}
else:
currentsong = self._mpd_command(self.s, "currentsong") or {}
except Exception:
if self.hide_inactive:
self.output = {
"full_text": ""
}
if hasattr(self, "data"):
del self.data
return
fdict = {
"pos": int(status.get("song", 0)) + 1,
"len": int(status.get("playlistlength", 0)),
"status": self.status[playback_state],
"volume": int(status.get("volume", 0)),
"title": currentsong.get("Title", ""),
"album": currentsong.get("Album", ""),
"artist": currentsong.get("Artist", ""),
"album_artist": currentsong.get("AlbumArtist", ""),
"song_length": TimeWrapper(currentsong.get("Time", 0), default_format=self.time_format),
"song_elapsed": TimeWrapper(float(status.get("elapsed", 0)), default_format=self.time_format),
"bitrate": int(status.get("bitrate", 0)),
}
if not fdict["title"] and "file" in currentsong:
fdict["filename"] = '.'.join(
basename(currentsong["file"]).split('.')[:-1])
else:
fdict["filename"] = ""
if self.max_field_len > 0:
for key in self.truncate_fields:
if len(fdict[key]) > self.max_field_len:
fdict[key] = fdict[key][:self.max_field_len - 1] + "…"
self.data = fdict
full_text = formatp(self.format, **fdict).strip()
full_text_len = len(full_text)
if full_text_len > self.max_len and self.max_len > 0:
shrink = floor((self.max_len - full_text_len)
/ len(self.truncate_fields)) - 1
for key in self.truncate_fields:
fdict[key] = fdict[key][:shrink] + "…"
full_text = formatp(self.format, **fdict).strip()
color_map = defaultdict(lambda: self.color, self.color_map)
self.output = {
"full_text": full_text,
"color": color_map[playback_state],
}
def switch_playpause(self):
try:
self._mpd_command(self.s, "play"
if self._mpd_command(self.s, "status")["state"]
in ["pause", "stop"] else "pause 1")
except Exception as e:
pass
def stop(self):
try:
self._mpd_command(self.s, "stop")
except Exception as e:
pass
def next_song(self):
try:
self._mpd_command(self.s, "next")
except Exception as e:
pass
def previous_song(self):
try:
self._mpd_command(self.s, "previous")
except Exception as e:
pass
def mpd_command(self, command):
try:
self._mpd_command(self.s, command)
except Exception as e:
pass