fabio / Platypush+Mopidy+Termux configuration

#!~/.config/platypush/config.yaml # Platypush configuration file for Termux Mopidy Notifier # A name for your device device_id: phone # Enable the Web server backend.http: # port: 8008 # Reduce the number of workers to avoid phantom process SIGKILL # See https://ivonblog.com/en-us/posts/fix-termux-signal9-error/ num_workers: 1 ntfy: # Uncomment and set your custom server URL if you're using a self-hosted ntfy server # Otherwise, the default public server (https://ntfy.sh) will be used. # server_url: https://your-ntfy-server.com subscriptions: # Subscribe to the topic where music commands will be sent. # NOTE: It must match the topic configured in your ntfy mobile app - music-commands-<your-unique-id> music.mopidy: host: localhost # port: 6680 # Optional, if you also want to enable the MPD bindings music.mpd: host: localhost # port: 6600 # Set poll_interval to null - music.mopidy is already enabled and it # receives updates over WebSocket. Avoid polling the MPD interface to # prevent battery drain. poll_interval: null

#!/data/data/com.termux/files/usr/bin/sh # A utility script to install the required configuration files # in the Platypush configuration prefix CONFDIR="${CONFDIR:-$HOME/.config/platypush}" mkdir -p "$CONFDIR/scripts" install -m 600 config.yaml "$CONFDIR" install -m 600 mopidy_commands.py "$CONFDIR/scripts" install -m 600 mopidy_notifications.py "$CONFDIR/scripts"

""" This module integrates Mopidy music playback with ntfy notifications. It listens for music control commands sent via ntfy and executes them on the Mopidy server. """ import json import logging from collections import namedtuple from enum import Enum from platypush import Config, hook, run # Listen for ntfy message notifications from platypush.message.event.ntfy import NotificationEvent # NOTE: The topic used to send music commands to Mopidy via ntfy. # It must match the one configured in your ntfy mobile app. music_commands_topic = "music-commands-<your-unique-id>" music_plugin = "music.mopidy" # Or "music.mpd" if using MPD logger = logging.getLogger("music_integrations") ClassCommand = namedtuple("ClassCommand", ["name", "action"]) class MusicCommands(Enum): """ Enum representing music control commands and their corresponding actions. """ PLAY = ClassCommand("play", f"{music_plugin}.play") PAUSE = ClassCommand("pause", f"{music_plugin}.pause") STOP = ClassCommand("stop", f"{music_plugin}.stop") NEXT = ClassCommand("next", f"{music_plugin}.next") PREVIOUS = ClassCommand("previous", f"{music_plugin}.previous") TOGGLE = ClassCommand("toggle", f"{music_plugin}.toggle") supported_commands = {cmd.value.name: cmd.value.action for cmd in MusicCommands} @hook(NotificationEvent, topic=music_commands_topic) def on_music_command(event: NotificationEvent): """ Handles incoming music commands from ntfy notifications. """ try: payload = json.loads(event.message) except Exception as e: logger.warning("Invalid music command payload: %s", e) return # Ignore the command if it's not for this device. # This is useful if multiple devices share the same ntfy topic. device_id = payload.get("device_id") if device_id and device_id != Config.get_device_id(): return command = next( (cmd for cmd in MusicCommands if cmd.value.name == payload.get("command")), None ) if not command: logger.warning( "Unsupported music command: %s. Supported commands: %s", payload.get("command"), list(supported_commands.keys()), ) return # Handle the TOGGLE case depending on current playback state if command == MusicCommands.TOGGLE: status = run("music.mopidy.status") if status.get("state") == "play": action = MusicCommands.PAUSE.value.action else: action = MusicCommands.PLAY.value.action else: action = command.value.action logger.info( "Executing music command on %s@%s: %s", music_plugin, device_id, command.value.name, ) run(action)

""" This module integrates Mopidy music events with mobile notifications using the ntfy service. It listens for music playback events and updates the mobile notification with the current track information, including cover art if available. """ import json import logging import os import pathlib import time import requests from platypush import Config, hook, run # Music events to hook into from platypush.message.event.music import ( MusicPlayEvent, MusicPauseEvent, MusicStopEvent, NewPlayingTrackEvent, ) # NOTE: The topic that will be used to send music notifications. # It must match the one configured in your ntfy mobile app. music_notifications_topic = "music-notifications-<your-unique-id>" logger = logging.getLogger("music_integrations") def get_cover_by_uri(uri): """ A utility method that retrieves the cover image URL for a given Mopidy URI, using directly the Mopidy HTTP JSON-RPC API. It also caches the results locally to avoid repeated requests. """ cache_path = os.path.join( os.path.expanduser("~"), ".cache", "platypush", "mopidy", "covers.json" ) cached_images = {} pathlib.Path(cache_path).parent.mkdir(exist_ok=True, parents=True) if os.path.isfile(cache_path): try: with open(cache_path, "r") as f: cached_images = json.load(f) except Exception as e: logger.warning("Could not load cached cover images: %s", e) # Return cached image URL if available if cached_images.get(uri): return cached_images[uri] resp = None try: # Make a POST request to the Mopidy JSON-RPC API to get the image URL resp = requests.post( "http://localhost:6680/mopidy/rpc", timeout=5.0, json={ "jsonrpc": "2.0", "id": 1, "method": "core.library.get_images", "params": {"uris": [uri]}, }, ) resp.raise_for_status() except Exception as e: logger.warning("Could not retrieve cover for %s: %s", uri, e) return None image_url = cached_images[uri] = ( resp.json().get("result", {}).get(uri, [{}])[0].get("uri") ) # Update the cache file with the new image URL with open(cache_path, "w") as f: f.write(json.dumps(cached_images)) return image_url def update_mobile_music_notification(): """ Updates the mobile music notification with the current track info retrieved from Mopidy, sending it via ntfy. """ # This matches the device_id configured in config.yaml device_id = Config.get_device_id() track = run("music.mopidy.current_track") status = run("music.mopidy.status") state = status.get("state") image = None if not state: return # Retrieve cover image if available if track.get("x-albumuri"): start_t = time.time() image = get_cover_by_uri(track["x-albumuri"]) logger.info("Retrieved track image info in %.2f seconds", time.time() - start_t) # Send the music notification via ntfy run( "ntfy.send_message", topic=music_notifications_topic, message=json.dumps( { "file": track.get("file"), "artist": track.get("artist"), "title": track.get("title"), "album": track.get("album"), "date": track.get("date"), "duration": status.get("track", {}).get("time"), "elapsed": status.get("time"), "image": image, "device_id": device_id, "state": state, "priority": "min", } ), ) @hook(MusicPlayEvent) def on_music_play(): update_mobile_music_notification() @hook(MusicPauseEvent) def on_music_pause(): update_mobile_music_notification() @hook(MusicStopEvent) def on_music_stop(): update_mobile_music_notification() @hook(NewPlayingTrackEvent) def on_new_track(): update_mobile_music_notification()