Source code for soco.plugins.plex

"""This plugin supports playback from a linked Plex music service.
See: https://support.plex.tv/articles/218168898-installing-plex-for-sonos/

Requires:
    * Plex music service must be linked in the Sonos app
    * Use of 'plexapi' library (https://github.com/pkkid/python-plexapi)
    * Plex server URI used in 'plexapi' must be reachable from Sonos speakers

    Example usage:

        >>> from plexapi.server import PlexServer
        >>> from soco import SoCo
        >>> from soco.plugins.plex import PlexPlugin
        >>>
        >>> s = SoCo("<SPEAKER_IP>")
        >>> plugin = PlexPlugin(s)
        >>>
        >>> plex_uri = "http://1.2.3.4:32400"
        >>> plex_token = "<YOUR_PLEX_TOKEN>"
        >>> plex = PlexServer(plex_uri, token=plex_token)
        >>> music = plex.library.section("Music")
        >>> artist = music.get("Stevie Wonder")
        >>> album = artist.album("Innervisions")
        >>> track = album.tracks()[4]
        >>> playlist = plex.playlist("My Playlist")
        >>>
        >>> plugin.play_now(artist)     # Play all tracks from an artist
        >>> plugin.add_to_queue(track)  # Add track to the end of queue
        >>> pos = plugin.add_to_queue([album, playlist])  # Enqueue multiple
        >>> s.play_from_queue(pos)      # Play items just enqueued
"""

from urllib.parse import quote

from ..core import to_didl_string
from ..data_structures import (
    DidlMusicAlbum,
    DidlMusicArtist,
    DidlMusicTrack,
    DidlPlaylistContainer,
)
from ..exceptions import SoCoException
from ..music_services import MusicService
from ..plugins import SoCoPlugin


PREFIX_LOOKUP = {
    "album": "1004206c",
    "artist": "1005004c",
    "playlist": "1006206c",
    "track": "10036020",
    "albums:directory": "100d2066",
    "artists:directory": "10fe2066",
    "playlists:directory": "10fe2064",
}

PARENT_TYPE = {
    "album": "artist",
    "artist": "artists:directory",
    "playlist": "playlists:directory",
    "track": "album",
}

CLASS_MAPPING = {
    "album": DidlMusicAlbum,
    "artist": DidlMusicArtist,
    "playlist": DidlPlaylistContainer,
    "track": DidlMusicTrack,
}


[docs]class PlexPlugin(SoCoPlugin): """A SoCo plugin for playing Plex media using the plexapi library.""" def __init__(self, soco): """Initialize the plugin.""" super().__init__(soco) self._service_info = None @property def name(self): """Return the name of the plugin.""" return "Plex Plugin" @property def service_name(self): """Return the service name of the Plex music service.""" return "Plex" @property def service_info(self): """Cache and return the service info of the Plex music service.""" if not self._service_info: self._service_info = MusicService.get_data_for_name(self.service_name) return self._service_info @property def service_id(self): """Return the service ID of the Plex music service.""" return self.service_info["ServiceID"] @property def service_type(self): """Return the service type of the Plex music service.""" return self.service_info["ServiceType"]
[docs] def play_now(self, plex_media): """Add the media to the end of the queue and immediately begin playback.""" position = self.add_to_queue(plex_media) self.soco.play_from_queue(position - 1)
[docs] def add_to_queue(self, plex_media, position=0, as_next=False, **kwargs): """Add the provided media to the speaker's playback queue. Args: plex_media (plexapi): The plexapi object representing the Plex media to be enqueued. Can be one of plexapi.audio.Track, plexapi.audio.Album, plexapi.audio.Artist or plexapi.playlist.Playlist. Can also be a list of the above items. position (int): The index (1-based) at which the media should be added. Default is 0 (append to the end of the queue). as_next (bool): Whether this media should be played as the next track in shuffle mode. This only works if "play_mode=SHUFFLE". Note: Enqueuing multi-track items like albums or playlists will select one track randomly as the next item and shuffle the remaining tracks throughout the queue. Returns: int: The index of the first item added to the queue. """ # Handle a list of Plex media items if isinstance(plex_media, list): position_result = first_added_position = None # If inserting into the queue, repeatedly insert the items in reverse order media_items = reversed(plex_media) if (as_next or position) else plex_media for media_item in media_items: if as_next or position: # Insert each item at the initial queue position in reverse order position_result = self.add_to_queue( media_item, as_next=as_next, position=(first_added_position or position), ) else: # Append each item to the end of the queue in order position_result = self.add_to_queue(media_item) first_added_position = first_added_position or position_result if not as_next: return first_added_position return position_result plex_server = plex_media._server # pylint: disable=protected-access try: base_id = "{}:{}".format( plex_server.machineIdentifier, plex_media.librarySectionID ) except AttributeError: base_id = "{}:".format(plex_server.machineIdentifier) item_type = plex_media.TYPE parent_type = PARENT_TYPE[item_type] didl_class = CLASS_MAPPING[item_type] item_uri = "{}:{}:{}".format(base_id, plex_media.ratingKey, item_type) desc = "SA_RINCON{st}_X_#Svc{st}-0-Token".format(st=self.service_type) if item_type == "track": parent_uri = "{}:{}:{}".format( base_id, plex_media.album().ratingKey, parent_type ) elif item_type == "album": parent_uri = "{}:{}:{}".format( base_id, plex_media.artist().ratingKey, parent_type ) elif item_type == "artist": parent_uri = "{}:{}".format( "00020000artist", plex_media.title.split(" ")[0] ) elif item_type == "playlist": if not plex_media.isAudio: raise SoCoException("Non-audio playlists are not supported") parent_uri = "{}:{}".format(base_id, parent_type) item_didl = didl_class( plex_media.title, PREFIX_LOOKUP[parent_type] + quote(parent_uri), PREFIX_LOOKUP[item_type] + quote(item_uri), desc=desc, ) metadata = to_didl_string(item_didl) enqueued_uri = "x-rincon-cpcontainer:{}?sid={}&flags=8300&sn=9".format( item_didl.item_id, self.service_id ) response = self.soco.avTransport.AddURIToQueue( [ ("InstanceID", 0), ("EnqueuedURI", enqueued_uri), ("EnqueuedURIMetaData", metadata), ("DesiredFirstTrackNumberEnqueued", position), ("EnqueueAsNext", int(as_next)), ], **kwargs ) qnumber = response["FirstTrackNumberEnqueued"] return int(qnumber)