Initial commit

mpd_mm.py

@@ -0,0 +1,164 @@
+#!/usr/bin/env python3
+
+import sys
+import os
+import requests
+import datetime
+import mpd
+import time
+import logging
+
+NAME = "MPD MM now playing status"
+VERSION = "0.0.1"
+
+### Config ########
+
+# E.g. https://mattermost.zeus.gent/api
+MM_API = os.getenv("MM_API")
+# Get one with mmcli, or by looking in your browser cookie. E.g. gh3cd6ef10abij2kl2nm9op8qr
+MM_API_ACCESS_TOKEN = os.getenv("MM_API_ACCESS_TOKEN")
+
+MPD_HOST = "/var/lib/mpd/socket"
+MPD_PORT = None
+# or to connect with TCP
+#MPD_HOST = "localhost"
+#MPD_PORT = "6600"
+
+EMOJI_PAUSED = "pause_button"
+EMOJI_PLAYING = "musical_note"
+
+# Make the status automatically expire after this many minutes in case this script fails to clear
+# its status when it's done, e.g. due to a loss of internet connection.
+# This expiration is handled by the Mattermost server.
+# Set to 0 or None to disable.
+AUTO_EXPIRE_MINUTES = 60
+
+###################
+
+
+HEADERS = {
+	"Authorization": f"Bearer {MM_API_ACCESS_TOKEN}",
+	"User-Agent": f"{NAME}/{VERSION}"
+}
+
+LOGGER = logging.getLogger("MPD MM now playing")
+
+MPD_STATE_PLAY = "play"
+MPD_STATE_PAUSE = "pause"
+
+# Util {{{1
+# ----
+
+class FastLoopAvoider:
+	def __init__(self, logger, threshold_seconds=1, warn_count=10, stop_count=20, sleep=0):
+		self.threshold_seconds = threshold_seconds
+		self.warn_count = warn_count
+		self.stop_count = stop_count
+		self.sleep = sleep
+		self._last_loop_start = 0
+		self._fast_loops = 0
+		self.logger = logging.getLogger(logger)
+
+	def loop(self):
+		now = time.time()
+		if now - self._last_loop_start < self.threshold_seconds:
+			self._fast_loops += 1
+		self._last_loop_start = now
+
+		if self.stop_count and self._fast_loops == self.stop_count:
+			self.logger.error(f"Stopping because loop reached {self._fast_loops} fast runs")
+			return False
+		if self.warn_count and self._fast_loops == self.warn_count:
+			if self.stop_count:
+				self.logger.warn(f"Loop reached {self._fast_loops} fast runs, stopping at {self.stop_count}")
+			else:
+				self.logger.warn(f"Loop reached {self._fast_loops} fast runs")
+
+		if self.sleep:
+			time.sleep(self.sleep)
+
+		return True
+
+# Mattermost stuff {{{1
+# ----------------
+
+def set_status(emoji, text, expires_datetime):
+	expires = expires_datetime.isoformat() if expires_datetime else None
+	LOGGER.info(f"Setting custom status to :{emoji}: {text} expiring {expires or 'never'}")
+
+	r = requests.put(
+		f"{MM_API}/v4/users/me/status/custom",
+		json={
+			"emoji": emoji,
+			"text": text,
+			"expires_at": expires
+		},
+		headers=HEADERS
+	)
+	r.raise_for_status()
+
+
+def clear_status():
+	LOGGER.info("Clearing custom status")
+	r = requests.delete(f"{MM_API}/v4/users/me/status/custom", headers=HEADERS)
+	r.raise_for_status()
+
+# 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 set_status_from_mpd(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):
+		clear_status()
+		return
+
+	song = mpd_client.currentsong()
+	song_str = song_string(song)
+	emoji = EMOJI_PAUSED if state == MPD_STATE_PAUSE else EMOJI_PLAYING
+
+	if AUTO_EXPIRE_MINUTES:
+		now = datetime.datetime.now().astimezone()
+		expire = now + datetime.timedelta(seconds=AUTO_EXPIRE_MINUTES * 60)
+	else:
+		expire = None
+
+	set_status(emoji, song_str, expire)
+
+
+def loop(mpd_client, on_status_change):
+	fla = FastLoopAvoider("MPD-MM main loop", sleep=1)
+	while fla.loop():
+		mpd_client.idle("player")
+		LOGGER.debug("Got event from MPD")
+		on_status_change(mpd_client)
+
+
+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)