Source code for platypush.plugins.music.mopidy

from typing import Dict, Iterable, List, Optional, Union

from platypush.plugins import RunnablePlugin, action
from platypush.plugins.media import PlayerState
from platypush.schemas.mopidy import (
    MopidyAlbumSchema,
    MopidyArtistSchema,
    MopidyDirectorySchema,
    MopidyFilterSchema,
    MopidyPlaylistSchema,
    MopidyStatusSchema,
    MopidyTrackSchema,
)
from platypush.utils import wait_for_either

from ._client import MopidyClient
from ._common import DEFAULT_TIMEOUT
from ._conf import MopidyConfig
from ._exc import EmptyTrackException
from ._playlist import MopidyPlaylist
from ._status import MopidyStatus
from ._sync import PlaylistSync
from ._task import MopidyTask
from ._track import MopidyTrack


[docs] class MusicMopidyPlugin(RunnablePlugin): """ This plugin allows you to track the events from a Mopidy instance and control it through the Mopidy HTTP API. Requires: * A Mopidy instance running with the HTTP service enabled. """
[docs] def __init__( self, host: str = 'localhost', port: int = 6680, ssl: bool = False, timeout: Optional[float] = DEFAULT_TIMEOUT, **kwargs, ): """ :param host: Mopidy host (default: localhost). :param port: Mopidy HTTP port (default: 6680). :param ssl: Set to True if the Mopidy server is running on HTTPS. :param timeout: Default timeout for the Mopidy requests (default: 20s). """ super().__init__(**kwargs) self.config = MopidyConfig(host=host, port=port, ssl=ssl, timeout=timeout) self._status = MopidyStatus() self._tasks: Dict[int, MopidyTask] = {} self._client: Optional[MopidyClient] = None self._playlist_sync = PlaylistSync()
def _exec(self, *msgs: dict, **kwargs): assert self._client, "Mopidy client not running" return self._client.exec( *msgs, timeout=kwargs.pop('timeout', self.config.timeout) ) def _exec_with_status(self, *msgs: dict, **kwargs): self._exec(*msgs, **kwargs) return self._dump_status() def _dump_status(self): assert self._client, "Mopidy client not running" return MopidyStatusSchema().dump(self._client.status) def _dump_results(self, results: List[dict]) -> List[dict]: schema_by_type = { 'artist': MopidyArtistSchema(), 'album': MopidyAlbumSchema(), 'directory': MopidyDirectorySchema(), 'playlist': MopidyPlaylistSchema(), 'track': MopidyTrackSchema(), } return [ { **( MopidyTrack.parse(item).to_dict() # type: ignore if item['type'] == 'track' else schema_by_type[item['type']].dump(item) ), 'type': item['type'], } for item in results ] def _dump_search_results(self, results: List[dict]) -> List[dict]: return self._dump_results( [ { **item, 'type': res_type, } for search_provider in results for res_type in ['artist', 'album', 'track'] for item in search_provider.get(res_type + 's', []) ] ) def _lookup(self, *uris: str) -> Dict[str, List[MopidyTrack]]: if not uris: return {} if len(uris) > 1: # If more than one URI is specified, we need to call only # library.lookup, as playlist.lookup does not support multiple # URIs. result = self._exec( {'method': 'core.library.lookup', 'uris': uris}, )[0] else: # Otherwise, search both in the library and in the playlist # controllers. uri = uris[0] result = self._exec( {'method': 'core.playlists.lookup', 'uri': uri}, {'method': 'core.library.lookup', 'uris': [uri]}, ) result = { uri: ( result[0].get('tracks', []) if result[0] else list(result[1].values())[0] ) } ret = {} for uri, tracks in result.items(): ret[uri] = [] for track in tracks: parsed_track = MopidyTrack.parse(track) if parsed_track: ret[uri].append(parsed_track) return ret def _add( self, *resources: str, position: Optional[int] = None, clear: bool = False, lookup: bool = True, ): batch_size = 50 results = self._lookup(*resources).values() ret = [] uris = ( [track.uri for tracks in results for track in tracks if track and track.uri] if lookup else list(resources) ) with self._playlist_sync: if clear: self.clear() for i in range(0, len(uris), batch_size): ret += self._exec( { 'method': 'core.tracklist.add', 'uris': uris[i : i + batch_size], 'at_position': position, } )[0] self.logger.info('Loaded %d/%d tracks', len(ret), len(uris)) return ret def _get_playlist(self, playlist: str, with_tracks: bool = False) -> MopidyPlaylist: playlists = self._get_playlists() pl_by_name = {p.name: p for p in playlists} pl_by_uri = {p.uri: p for p in playlists} pl = pl_by_uri.get(playlist, pl_by_name.get(playlist)) assert pl, f"Playlist {playlist} not found" if with_tracks: pl.tracks = self._get_playlist_tracks(playlist) return pl def _get_playlist_tracks(self, playlist: str) -> List[MopidyTrack]: playlists = self._get_playlists() pl_by_name = {p.name: p for p in playlists} pl_by_uri = {p.uri: p for p in playlists} pl = pl_by_uri.get(playlist, pl_by_name.get(playlist)) assert pl, f"Playlist {playlist} not found" tracks = self._exec({'method': 'core.playlists.get_items', 'uri': pl.uri})[0] assert tracks is not None, f"Playlist {playlist} not found" ret = [] for track in tracks: parsed_track = MopidyTrack.parse(track) if parsed_track: ret.append(parsed_track) return ret def _get_playlists(self, **__) -> List[MopidyPlaylist]: return [ MopidyPlaylist.parse(pl) for pl in self._exec({'method': 'core.playlists.as_list'})[0] ] def _save_playlist(self, playlist: MopidyPlaylist): return self._exec( { 'method': 'core.playlists.save', 'playlist': { '__model__': 'Playlist', 'uri': playlist.uri, 'name': playlist.name, 'tracks': [ { '__model__': 'Track', 'uri': track.uri, } for track in playlist.tracks ], }, } )[0]
[docs] @action def play( self, resource: Optional[str] = None, position: Optional[int] = None, track_id: Optional[int] = None, **__, ): """ Start playback, or play a resource by URI. :param resource: Resource path/URI. If not specified, it will resume the playback if paused/stopped, otherwise it will start playing the selected track. :param track_id: The ID of track (or ``tlid``) in the current playlist that should be played, if you want to play a specific track already loaded in the current playlist. :param position: Position number (0-based) of the track in the current playlist that should be played. :return: .. schema:: mopidy.MopidyStatusSchema """ if resource: ret = self._add(resource, position=0) if not ret: self.logger.warning('Failed to add %s to the tracklist', resource) elif isinstance(ret, list): track_id = ret[0].get('tlid') elif isinstance(ret, dict): track_id = ret.get('tlid') elif position is not None: tracklist = self._exec({'method': 'core.tracklist.get_tl_tracks'})[0] if position < 0 or position >= len(tracklist): self.logger.warning( 'Position %d is out of bounds for the current tracklist', position ) return None track_id = tracklist[position]['tlid'] return self._exec_with_status( {'method': 'core.playback.play', 'tlid': track_id} )
[docs] @action def play_pos(self, pos: int): """ Play a track in the current playlist by position number. Legacy alias for :meth:`.play` with a ``position`` parameter. :param pos: Position number (0-based). """ return self.play(position=pos)
[docs] @action def load(self, playlist: str, play: bool = True): """ Load/play a playlist. This method will clear the current playlist and load the tracks from the given playlist. You should usually prefer :meth:`.add` to this method, as it is more general-purpose (``load`` only works with playlists). This method exists mainly for compatibility with the MPD plugin. :param playlist: Playlist URI. :param play: Start playback after loading the playlist (default: True). """ self._add(playlist, clear=True) if play: self.play()
[docs] @action def lookup(self, resources: Iterable[str], **__): """ Lookup (one or) resources by URI. Given a list of URIs, this method will return a dictionary in the form ``{uri: [track1, track2, ...]}``. :param resource: Resource URI(s). :return: .. schema:: mopidy.MopidyTrackSchema(many=True) """ return { uri: [track.to_dict() for track in tracks] for uri, tracks in self._lookup(*resources).items() }
[docs] @action def pause(self, **__): """ Pause the playback. :return: .. schema:: mopidy.MopidyStatusSchema """ return self._exec_with_status({'method': 'core.playback.pause'})
[docs] @action def stop(self, **__): # type: ignore """ Stop the playback. :return: .. schema:: mopidy.MopidyStatusSchema """ # Note: stop could be interpreted both as "stop the playback" and "stop # the plugin". If the stop event is set, we assume that the user wants # to stop the plugin. if self.should_stop(): if self._client: self._client.stop() return None return self._exec_with_status({'method': 'core.playback.stop'})
[docs] @action def prev(self, **__): """ Play the previous track. :return: .. schema:: mopidy.MopidyStatusSchema """ return self._exec_with_status({'method': 'core.playback.previous'})
[docs] @action def next(self, **__): """ Play the next track. :return: .. schema:: mopidy.MopidyStatusSchema """ return self._exec_with_status({'method': 'core.playback.next'})
[docs] @action def add( self, resource: Union[str, Iterable[str]], *_, position: Optional[int] = None, **__, ): """ Add a resource (track, album, artist, folder etc.) to the current playlist. :param resource: Resource URI(s). :param position: Position (0-based) where the track(s) will be inserted (default: end of the playlist). :return: The list of tracks added to the queue. .. schema:: mopidy.MopidyTrackSchema(many=True) """ resources = [resource] if isinstance(resource, str) else resource tracks = [ MopidyTrack.parse(track) for track in self._add(*resources, position=position) ] return [track.to_dict() for track in tracks if track]
[docs] @action def pause_if_playing(self, **__): """ Pause the playback if it's currently playing. :return: .. schema:: mopidy.MopidyStatusSchema """ if self._status.state == PlayerState.PLAY: return self.pause() return self._dump_status()
[docs] @action def play_if_paused(self, **__): """ Resume the playback if it's currently paused. :return: .. schema:: mopidy.MopidyStatusSchema """ if self._status.state == PlayerState.PAUSE: return self.play() return self._dump_status()
[docs] @action def play_if_paused_or_stopped(self): """ Resume the playback if it's currently paused or stopped. :return: .. schema:: mopidy.MopidyStatusSchema """ if self._status.state in {PlayerState.PAUSE, PlayerState.STOP}: return self.play() return self._dump_status()
[docs] @action def play_or_stop(self): """ Play if the playback is stopped, stop if it's playing, otherwise resume playback. :return: .. schema:: mopidy.MopidyStatusSchema """ if self._status.state == PlayerState.PLAY: return self.stop() return self.play()
[docs] @action def set_volume(self, volume: int, **__): """ Set the volume. :param volume: Volume level (0-100). :return: .. schema:: mopidy.MopidyStatusSchema """ return self._exec_with_status( {'method': 'core.mixer.set_volume', 'volume': volume} )
[docs] @action def volup(self, step: int = 5, **__): """ Increase the volume by a given step. :param step: Volume step (default: 5%). :return: .. schema:: mopidy.MopidyStatusSchema """ return self.set_volume(volume=min(100, self._status.volume + step))
[docs] @action def voldown(self, step: int = 5, **__): """ Decrease the volume by a given step. :param step: Volume step (default: 5%). :return: .. schema:: mopidy.MopidyStatusSchema """ return self.set_volume(volume=max(0, self._status.volume - step))
[docs] @action def random(self, value: Optional[bool] = None, **__): """ Set the random mode. :param value: Random mode. If not specified, it will toggle the current random mode. :return: .. schema:: mopidy.MopidyStatusSchema """ if value is None: value = not self._status.random return self._exec_with_status( {'method': 'core.tracklist.set_random', 'value': bool(value)} )
[docs] @action def repeat(self, value: Optional[bool] = None, **__): """ Set the repeat mode. :param value: Repeat mode. If not specified, it will toggle the current repeat mode. :return: .. schema:: mopidy.MopidyStatusSchema """ if value is None: value = not self._status.repeat return self._exec_with_status( {'method': 'core.tracklist.set_repeat', 'value': bool(value)} )
[docs] @action def consume(self, value: Optional[bool] = None, **__): """ Set the consume mode. :param value: Consume mode. If not specified, it will toggle the current consume mode. :return: .. schema:: mopidy.MopidyStatusSchema """ if value is None: value = not self._status.consume return self._exec_with_status( {'method': 'core.tracklist.set_consume', 'value': bool(value)} )
[docs] @action def single(self, value: Optional[bool] = None, **__): """ Set the single mode. :param value: Single mode. If not specified, it will toggle the current single mode. :return: .. schema:: mopidy.MopidyStatusSchema """ if value is None: value = not self._status.single return self._exec_with_status( {'method': 'core.tracklist.set_single', 'value': bool(value)} )
[docs] @action def shuffle(self, **__): """ Shuffle the current playlist. :return: .. schema:: mopidy.MopidyStatusSchema """ return self._exec_with_status({'method': 'core.tracklist.shuffle'})
[docs] @action def save(self, name: str, **__): """ Save the current tracklist to a new playlist with the given name. :param name: New playlist name. """ return self._exec({'method': 'core.playlists.save', 'name': name})[0]
[docs] @action def delete( self, positions: Optional[Iterable[int]] = None, uris: Optional[Iterable[str]] = None, ): """ Delete tracks from the current tracklist. .. note:: At least one of the ``positions`` or ``uris`` parameters must be specified. :param positions: (0-based) positions of the tracks to be deleted. :param uris: URIs of the tracks to be deleted. """ assert ( positions or uris ), "At least one of 'positions' or 'uris' must be specified" criteria = {} if positions: assert self._client, "Mopidy client not running" positions = set(positions) criteria['tlid'] = list( { track.track_id for i, track in enumerate(self._client.tracks) if i in positions } ) if uris: criteria['uri'] = list(uris) return self._exec( { 'method': 'core.tracklist.remove', 'criteria': criteria, } )[0]
[docs] @action def move( self, start: Optional[int] = None, end: Optional[int] = None, position: Optional[int] = None, from_pos: Optional[int] = None, to_pos: Optional[int] = None, **__, ): """ Move one or more tracks in the current playlist to a new position. You can pass either: - ``start``, ``end`` and ``position`` to move a slice of tracks from ``start`` to ``end`` to the new position ``position``. - ``from_pos`` and ``to_pos`` to move a single track from ``from_pos`` to ``to_pos``. .. note: Positions are 0-based (i.e. the first track has position 0). :param start: Start position of the slice of tracks to be moved. :param end: End position of the slice of tracks to be moved. :param position: New position where the tracks will be inserted. :param from_pos: Alias for ``start`` - it only works with one track at the time. Maintained for compatibility with :meth:`platypush.plugins.music.mpd.MusicMpdPlugin.move`. :param to_pos: Alias for ``position`` - it only works with one track at the time. Maintained for compatibility with :meth:`platypush.plugins.music.mpd.MusicMpdPlugin.move`. """ assert (from_pos is not None and to_pos is not None) or ( start is not None and end is not None and position is not None ), 'Either "start", "end" and "position", or "from_pos" and "to_pos" must be specified' if (from_pos is not None) and (to_pos is not None): start, end, position = from_pos, from_pos, to_pos ret = self._exec( { 'method': 'core.tracklist.move', 'start': start, 'end': end, 'to_position': position, } )[0] if self._client: self._client.refresh_status(with_tracks=True) return ret
[docs] @action def clear(self, **__): """ Clear the current playlist. """ self._exec_with_status({'method': 'core.tracklist.clear'})
[docs] @action def seek(self, position: float, **__): """ Seek to a given position in the current track. :param position: Position in seconds. :return: .. schema:: mopidy.MopidyStatusSchema """ return self._exec_with_status( {'method': 'core.playback.seek', 'time_position': int(position * 1000)} )
[docs] @action def back(self, delta: float = 10, **__): """ Seek back by a given number of seconds. :param delta: Number of seconds to seek back (default: 10s). :return: .. schema:: mopidy.MopidyStatusSchema """ if self._status.playing_pos is None: return self._dump_status() return self.seek(position=self._status.playing_pos - delta)
[docs] @action def forward(self, delta: float = 10, **__): """ Seek forward by a given number of seconds. :param delta: Number of seconds to seek forward (default: 10s). :return: .. schema:: mopidy.MopidyStatusSchema """ if self._status.playing_pos is None: return self._dump_status() return self.seek(position=self._status.playing_pos + delta)
[docs] @action def status(self, **__): """ Get the current Mopidy status. :return: .. schema:: mopidy.MopidyStatusSchema """ assert self._client, "Mopidy client not running" self._client.refresh_status() return self._dump_status()
[docs] @action def current_track(self, **__): """ Get the current track. :return: .. schema:: mopidy.MopidyTrackSchema """ assert self._client, "Mopidy client not running" if not self._client.status.track: return None return self._client.status.track.to_dict()
[docs] @action def get_tracks(self, **__): """ Get the current playlist tracks. :return: .. schema:: mopidy.MopidyTrackSchema(many=True) """ assert self._client, "Mopidy client not running" return [t.to_dict() for t in self._client.tracks]
[docs] @action def get_playlists(self, **__): """ Get the available playlists. :return: .. schema:: mopidy.MopidyPlaylistSchema(many=True) """ return MopidyPlaylistSchema().dump(self._get_playlists(), many=True)
[docs] @action def get_playlist(self, playlist: str, **__): """ Get the items in a playlist. :param playlist: Playlist URI. :param only_tracks: If True, only the tracks will be returned, otherwise the full playlist object will be returned - including name and other metadata. :return: .. schema:: mopidy.MopidyTrackSchema(many=True) """ tracks = self._get_playlist_tracks(playlist) tracks_by_uri = {t.uri: t for t in tracks if t.uri} looked_up = self._lookup(*tracks_by_uri.keys()) return [ track.to_dict() for track in [ (looked_up[uri][0] if looked_up.get(uri) else track) for uri, track in tracks_by_uri.items() ] ]
[docs] @action def get_playlist_uri_schemes(self, **__): """ Get the available playlist URI schemes. :return: List of available playlist URI schemes. """ return self._exec({'method': 'core.playlists.get_uri_schemes'})[0]
[docs] @action def create_playlist(self, name: str, uri_scheme: str = 'm3u', **__): """ Create a new playlist. :param name: Playlist name. :param uri_scheme: URI scheme for the playlist (default: ``m3u``). You can get a full list of the available URI schemes that support playlist creation on the Mopidy instance by calling :meth:`.get_playlist_uri_schemes`. :return: .. schema:: mopidy.MopidyPlaylistSchema """ return MopidyPlaylistSchema().dump( self._exec( { 'method': 'core.playlists.create', 'name': name, 'uri_scheme': uri_scheme, } )[0] )
[docs] @action def delete_playlist(self, playlist: str, **__): """ Delete a playlist. :param playlist: Playlist URI. :return: ``True`` if the playlist was deleted, ``False`` otherwise. """ return self._exec({'method': 'core.playlists.delete', 'uri': playlist})[0]
[docs] @action def add_to_playlist( self, playlist: str, resources: Iterable[str], position: Optional[int] = None, allow_duplicates: bool = False, **__, ): """ Add tracks to a playlist. :param playlist: Playlist URI/name. :param resources: List of track URIs. :param position: Position where the tracks will be inserted (default: end of the playlist). :param allow_duplicates: If True, the tracks will be added even if they are already present in the playlist (default: False). :return: The modified playlist. .. schema:: mopidy.MopidyPlaylistSchema """ pl = self._get_playlist(playlist, with_tracks=True) if not allow_duplicates: existing_uris = {t.uri for t in pl.tracks} resources = [t for t in resources if t not in existing_uris] new_tracks = [MopidyTrack(uri=t) for t in resources] if position is not None: pl.tracks = pl.tracks[:position] + new_tracks + pl.tracks[position:] else: pl.tracks += new_tracks self._save_playlist(pl) return pl.to_dict()
[docs] @action def remove_from_playlist( self, playlist: str, resources: Optional[Iterable[Union[str, int]]] = None, from_pos: Optional[int] = None, to_pos: Optional[int] = None, **__, ): """ Remove tracks from a playlist. This action can work in three different ways: - If the ``resources`` parameter is specified, and it contains strings, it will remove the tracks matching the provided URIs. - If the ``resources`` parameter is specified, and it contains integers, it will remove the tracks in the specified positions. - If the ``from_pos`` and ``to_pos`` parameters are specified, it will remove the tracks in the specified range (inclusive). .. note: Positions are 0-based (i.e. the first track has position 0). :param playlist: Playlist URI/name. :param tracks: List of track URIs. :param from_pos: Start position of the slice of tracks to be removed. :param to_pos: End position of the slice of tracks to be removed. :return: The modified playlist. .. schema:: mopidy.MopidyPlaylistSchema """ assert resources or ( from_pos is not None and to_pos is not None ), "Either 'tracks', or 'positions', or 'from_pos' and 'to_pos' must be specified" pl = self._get_playlist(playlist, with_tracks=True) if resources: resources = set(resources) positions = { i for i, t in enumerate(pl.tracks) if t.uri in resources or i in resources } pl.tracks = [t for i, t in enumerate(pl.tracks) if i not in positions] elif from_pos is not None and to_pos is not None: from_pos, to_pos = (min(from_pos, to_pos), max(from_pos, to_pos)) pl.tracks = pl.tracks[: from_pos - 1] + pl.tracks[to_pos:] self._save_playlist(pl) return pl.to_dict()
[docs] @action def playlist_move( self, playlist: str, start: Optional[int] = None, end: Optional[int] = None, position: Optional[int] = None, from_pos: Optional[int] = None, to_pos: Optional[int] = None, **__, ): """ Move tracks in a playlist. This action can work in two different ways: - If the ``start``, ``end`` and ``position`` parameters are specified, it will move an individual track from the position ``start`` to the position ``end`` to the new position ``position``. - If the ``from_pos``, ``to_pos`` and ``position`` parameters are specified, it will move the tracks in the specified range (inclusive) to the new position ``position``. .. note: Positions are 0-based (i.e. the first track has position 0). :param playlist: Playlist URI. :param start: Start position of the slice of tracks to be moved. :param end: End position of the slice of tracks to be moved. :param position: New position where the tracks will be inserted. :return: The modified playlist. .. schema:: mopidy.MopidyPlaylistSchema """ assert (start is not None and end is not None and position is not None) or ( from_pos is not None and to_pos is not None ), "Either 'start', 'end' and 'position', or 'from_pos' and 'to_pos' must be specified" pl = self._get_playlist(playlist, with_tracks=True) if from_pos is not None and to_pos is not None: from_pos, to_pos = (min(from_pos, to_pos), max(from_pos, to_pos)) pl.tracks = ( pl.tracks[:from_pos] + pl.tracks[to_pos : to_pos + 1] + pl.tracks[from_pos + 1 : to_pos] + pl.tracks[from_pos : from_pos + 1] + pl.tracks[to_pos + 1 :] ) elif start is not None and end is not None and position is not None: start, end = (min(start, end), max(start, end)) if start == end: end += 1 if start < position: pl.tracks = ( pl.tracks[:start] + pl.tracks[end : end + (position - start)] + pl.tracks[start:end] + pl.tracks[end + (position - start) :] ) else: pl.tracks = ( pl.tracks[:position] + pl.tracks[start:end] + pl.tracks[position:start] + pl.tracks[end:] ) self._save_playlist(pl) return pl.to_dict()
[docs] @action def playlist_clear(self, playlist: str, **__): """ Remove all the tracks from a playlist. :param playlist: Playlist URI/name. :return: The modified playlist. .. schema:: mopidy.MopidyPlaylistSchema """ pl = self._get_playlist(playlist) pl.tracks = [] self._save_playlist(pl) return pl.to_dict()
[docs] @action def rename_playlist(self, playlist: str, new_name: str, **__): """ Rename a playlist. :param playlist: Playlist URI/name. :param new_name: New playlist name. :return: The modified playlist. .. schema:: mopidy.MopidyPlaylistSchema """ pl = self._get_playlist(playlist, with_tracks=True) pl.name = new_name self._save_playlist(pl) return pl.to_dict()
[docs] @action def get_images(self, resources: Iterable[str], **__) -> Dict[str, Optional[str]]: """ Get the images for a list of URIs. :param resources: List of URIs. :return: Dictionary in the form ``{uri: image_url}``. """ return { uri: next(iter(images or []), {}).get('uri') for uri, images in self._exec( {'method': 'core.library.get_images', 'uris': list(resources)} )[0].items() }
[docs] @action def search( # pylint: disable=redefined-builtin self, filter: dict, exact: bool = False, **__ ): """ Search items that match the given query. :param filter: .. schema:: mopidy.MopidyFilterSchema :param exact: If True, the search will only return exact matches. :return: A list of result, including: - Tracks .. schema:: mopidy.MopidyTrackSchema(many=True) - Albums .. schema:: mopidy.MopidyAlbumSchema(many=True) - Artists .. schema:: mopidy.MopidyArtistSchema(many=True) """ filter = dict(MopidyFilterSchema().load(filter) or {}) uris = filter.pop('uris', None) kwargs = { 'exact': exact, 'query': filter, **({'uris': uris} if uris else {}), } return self._dump_search_results( self._exec({'method': 'core.library.search', **kwargs})[0] )
[docs] @action def find( # pylint: disable=redefined-builtin self, filter: dict, exact: bool = False, **__ ): """ Alias for :meth:`search`, for MPD compatibility. :param filter: .. schema:: mopidy.MopidyFilterSchema :param exact: If True, the search will only return exact matches. :return: .. schema:: mopidy.MopidyTrackSchema(many=True) """ return self.search(filter=filter, exact=exact)
[docs] @action def browse(self, uri: Optional[str] = None): """ Browse the items under the specified URI. :param uri: URI to browse (default: root directory). :return: A list of result under the specified resource, including: - Directories .. schema:: mopidy.MopidyDirectorySchema(many=True) - Tracks .. schema:: mopidy.MopidyTrackSchema(many=True) - Albums .. schema:: mopidy.MopidyAlbumSchema(many=True) - Artists .. schema:: mopidy.MopidyArtistSchema(many=True) """ return self._dump_results( self._exec({'method': 'core.library.browse', 'uri': uri})[0] )
[docs] def main(self): while not self.should_stop(): try: with MopidyClient( config=self.config, status=self._status, stop_event=self._should_stop, playlist_sync=self._playlist_sync, tasks=self._tasks, ) as self._client: self._client.start() wait_for_either(self._should_stop, self._client.closed_event) finally: self._client = None self.wait_stop(10)
__all__ = ['EmptyTrackException', 'MusicMopidyPlugin', 'MopidyStatus', 'MopidyTrack'] # vim:sw=4:ts=4:et: