Source code for platypush.plugins.music.mpd

import re
import threading
import time
from typing import Optional, Union

from platypush.plugins import action
from platypush.plugins.music import MusicPlugin


[docs]class MusicMpdPlugin(MusicPlugin): """ This plugin allows you to interact with an MPD/Mopidy music server. MPD (https://www.musicpd.org/) is a flexible server-side protocol/application for handling music collections and playing music, mostly aimed to manage local libraries. Mopidy (https://www.mopidy.com/) is an evolution of MPD, compatible with the original protocol and with support for multiple music sources through plugins (e.g. Spotify, TuneIn, Soundcloud, local files etc.). **NOTE**: As of Mopidy 3.0 MPD is an optional interface provided by the ``mopidy-mpd`` extension. Make sure that you have the extension installed and enabled on your instance to use this plugin with your server. Requires: * **python-mpd2** (``pip install python-mpd2``) """ _client_lock = threading.RLock()
[docs] def __init__(self, host, port=6600): """ :param host: MPD IP/hostname :type host: str :param port: MPD port (default: 6600) :type port: int """ super().__init__() self.host = host self.port = port self.client = None
def _connect(self, n_tries=2): import mpd with self._client_lock: if self.client: return error = None while n_tries > 0: try: n_tries -= 1 self.client = mpd.MPDClient() self.client.connect(self.host, self.port) return self.client except Exception as e: error = e self.logger.warning('Connection exception: {}{}'. format(str(e), (': Retrying' if n_tries > 0 else ''))) time.sleep(0.5) self.client = None if error: raise error def _exec(self, method, *args, **kwargs): error = None n_tries = int(kwargs.pop('n_tries')) if 'n_tries' in kwargs else 2 return_status = kwargs.pop('return_status') \ if 'return_status' in kwargs else True while n_tries > 0: try: self._connect() n_tries -= 1 with self._client_lock: response = getattr(self.client, method)(*args, **kwargs) if return_status: return self.status().output return response except Exception as e: error = str(e) self.logger.warning('Exception while executing MPD method {}: {}'. format(method, error)) self.client = None return None, error
[docs] @action def play(self, resource=None): """ Play a resource by path/URI :param resource: Resource path/URI :type resource: str """ if resource: self.add(resource, position=0) return self.play_pos(0) return self._exec('play')
[docs] @action def play_pos(self, pos): """ Play a track in the current playlist by position number :param pos: Position number """ return self._exec('play', pos)
[docs] @action def pause(self): """ Pause playback """ status = self.status().output['state'] if status == 'play': return self._exec('pause') else: return self._exec('play')
[docs] @action def pause_if_playing(self): """ Pause playback only if it's playing """ status = self.status().output['state'] if status == 'play': return self._exec('pause')
[docs] @action def play_if_paused(self): """ Play only if it's paused (resume) """ status = self.status().output['state'] if status == 'pause': return self._exec('play')
[docs] @action def play_if_paused_or_stopped(self): """ Play only if it's paused or stopped """ status = self.status().output['state'] if status == 'pause' or status == 'stop': return self._exec('play')
[docs] @action def stop(self): """ Stop playback """ return self._exec('stop')
[docs] @action def play_or_stop(self): """ Play or stop (play state toggle) """ status = self.status().output['state'] if status == 'play': return self._exec('stop') else: return self._exec('play')
[docs] @action def playid(self, track_id): """ Play a track by ID :param track_id: Track ID :type track_id: str """ return self._exec('playid', track_id)
[docs] @action def next(self): """ Play the next track """ return self._exec('next')
[docs] @action def previous(self): """ Play the previous track """ return self._exec('previous')
[docs] @action def setvol(self, vol): """ Set the volume (DEPRECATED, use :meth:`.set_volume` instead). :param vol: Volume value (range: 0-100) :type vol: int """ return self.set_volume(vol)
[docs] @action def set_volume(self, volume): """ Set the volume. :param volume: Volume value (range: 0-100) :type volume: int """ return self._exec('setvol', str(volume))
[docs] @action def volup(self, delta=10): """ Turn up the volume :param delta: Volume up delta (default: +10%) :type delta: int """ volume = int(self.status().output['volume']) new_volume = min(volume + delta, 100) return self.setvol(new_volume)
[docs] @action def voldown(self, delta=10): """ Turn down the volume :param delta: Volume down delta (default: -10%) :type delta: int """ volume = int(self.status().output['volume']) new_volume = max(volume - delta, 0) return self.setvol(new_volume)
[docs] @action def random(self, value=None): """ Set random mode :param value: If set, set the random state this value (true/false). Default: None (toggle current state) :type value: bool """ if value is None: value = int(self.status().output['random']) value = 1 if value == 0 else 0 return self._exec('random', value)