media_player.py

Go to the documentation of this file.

"""Media player support for Android TV Remote."""
 
from __future__ import annotations
 
import asyncio
from typing import Any
 
from androidtvremote2 import AndroidTVRemote, ConnectionClosed
 
from homeassistant.components.media_player import (
    BrowseMedia,
    MediaClass,
    MediaPlayerDeviceClass,
    MediaPlayerEntity,
    MediaPlayerEntityFeature,
    MediaPlayerState,
    MediaType,
)
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.entity_platform import AddEntitiesCallback
 
from . import AndroidTVRemoteConfigEntry
from .const import CONF_APP_ICON, CONF_APP_NAME
from .entity import AndroidTVRemoteBaseEntity
 
PARALLEL_UPDATES = 0
 
 
async def async_setup_entry(
    hass: HomeAssistant,
    config_entry: AndroidTVRemoteConfigEntry,
    async_add_entities: AddEntitiesCallback,
) -> None:
    """Set up the Android TV media player entity based on a config entry."""
    api = config_entry.runtime_data
    async_add_entities([AndroidTVRemoteMediaPlayerEntity(api, config_entry)])
 
 
class AndroidTVRemoteMediaPlayerEntity(AndroidTVRemoteBaseEntity, MediaPlayerEntity):
    """Android TV Remote Media Player Entity."""
 
    _attr_assumed_state = True
    _attr_device_class = MediaPlayerDeviceClass.TV
    _attr_supported_features = (
        MediaPlayerEntityFeature.PAUSE
        | MediaPlayerEntityFeature.VOLUME_STEP
        | MediaPlayerEntityFeature.VOLUME_MUTE
        | MediaPlayerEntityFeature.PREVIOUS_TRACK
        | MediaPlayerEntityFeature.NEXT_TRACK
        | MediaPlayerEntityFeature.TURN_ON
        | MediaPlayerEntityFeature.TURN_OFF
        | MediaPlayerEntityFeature.PLAY
        | MediaPlayerEntityFeature.STOP
        | MediaPlayerEntityFeature.PLAY_MEDIA
        | MediaPlayerEntityFeature.BROWSE_MEDIA
    )
 
    def __init__(
        self, api: AndroidTVRemote, config_entry: AndroidTVRemoteConfigEntry
    ) -> None:
        """Initialize the entity."""
        super().__init__(api, config_entry)
 
        # This task is needed to create a job that sends a key press
        # sequence that can be canceled if concurrency occurs
        self._channel_set_task_channel_set_task: asyncio.Task | None = None
 
    def _update_current_app(self, current_app: str) -> None:
        """Update current app info."""
        self._attr_app_id_attr_app_id = current_app
        self._attr_app_name_attr_app_name = (
            self._apps[current_app].get(CONF_APP_NAME, current_app)
            if current_app in self._apps
            else current_app
        )
 
    def _update_volume_info(self, volume_info: dict[str, str | bool]) -> None:
        """Update volume info."""
        if volume_info.get("max"):
            self._attr_volume_level_attr_volume_level = int(volume_info["level"]) / int(
                volume_info["max"]
            )
            self._attr_is_volume_muted_attr_is_volume_muted = bool(volume_info["muted"])
        else:
            self._attr_volume_level_attr_volume_level = None
            self._attr_is_volume_muted_attr_is_volume_muted = None
 
    @callback
    def _current_app_updated(self, current_app: str) -> None:
        """Update the state when the current app changes."""
        self._update_current_app_update_current_app(current_app)
        self.async_write_ha_stateasync_write_ha_state()
 
    @callback
    def _volume_info_updated(self, volume_info: dict[str, str | bool]) -> None:
        """Update the state when the volume info changes."""
        self._update_volume_info_update_volume_info(volume_info)
        self.async_write_ha_stateasync_write_ha_state()
 
    async def async_added_to_hass(self) -> None:
        """Register callbacks."""
        await super().async_added_to_hass()
 
        self._update_current_app_update_current_app(self._api_api.current_app)
        self._update_volume_info_update_volume_info(self._api_api.volume_info)
 
        self._api_api.add_current_app_updated_callback(self._current_app_updated_current_app_updated)
        self._api_api.add_volume_info_updated_callback(self._volume_info_updated_volume_info_updated)
 
    async def async_will_remove_from_hass(self) -> None:
        """Remove callbacks."""
        await super().async_will_remove_from_hass()
 
        self._api_api.remove_current_app_updated_callback(self._current_app_updated_current_app_updated)
        self._api_api.remove_volume_info_updated_callback(self._volume_info_updated_volume_info_updated)
 
    @property
    def state(self) -> MediaPlayerState:
        """Return the state of the device."""
        if self._attr_is_on_attr_is_on:
            return MediaPlayerState.ON
        return MediaPlayerState.OFF
 
    async def async_turn_on(self) -> None:
        """Turn the Android TV on."""
        if not self._attr_is_on_attr_is_on:
            self._send_key_command_send_key_command("POWER")
 
    async def async_turn_off(self) -> None:
        """Turn the Android TV off."""
        if self._attr_is_on_attr_is_on:
            self._send_key_command_send_key_command("POWER")
 
    async def async_volume_up(self) -> None:
        """Turn volume up for media player."""
        self._send_key_command_send_key_command("VOLUME_UP")
 
    async def async_volume_down(self) -> None:
        """Turn volume down for media player."""
        self._send_key_command_send_key_command("VOLUME_DOWN")
 
    async def async_mute_volume(self, mute: bool) -> None:
        """Mute the volume."""
        if mute != self.is_volume_mutedis_volume_muted:
            self._send_key_command_send_key_command("VOLUME_MUTE")
 
    async def async_media_play(self) -> None:
        """Send play command."""
        self._send_key_command_send_key_command("MEDIA_PLAY")
 
    async def async_media_pause(self) -> None:
        """Send pause command."""
        self._send_key_command_send_key_command("MEDIA_PAUSE")
 
    async def async_media_play_pause(self) -> None:
        """Send play/pause command."""
        self._send_key_command_send_key_command("MEDIA_PLAY_PAUSE")
 
    async def async_media_stop(self) -> None:
        """Send stop command."""
        self._send_key_command_send_key_command("MEDIA_STOP")
 
    async def async_media_previous_track(self) -> None:
        """Send previous track command."""
        self._send_key_command_send_key_command("MEDIA_PREVIOUS")
 
    async def async_media_next_track(self) -> None:
        """Send next track command."""
        self._send_key_command_send_key_command("MEDIA_NEXT")
 
    async def async_play_media(
        self, media_type: MediaType | str, media_id: str, **kwargs: Any
    ) -> None:
        """Play a piece of media."""
        if media_type == MediaType.CHANNEL:
            if not media_id.isnumeric():
                raise ValueError(f"Channel must be numeric: {media_id}")
            if self._channel_set_task_channel_set_task:
                self._channel_set_task_channel_set_task.cancel()
            self._channel_set_task_channel_set_task = asyncio.create_task(
                self._send_key_commands_send_key_commands(list(media_id))
            )
            await self._channel_set_task_channel_set_task
            return
 
        if media_type in [MediaType.URL, MediaType.APP]:
            self._send_launch_app_command_send_launch_app_command(media_id)
            return
 
        raise ValueError(f"Invalid media type: {media_type}")
 
    async def async_browse_media(
        self,
        media_content_type: MediaType | str | None = None,
        media_content_id: str | None = None,
    ) -> BrowseMedia:
        """Browse apps."""
        children = [
            BrowseMedia(
                media_class=MediaClass.APP,
                media_content_type=MediaType.APP,
                media_content_id=app_id,
                title=app.get(CONF_APP_NAME, ""),
                thumbnail=app.get(CONF_APP_ICON, ""),
                can_play=False,
                can_expand=False,
            )
            for app_id, app in self._apps.items()
        ]
        return BrowseMedia(
            title="Applications",
            media_class=MediaClass.DIRECTORY,
            media_content_id="apps",
            media_content_type=MediaType.APPS,
            children_media_class=MediaClass.APP,
            can_play=False,
            can_expand=True,
            children=children,
        )
 
    async def _send_key_commands(
        self, key_codes: list[str], delay_secs: float = 0.1
    ) -> None:
        """Send a key press sequence to Android TV.
 
        The delay is necessary because device may ignore
        some commands if we send the sequence without delay.
        """
        try:
            for key_code in key_codes:
                self._api_api.send_key_command(key_code)
                await asyncio.sleep(delay_secs)
        except ConnectionClosed as exc:
            raise HomeAssistantError(
                "Connection to Android TV device is closed"
            ) from exc