mmmpd/mmmpd

#!/usr/bin/env python3

import os
import sys
import subprocess
import datetime
import time
import logging
import mpd

NAME = "Mattermost MPD now playing status"
VERSION = "1.0.0"

### Config ########
EMOJI_PAUSED = "pause_button"
EMOJI_PLAYING = "musical_note"
###################

MPD_HOST = os.getenv("MPD_HOST", "localhost") # Host (for TCP) or socket path
MPD_PORT = os.getenv("MPD_PORT", "6600")      # Ignored when not using TCP

LOGGER = logging.getLogger("mmmpd")

MPD_STATE_PLAY = "play"
MPD_STATE_PAUSE = "pause"


# Mattermost stuff {{{1
# ----------------

def set_status(emoji, text, expires_datetime):
	LOGGER.info(f"Custom status expiring {expires_datetime or 'never'}: :{emoji}: {text}")
	subprocess.run([
		"mmcli", "customstatus",
		"--until", expires_datetime.isoformat() if expires_datetime else "",
		"--emoji", emoji,
		"--", text
	], check=True)


def clear_status():
	LOGGER.info("Clearing status")
	subprocess.run(["mmcli", "customstatus"], check=True)


# MPD stuff {{{1
# ---------

def song_string(song_info):
	artist = song_info.get("artist") or "Unknown artist"
	title = song_info.get("title") or "Unknown song"
	return f"{artist} – {title}"


def formatted_status(mpd_client):
	status = mpd_client.status()
	state = status.get("state")
	LOGGER.debug(f"Player state: {state!r}")
	if state not in (MPD_STATE_PLAY, MPD_STATE_PAUSE):
		return None

	song = mpd_client.currentsong()
	song_str = song_string(song)
	emoji = EMOJI_PAUSED if state == MPD_STATE_PAUSE else EMOJI_PLAYING

	expire = None
	elapsed = status.get("elapsed")
	duration = status.get("duration")
	if state == MPD_STATE_PLAY and elapsed is not None and duration:
		now = datetime.datetime.now().astimezone()
		try:
			# 1 second extra to allow some time to set the new song without flickering status
			expire = now + datetime.timedelta(seconds=1 + float(duration) - float(elapsed))
		except ValueError as e:
			LOGGER.error("Could not calculate expiry time", exc_info=e)

	return (emoji, song_str, expire)


# Driving stuff {{{1
# -------------

def set_status_from_mpd(mpd_client):
	status = formatted_status(mpd_client)
	if status is None:
		clear_status()
	else:
		emoji, song_str, expire = status
		set_status(emoji, song_str, expire)


def loop(mpd_client, on_status_change):
	while True:
		mpd_client.idle("player")
		LOGGER.debug("Got event from MPD")
		on_status_change(mpd_client)
		time.sleep(1)


def main(mpd_host, mpd_port):
	if "-v" in sys.argv[1:]:
		logging.basicConfig(level=logging.INFO)
		LOGGER.info("Log level INFO")
	elif "-vv" in sys.argv[1:]:
		logging.basicConfig(level=logging.DEBUG)
		LOGGER.info("Log level DEBUG")

	mpd_client = mpd.MPDClient()
	mpd_client.connect(mpd_host, port=mpd_port)
	LOGGER.info("Connected")

	set_status_from_mpd(mpd_client)
	try:
		loop(mpd_client, set_status_from_mpd)
	finally:
		clear_status()


if __name__ == "__main__":
	main(MPD_HOST, MPD_PORT)