media_player.py

Go to the documentation of this file.

"""Support to interface with the Plex API."""
 
from __future__ import annotations
 
from collections.abc import Callable
from functools import wraps
import logging
from typing import Any, Concatenate, cast
 
import plexapi.exceptions
import requests.exceptions
 
from homeassistant.components.media_player import (
    DOMAIN as MP_DOMAIN,
    BrowseMedia,
    MediaPlayerEntity,
    MediaPlayerEntityFeature,
    MediaPlayerState,
    MediaType,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import entity_registry as er
from homeassistant.helpers.device_registry import DeviceEntryType, DeviceInfo
from homeassistant.helpers.dispatcher import (
    async_dispatcher_connect,
    async_dispatcher_send,
)
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.network import is_internal_request
 
from .const import (
    COMMON_PLAYERS,
    CONF_SERVER_IDENTIFIER,
    DISPATCHERS,
    DOMAIN,
    NAME_FORMAT,
    PLEX_NEW_MP_SIGNAL,
    PLEX_UPDATE_MEDIA_PLAYER_SESSION_SIGNAL,
    PLEX_UPDATE_MEDIA_PLAYER_SIGNAL,
    PLEX_UPDATE_SENSOR_SIGNAL,
    TRANSIENT_DEVICE_MODELS,
)
from .helpers import get_plex_data, get_plex_server
from .media_browser import browse_media
from .services import process_plex_payload
 
_LOGGER = logging.getLogger(__name__)
 
 
def needs_session[_PlexMediaPlayerT: PlexMediaPlayer, **_P, _R](
    func: Callable[Concatenate[_PlexMediaPlayerT, _P], _R],
) -> Callable[Concatenate[_PlexMediaPlayerT, _P], _R | None]:
    """Ensure session is available for certain attributes."""
 
    @wraps(func)
    def get_session_attribute(
        self: _PlexMediaPlayerT, *args: _P.args, **kwargs: _P.kwargs
    ) -> _R | None:
        if self.session is None:
            return None
        return func(self, *args, **kwargs)
 
    return get_session_attribute
 
 
async def async_setup_entry(
    hass: HomeAssistant,
    config_entry: ConfigEntry,
    async_add_entities: AddEntitiesCallback,
) -> None:
    """Set up Plex media_player from a config entry."""
    server_id = config_entry.data[CONF_SERVER_IDENTIFIER]
    registry = er.async_get(hass)
 
    @callback
    def async_new_media_players(new_entities):
        _async_add_entities(hass, registry, async_add_entities, server_id, new_entities)
 
    unsub = async_dispatcher_connect(
        hass, PLEX_NEW_MP_SIGNAL.format(server_id), async_new_media_players
    )
    get_plex_data(hass)[DISPATCHERS][server_id].append(unsub)
    _LOGGER.debug("New entity listener created")
 
 
@callback
def _async_add_entities(hass, registry, async_add_entities, server_id, new_entities):
    """Set up Plex media_player entities."""
    _LOGGER.debug("New entities: %s", new_entities)
    entities = []
    plexserver = get_plex_server(hass, server_id)
    for entity_params in new_entities:
        plex_mp = PlexMediaPlayer(plexserver, **entity_params)
        entities.append(plex_mp)
 
        # Migration to per-server unique_ids
        old_entity_id = registry.async_get_entity_id(
            MP_DOMAIN, DOMAIN, plex_mp.machine_identifier
        )
        if old_entity_id is not None:
            new_unique_id = f"{server_id}:{plex_mp.machine_identifier}"
            _LOGGER.debug(
                "Migrating unique_id from [%s] to [%s]",
                plex_mp.machine_identifier,
                new_unique_id,
            )
            registry.async_update_entity(old_entity_id, new_unique_id=new_unique_id)
 
    async_add_entities(entities, True)
 
 
class PlexMediaPlayer(MediaPlayerEntity):
    """Representation of a Plex device."""
 
    _attr_available = False
    _attr_should_poll = False
    _attr_state = MediaPlayerState.IDLE
 
    def __init__(self, plex_server, device, player_source, session=None):
        """Initialize the Plex device."""
        self.plex_serverplex_server = plex_server
        self.devicedevice = device
        self.player_sourceplayer_source = player_source
 
        self.device_makedevice_make = None
        self.device_platformdevice_platform = None
        self.device_productdevice_product = None
        self.device_titledevice_title = None
        self.device_versiondevice_version = None
        self.machine_identifiermachine_identifier = device.machineIdentifier
        self.session_devicesession_device = None
 
        self._device_protocol_capabilities_device_protocol_capabilities = None
        self._previous_volume_level_previous_volume_level = 1  # Used in fake muting
        self._volume_level_volume_level = 1  # since we can't retrieve remotely
        self._volume_muted_volume_muted = False  # since we can't retrieve remotely
 
        self._attr_unique_id_attr_unique_id = (
            f"{self.plex_server.machine_identifier}:{self.machine_identifier}"
        )
 
        # Initializes other attributes
        self.sessionsessionsessionsession = session
 
    async def async_added_to_hass(self) -> None:
        """Run when about to be added to hass."""
        _LOGGER.debug("Added %s [%s]", self.entity_identity_id, self.unique_idunique_id)
        self.async_on_removeasync_on_remove(
            async_dispatcher_connect(
                self.hasshass,
                PLEX_UPDATE_MEDIA_PLAYER_SIGNAL.format(self.unique_idunique_id),
                self.async_refresh_media_playerasync_refresh_media_player,
            )
        )
 
        self.async_on_removeasync_on_remove(
            async_dispatcher_connect(
                self.hasshass,
                PLEX_UPDATE_MEDIA_PLAYER_SESSION_SIGNAL.format(self.unique_idunique_id),
                self.async_update_from_websocketasync_update_from_websocket,
            )
        )
 
    @callback
    def async_refresh_media_player(self, device, session, source):
        """Set instance objects and trigger an entity state update."""
        _LOGGER.debug("Refreshing %s [%s / %s]", self.entity_identity_id, device, session)
        self.devicedevice = device
        self.sessionsessionsessionsession = session
        if source:
            self.player_sourceplayer_source = source
        self.async_schedule_update_ha_stateasync_schedule_update_ha_state(True)
 
        async_dispatcher_send(
            self.hasshass,
            PLEX_UPDATE_SENSOR_SIGNAL.format(self.plex_serverplex_server.machine_identifier),
        )
 
    @callback
    def async_update_from_websocket(self, state):
        """Update the entity based on new websocket data."""
        self.update_stateupdate_state(state)
        self.async_write_ha_stateasync_write_ha_state()
 
        async_dispatcher_send(
            self.hasshass,
            PLEX_UPDATE_SENSOR_SIGNAL.format(self.plex_serverplex_server.machine_identifier),
        )
 
    def update(self):
        """Refresh key device data."""
        if not self.sessionsessionsessionsession:
            self.force_idleforce_idle()
            if not self.devicedevice:
                self._attr_available_attr_available_attr_available = False
                return
 
        self._attr_available_attr_available_attr_available = True
 
        try:
            device_url = self.devicedevice.url("/")
        except plexapi.exceptions.BadRequest:
            device_url = "127.0.0.1"
        if "127.0.0.1" in device_url:
            self.devicedevice.proxyThroughServer()
        self._device_protocol_capabilities_device_protocol_capabilities = self.devicedevice.protocolCapabilities
 
        for device in filter(None, [self.devicedevice, self.session_devicesession_device]):
            self.device_makedevice_make = self.device_makedevice_make or device.device
            self.device_platformdevice_platform = self.device_platformdevice_platform or device.platform
            self.device_productdevice_product = self.device_productdevice_product or device.product
            self.device_titledevice_title = self.device_titledevice_title or device.title
            self.device_versiondevice_version = self.device_versiondevice_version or device.version
 
        name_parts = [self.device_productdevice_product, self.device_titledevice_title or self.device_platformdevice_platform]
        if (self.device_productdevice_product in COMMON_PLAYERS) and self.device_makedevice_make:
            # Add more context in name for likely duplicates
            name_parts.append(self.device_makedevice_make)
        if self.usernameusername and self.usernameusername != self.plex_serverplex_server.owner:
            # Prepend username for shared/managed clients
            name_parts.insert(0, self.usernameusername)
        self._attr_name_attr_name = NAME_FORMAT.format(" - ".join(name_parts))
 
    def force_idle(self):
        """Force client to idle."""
        self._attr_state_attr_state = MediaPlayerState.IDLE
        if self.player_sourceplayer_source == "session":
            self.devicedevice = None
            self.session_devicesession_device = None
            self._attr_available_attr_available_attr_available = False
 
    @property
    def session(self):
        """Return the active session for this player."""
        return self._session_session
 
    @session.setter
    def session(self, session):
        self._session_session = session
        if session:
            self.session_devicesession_device = self.sessionsessionsessionsession.player
            self.update_stateupdate_state(self.sessionsessionsessionsession.state)
        else:
            self._attr_state_attr_state = MediaPlayerState.IDLE
 
    @property
    @needs_session
    def username(self):
        """Return the username of the client owner."""
        return self.sessionsessionsessionsession.username
 
    def update_state(self, state):
        """Set the state of the device, handle session termination."""
        if state == "playing":
            self._attr_state_attr_state = MediaPlayerState.PLAYING
        elif state == "paused":
            self._attr_state_attr_state = MediaPlayerState.PAUSED
        elif state == "stopped":
            self.sessionsessionsessionsession = None
            self.force_idleforce_idle()
        else:
            self._attr_state_attr_state = MediaPlayerState.IDLE
 
    @property
    def _is_player_active(self):
        """Report if the client is playing media."""
        return self.statestatestatestate in {MediaPlayerState.PLAYING, MediaPlayerState.PAUSED}
 
    @property
    def _active_media_plexapi_type(self):
        """Get the active media type required by PlexAPI commands."""
        if self.media_content_typemedia_content_typemedia_content_typemedia_content_type == MediaType.MUSIC:
            return "music"
 
        return "video"
 
    @property
    @needs_session
    def session_key(self):
        """Return current session key."""
        return self.sessionsessionsessionsession.sessionKey
 
    @property
    @needs_session
    def media_library_title(self):
        """Return the library name of playing media."""
        return self.sessionsessionsessionsession.media_library_title
 
    @property
    @needs_session
    def media_content_id(self):
        """Return the content ID of current playing media."""
        return self.sessionsessionsessionsession.media_content_id
 
    @property
    @needs_session
    def media_content_type(self):
        """Return the content type of current playing media."""
        return self.sessionsessionsessionsession.media_content_type
 
    @property
    @needs_session
    def media_content_rating(self):
        """Return the content rating of current playing media."""
        return self.sessionsessionsessionsession.media_content_rating
 
    @property
    @needs_session
    def media_artist(self):
        """Return the artist of current playing media, music track only."""
        return self.sessionsessionsessionsession.media_artist
 
    @property
    @needs_session
    def media_album_name(self):
        """Return the album name of current playing media, music track only."""
        return self.sessionsessionsessionsession.media_album_name
 
    @property
    @needs_session
    def media_album_artist(self):
        """Return the album artist of current playing media, music only."""
        return self.sessionsessionsessionsession.media_album_artist
 
    @property
    @needs_session
    def media_track(self):
        """Return the track number of current playing media, music only."""
        return self.sessionsessionsessionsession.media_track
 
    @property
    @needs_session
    def media_duration(self):
        """Return the duration of current playing media in seconds."""
        return self.sessionsessionsessionsession.media_duration
 
    @property
    @needs_session
    def media_position(self):
        """Return the duration of current playing media in seconds."""
        return self.sessionsessionsessionsession.media_position
 
    @property
    @needs_session
    def media_position_updated_at(self):
        """When was the position of the current playing media valid."""
        return self.sessionsessionsessionsession.media_position_updated_at
 
    @property
    @needs_session
    def media_image_url(self):
        """Return the image URL of current playing media."""
        return self.sessionsessionsessionsession.media_image_url
 
    @property
    @needs_session
    def media_summary(self):
        """Return the summary of current playing media."""
        return self.sessionsessionsessionsession.media_summary
 
    @property
    @needs_session
    def media_title(self):
        """Return the title of current playing media."""
        return self.sessionsessionsessionsession.media_title
 
    @property
    @needs_session
    def media_season(self):
        """Return the season of current playing media (TV Show only)."""
        return self.sessionsessionsessionsession.media_season
 
    @property
    @needs_session
    def media_series_title(self):
        """Return the title of the series of current playing media."""
        return self.sessionsessionsessionsession.media_series_title
 
    @property
    @needs_session
    def media_episode(self):
        """Return the episode of current playing media (TV Show only)."""
        return self.sessionsessionsessionsession.media_episode
 
    @property
    def supported_features(self) -> MediaPlayerEntityFeature:
        """Flag media player features that are supported."""
        if self.devicedevice and "playback" in self._device_protocol_capabilities_device_protocol_capabilities:
            return (
                MediaPlayerEntityFeature.PAUSE
                | MediaPlayerEntityFeature.PREVIOUS_TRACK
                | MediaPlayerEntityFeature.NEXT_TRACK
                | MediaPlayerEntityFeature.STOP
                | MediaPlayerEntityFeature.SEEK
                | MediaPlayerEntityFeature.VOLUME_SET
                | MediaPlayerEntityFeature.PLAY
                | MediaPlayerEntityFeature.PLAY_MEDIA
                | MediaPlayerEntityFeature.VOLUME_MUTE
                | MediaPlayerEntityFeature.BROWSE_MEDIA
            )
 
        return (
            MediaPlayerEntityFeature.BROWSE_MEDIA | MediaPlayerEntityFeature.PLAY_MEDIA
        )
 
    def set_volume_level(self, volume: float) -> None:
        """Set volume level, range 0..1."""
        if self.devicedevice and "playback" in self._device_protocol_capabilities_device_protocol_capabilities:
            self.devicedevice.setVolume(int(volume * 100), self._active_media_plexapi_type_active_media_plexapi_type)
            self._volume_level_volume_level = volume  # store since we can't retrieve
 
    @property
    def volume_level(self):
        """Return the volume level of the client (0..1)."""
        if (
            self._is_player_active_is_player_active
            and self.devicedevice
            and "playback" in self._device_protocol_capabilities_device_protocol_capabilities
        ):
            return self._volume_level_volume_level
        return None
 
    @property
    def is_volume_muted(self):
        """Return boolean if volume is currently muted."""
        if self._is_player_active_is_player_active and self.devicedevice:
            return self._volume_muted_volume_muted
        return None
 
    def mute_volume(self, mute: bool) -> None:
        """Mute the volume.
 
        Since we can't actually mute, we'll:
        - On mute, store volume and set volume to 0
        - On unmute, set volume to previously stored volume
        """
        if not (self.devicedevice and "playback" in self._device_protocol_capabilities_device_protocol_capabilities):
            return
 
        self._volume_muted_volume_muted = mute
        if mute:
            self._previous_volume_level_previous_volume_level = self._volume_level_volume_level
            self.set_volume_levelset_volume_levelset_volume_level(0)
        else:
            self.set_volume_levelset_volume_levelset_volume_level(self._previous_volume_level_previous_volume_level)
 
    def media_play(self) -> None:
        """Send play command."""
        if self.devicedevice and "playback" in self._device_protocol_capabilities_device_protocol_capabilities:
            self.devicedevice.play(self._active_media_plexapi_type_active_media_plexapi_type)
 
    def media_pause(self) -> None:
        """Send pause command."""
        if self.devicedevice and "playback" in self._device_protocol_capabilities_device_protocol_capabilities:
            self.devicedevice.pause(self._active_media_plexapi_type_active_media_plexapi_type)
 
    def media_stop(self) -> None:
        """Send stop command."""
        if self.devicedevice and "playback" in self._device_protocol_capabilities_device_protocol_capabilities:
            self.devicedevice.stop(self._active_media_plexapi_type_active_media_plexapi_type)
 
    def media_seek(self, position: float) -> None:
        """Send the seek command."""
        if self.devicedevice and "playback" in self._device_protocol_capabilities_device_protocol_capabilities:
            self.devicedevice.seekTo(position * 1000, self._active_media_plexapi_type_active_media_plexapi_type)
 
    def media_next_track(self) -> None:
        """Send next track command."""
        if self.devicedevice and "playback" in self._device_protocol_capabilities_device_protocol_capabilities:
            self.devicedevice.skipNext(self._active_media_plexapi_type_active_media_plexapi_type)
 
    def media_previous_track(self) -> None:
        """Send previous track command."""
        if self.devicedevice and "playback" in self._device_protocol_capabilities_device_protocol_capabilities:
            self.devicedevice.skipPrevious(self._active_media_plexapi_type_active_media_plexapi_type)
 
    def play_media(
        self, media_type: MediaType | str, media_id: str, **kwargs: Any
    ) -> None:
        """Play a piece of media."""
        if not (self.devicedevice and "playback" in self._device_protocol_capabilities_device_protocol_capabilities):
            raise HomeAssistantError(
                f"Client is not currently accepting playback controls: {self.name}"
            )
 
        result = process_plex_payload(
            self.hasshass, media_type, media_id, default_plex_server=self.plex_serverplex_server
        )
        _LOGGER.debug("Attempting to play %s on %s", result.media, self.namename)
 
        try:
            self.devicedevice.playMedia(result.media, offset=result.offset)
        except requests.exceptions.ConnectTimeout as exc:
            raise HomeAssistantError(
                f"Request failed when playing on {self.name}"
            ) from exc
 
    @property
    def extra_state_attributes(self):
        """Return the scene state attributes."""
        attributes = {}
        for attr in (
            "media_content_rating",
            "media_library_title",
            "player_source",
            "media_summary",
            "username",
        ):
            if value := getattr(self, attr, None):
                attributes[attr] = value
 
        return attributes
 
    @property
    def device_info(self) -> DeviceInfo | None:
        """Return a device description for device registry."""
        if self.machine_identifiermachine_identifier is None:
            return None
 
        if self.device_productdevice_product in TRANSIENT_DEVICE_MODELS:
            return DeviceInfo(
                identifiers={(DOMAIN, "plex.tv-clients")},
                name="Plex Client Service",
                manufacturer="Plex",
                model="Plex Clients",
                entry_type=DeviceEntryType.SERVICE,
            )
 
        return DeviceInfo(
            identifiers={(DOMAIN, self.machine_identifiermachine_identifier)},
            manufacturer=self.device_platformdevice_platform or "Plex",
            model=self.device_productdevice_product or self.device_makedevice_make,
            # Instead of setting the device name to the entity name, plex
            # should be updated to set has_entity_name = True, and set the entity
            # name to None
            name=cast(str | None, self.namename),
            sw_version=self.device_versiondevice_version,
            via_device=(DOMAIN, self.plex_serverplex_server.machine_identifier),
        )
 
    async def async_browse_media(
        self,
        media_content_type: MediaType | str | None = None,
        media_content_id: str | None = None,
    ) -> BrowseMedia:
        """Implement the websocket media browsing helper."""
        is_internal = is_internal_request(self.hasshass)
        return await self.hasshass.async_add_executor_job(
            browse_media,
            self.hasshass,
            is_internal,
            media_content_type,
            media_content_id,
        )