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

""" 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": int(float(track["time"])) if track.get("time") else None, "elapsed": ( int(float(status["elapsed"])) if status.get("elapsed") else None ), "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()