media_player.py

Go to the documentation of this file.

"""Support for the DirecTV receivers."""
 
from __future__ import annotations
 
import logging
from typing import Any
 
from directv import DIRECTV
 
from homeassistant.components.media_player import (
    MediaPlayerDeviceClass,
    MediaPlayerEntity,
    MediaPlayerEntityFeature,
    MediaPlayerState,
    MediaType,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.util import dt as dt_util
 
from .const import (
    ATTR_MEDIA_CURRENTLY_RECORDING,
    ATTR_MEDIA_RATING,
    ATTR_MEDIA_RECORDED,
    ATTR_MEDIA_START_TIME,
    DOMAIN,
)
from .entity import DIRECTVEntity
 
_LOGGER = logging.getLogger(__name__)
 
KNOWN_MEDIA_TYPES = {MediaType.MOVIE, MediaType.MUSIC, MediaType.TVSHOW}
 
SUPPORT_DTV = (
    MediaPlayerEntityFeature.PAUSE
    | MediaPlayerEntityFeature.TURN_ON
    | MediaPlayerEntityFeature.TURN_OFF
    | MediaPlayerEntityFeature.PLAY_MEDIA
    | MediaPlayerEntityFeature.STOP
    | MediaPlayerEntityFeature.NEXT_TRACK
    | MediaPlayerEntityFeature.PREVIOUS_TRACK
    | MediaPlayerEntityFeature.PLAY
)
 
SUPPORT_DTV_CLIENT = (
    MediaPlayerEntityFeature.PAUSE
    | MediaPlayerEntityFeature.PLAY_MEDIA
    | MediaPlayerEntityFeature.STOP
    | MediaPlayerEntityFeature.NEXT_TRACK
    | MediaPlayerEntityFeature.PREVIOUS_TRACK
    | MediaPlayerEntityFeature.PLAY
)
 
 
async def async_setup_entry(
    hass: HomeAssistant,
    entry: ConfigEntry,
    async_add_entities: AddEntitiesCallback,
) -> None:
    """Set up the DirecTV config entry."""
    dtv = hass.data[DOMAIN][entry.entry_id]
 
    async_add_entities(
        (
            DIRECTVMediaPlayer(
                dtv=dtv,
                name=str.title(location.name),
                address=location.address,
            )
            for location in dtv.device.locations
        ),
        True,
    )
 
 
class DIRECTVMediaPlayer(DIRECTVEntity, MediaPlayerEntity):
    """Representation of a DirecTV receiver on the network."""
 
    def __init__(self, *, dtv: DIRECTV, name: str, address: str = "0") -> None:
        """Initialize DirecTV media player."""
        super().__init__(
            dtv=dtv,
            name=name,
            address=address,
        )
 
        self._attr_unique_id_attr_unique_id = self._device_id_device_id
        self._attr_device_class_attr_device_class = MediaPlayerDeviceClass.RECEIVER
        self._attr_available_attr_available = False
 
        self._is_recorded_is_recorded = None
        self._is_standby_is_standby = True
        self._last_position_last_position = None
        self._last_update_last_update = None
        self._paused_paused = None
        self._program_program = None
 
    async def async_update(self) -> None:
        """Retrieve latest state."""
        state = await self.dtvdtv.state(self._address_address)
        self._attr_available_attr_available = state.available
        self._is_standby_is_standby = state.standby
        self._program_program = state.program
 
        if self._is_standby_is_standby:
            self._attr_assumed_state_attr_assumed_state = False
            self._is_recorded_is_recorded = None
            self._last_position_last_position = None
            self._last_update_last_update = None
            self._paused_paused = None
        elif self._program_program is not None:
            self._paused_paused = self._last_position_last_position == self._program_program.position
            self._is_recorded_is_recorded = self._program_program.recorded
            self._last_position_last_position = self._program_program.position
            if not self._paused_paused:
                self._last_update_last_update = dt_util.utcnow()
            self._attr_assumed_state_attr_assumed_state = self._is_recorded_is_recorded
 
    @property
    def extra_state_attributes(self):
        """Return device specific state attributes."""
        if self._is_standby_is_standby:
            return {}
        return {
            ATTR_MEDIA_CURRENTLY_RECORDING: self.media_currently_recordingmedia_currently_recording,
            ATTR_MEDIA_RATING: self.media_ratingmedia_rating,
            ATTR_MEDIA_RECORDED: self.media_recordedmedia_recorded,
            ATTR_MEDIA_START_TIME: self.media_start_timemedia_start_time,
        }
 
    # MediaPlayerEntity properties and methods
    @property
    def state(self) -> MediaPlayerState:
        """Return the state of the device."""
        if self._is_standby_is_standby:
            return MediaPlayerState.OFF
 
        # For recorded media we can determine if it is paused or not.
        # For live media we're unable to determine and will always return
        # playing instead.
        if self._paused_paused:
            return MediaPlayerState.PAUSED
 
        return MediaPlayerState.PLAYING
 
    @property
    def media_content_id(self):
        """Return the content ID of current playing media."""
        if self._is_standby_is_standby or self._program_program is None:
            return None
 
        return self._program_program.program_id
 
    @property
    def media_content_type(self) -> MediaType | None:
        """Return the content type of current playing media."""
        if self._is_standby_is_standby or self._program_program is None:
            return None
 
        if self._program_program.program_type in KNOWN_MEDIA_TYPES:
            return self._program_program.program_type
 
        return MediaType.MOVIE
 
    @property
    def media_duration(self):
        """Return the duration of current playing media in seconds."""
        if self._is_standby_is_standby or self._program_program is None:
            return None
 
        return self._program_program.duration
 
    @property
    def media_position(self):
        """Position of current playing media in seconds."""
        if self._is_standby_is_standby:
            return None
 
        return self._last_position_last_position
 
    @property
    def media_position_updated_at(self):
        """When was the position of the current playing media valid."""
        if self._is_standby_is_standby:
            return None
 
        return self._last_update_last_update
 
    @property
    def media_title(self):
        """Return the title of current playing media."""
        if self._is_standby_is_standby or self._program_program is None:
            return None
 
        if self.media_content_typemedia_content_typemedia_content_typemedia_content_type == MediaType.MUSIC:
            return self._program_program.music_title
 
        return self._program_program.title
 
    @property
    def media_artist(self):
        """Artist of current playing media, music track only."""
        if self._is_standby_is_standby or self._program_program is None:
            return None
 
        return self._program_program.music_artist
 
    @property
    def media_album_name(self):
        """Album name of current playing media, music track only."""
        if self._is_standby_is_standby or self._program_program is None:
            return None
 
        return self._program_program.music_album
 
    @property
    def media_series_title(self):
        """Return the title of current episode of TV show."""
        if self._is_standby_is_standby or self._program_program is None:
            return None
 
        return self._program_program.episode_title
 
    @property
    def media_channel(self):
        """Return the channel current playing media."""
        if self._is_standby_is_standby or self._program_program is None:
            return None
 
        return f"{self._program.channel_name} ({self._program.channel})"
 
    @property
    def source(self):
        """Name of the current input source."""
        if self._is_standby_is_standby or self._program_program is None:
            return None
 
        return self._program_program.channel
 
    @property
    def supported_features(self) -> MediaPlayerEntityFeature:
        """Flag media player features that are supported."""
        return SUPPORT_DTV_CLIENT if self._is_client_is_client else SUPPORT_DTV
 
    @property
    def media_currently_recording(self):
        """If the media is currently being recorded or not."""
        if self._is_standby_is_standby or self._program_program is None:
            return None
 
        return self._program_program.recording
 
    @property
    def media_rating(self):
        """TV Rating of the current playing media."""
        if self._is_standby_is_standby or self._program_program is None:
            return None
 
        return self._program_program.rating
 
    @property
    def media_recorded(self):
        """If the media was recorded or live."""
        if self._is_standby_is_standby:
            return None
 
        return self._is_recorded_is_recorded
 
    @property
    def media_start_time(self):
        """Start time the program aired."""
        if self._is_standby_is_standby or self._program_program is None:
            return None
 
        return dt_util.as_local(self._program_program.start_time)
 
    async def async_turn_on(self) -> None:
        """Turn on the receiver."""
        if self._is_client_is_client:
            raise NotImplementedError
 
        _LOGGER.debug("Turn on %s", self.namename)
        await self.dtvdtv.remote("poweron", self._address_address)
 
    async def async_turn_off(self) -> None:
        """Turn off the receiver."""
        if self._is_client_is_client:
            raise NotImplementedError
 
        _LOGGER.debug("Turn off %s", self.namename)
        await self.dtvdtv.remote("poweroff", self._address_address)
 
    async def async_media_play(self) -> None:
        """Send play command."""
        _LOGGER.debug("Play on %s", self.namename)
        await self.dtvdtv.remote("play", self._address_address)
 
    async def async_media_pause(self) -> None:
        """Send pause command."""
        _LOGGER.debug("Pause on %s", self.namename)
        await self.dtvdtv.remote("pause", self._address_address)
 
    async def async_media_stop(self) -> None:
        """Send stop command."""
        _LOGGER.debug("Stop on %s", self.namename)
        await self.dtvdtv.remote("stop", self._address_address)
 
    async def async_media_previous_track(self) -> None:
        """Send rewind command."""
        _LOGGER.debug("Rewind on %s", self.namename)
        await self.dtvdtv.remote("rew", self._address_address)
 
    async def async_media_next_track(self) -> None:
        """Send fast forward command."""
        _LOGGER.debug("Fast forward on %s", self.namename)
        await self.dtvdtv.remote("ffwd", self._address_address)
 
    async def async_play_media(
        self, media_type: MediaType | str, media_id: str, **kwargs: Any
    ) -> None:
        """Select input source."""
        if media_type != MediaType.CHANNEL:
            _LOGGER.error(
                "Invalid media type %s. Only %s is supported",
                media_type,
                MediaType.CHANNEL,
            )
            return
 
        _LOGGER.debug("Changing channel on %s to %s", self.namename, media_id)
        await self.dtvdtv.tune(media_id, self._address_address)