evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add base class AndroidTVEntity to AndroidTV (#105945)

Browse source
This commit is contained in:
ollo69 2024-02-22 23:38:38 +01:00 • committed by GitHub
parent 2ef71289b9
commit f0f3773858
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 159 additions and 139 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

145
homeassistant/components/androidtv/entity.py Normal file
View file

@ -0,0 +1,145 @@
"""Base AndroidTV Entity."""
from __future__ import annotations
from collections.abc import Awaitable, Callable, Coroutine
import functools
import logging
from typing import Any, Concatenate, ParamSpec, TypeVar
from androidtv.exceptions import LockNotAcquiredException
from androidtv.setup_async import AndroidTVAsync, FireTVAsync
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
ATTR_CONNECTIONS,
ATTR_IDENTIFIERS,
ATTR_MANUFACTURER,
ATTR_MODEL,
ATTR_SW_VERSION,
CONF_HOST,
CONF_NAME,
)
from homeassistant.helpers.device_registry import CONNECTION_NETWORK_MAC, DeviceInfo
from homeassistant.helpers.entity import Entity
from . import ADB_PYTHON_EXCEPTIONS, ADB_TCP_EXCEPTIONS, get_androidtv_mac
from .const import DEVICE_ANDROIDTV, DOMAIN
PREFIX_ANDROIDTV = "Android TV"
PREFIX_FIRETV = "Fire TV"
_LOGGER = logging.getLogger(__name__)
_ADBDeviceT = TypeVar("_ADBDeviceT", bound="AndroidTVEntity")
_R = TypeVar("_R")
_P = ParamSpec("_P")
_FuncType = Callable[Concatenate[_ADBDeviceT, _P], Awaitable[_R]]
_ReturnFuncType = Callable[Concatenate[_ADBDeviceT, _P], Coroutine[Any, Any, _R | None]]
def adb_decorator(
override_available: bool = False,
) -> Callable[[_FuncType[_ADBDeviceT, _P, _R]], _ReturnFuncType[_ADBDeviceT, _P, _R]]:
"""Wrap ADB methods and catch exceptions.
Allows for overriding the available status of the ADB connection via the
`override_available` parameter.
"""
def _adb_decorator(
func: _FuncType[_ADBDeviceT, _P, _R],
) -> _ReturnFuncType[_ADBDeviceT, _P, _R]:
"""Wrap the provided ADB method and catch exceptions."""
@functools.wraps(func)
async def _adb_exception_catcher(
self: _ADBDeviceT, *args: _P.args, **kwargs: _P.kwargs
) -> _R | None:
"""Call an ADB-related method and catch exceptions."""
if not self.available and not override_available:
return None
try:
return await func(self, *args, **kwargs)
except LockNotAcquiredException:
# If the ADB lock could not be acquired, skip this command
_LOGGER.info(
(
"ADB command %s not executed because the connection is"
" currently in use"
),
func.__name__,
)
return None
except self.exceptions as err:
_LOGGER.error(
(
"Failed to execute an ADB command. ADB connection re-"
"establishing attempt in the next update. Error: %s"
),
err,
)
await self.aftv.adb_close()
# pylint: disable-next=protected-access
self._attr_available = False
return None
except Exception:
# An unforeseen exception occurred. Close the ADB connection so that
# it doesn't happen over and over again, then raise the exception.
await self.aftv.adb_close()
# pylint: disable-next=protected-access
self._attr_available = False
raise
return _adb_exception_catcher
return _adb_decorator
class AndroidTVEntity(Entity):
"""Defines a base AndroidTV entity."""
_attr_has_entity_name = True
def __init__(
self,
aftv: AndroidTVAsync | FireTVAsync,
entry: ConfigEntry,
entry_data: dict[str, Any],
) -> None:
"""Initialize the AndroidTV base entity."""
self.aftv = aftv
self._attr_unique_id = entry.unique_id
self._entry_data = entry_data
device_class = aftv.DEVICE_CLASS
device_type = (
PREFIX_ANDROIDTV if device_class == DEVICE_ANDROIDTV else PREFIX_FIRETV
)
# CONF_NAME may be present in entry.data for configuration imported from YAML
device_name = entry.data.get(
CONF_NAME, f"{device_type} {entry.data[CONF_HOST]}"
)
info = aftv.device_properties
model = info.get(ATTR_MODEL)
self._attr_device_info = DeviceInfo(
model=f"{model} ({device_type})" if model else device_type,
name=device_name,
)
if self.unique_id:
self._attr_device_info[ATTR_IDENTIFIERS] = {(DOMAIN, self.unique_id)}
if manufacturer := info.get(ATTR_MANUFACTURER):
self._attr_device_info[ATTR_MANUFACTURER] = manufacturer
if sw_version := info.get(ATTR_SW_VERSION):
self._attr_device_info[ATTR_SW_VERSION] = sw_version
if mac := get_androidtv_mac(info):
self._attr_device_info[ATTR_CONNECTIONS] = {(CONNECTION_NETWORK_MAC, mac)}
# ADB exceptions to catch
if not aftv.adb_server_ip:
# Using "adb_shell" (Python ADB implementation)
self.exceptions = ADB_PYTHON_EXCEPTIONS
else:
# Using "pure-python-adb" (communicate with ADB server)
self.exceptions = ADB_TCP_EXCEPTIONS

146
homeassistant/components/androidtv/media_player.py
View file

@ -1,15 +1,12 @@
"""Support for functionality to interact with Android / Fire TV devices."""
from __future__ import annotations
from collections.abc import Awaitable, Callable, Coroutine
from datetime import timedelta
import functools
import hashlib
import logging
from typing import Any, Concatenate, ParamSpec, TypeVar
from typing import Any
from androidtv.constants import APPS, KEYS
from androidtv.exceptions import LockNotAcquiredException
from androidtv.setup_async import AndroidTVAsync, FireTVAsync
import voluptuous as vol
@ -21,23 +18,13 @@ from homeassistant.components.media_player import (
MediaPlayerState,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
ATTR_COMMAND,
ATTR_CONNECTIONS,
ATTR_MANUFACTURER,
ATTR_MODEL,
ATTR_SW_VERSION,
CONF_HOST,
CONF_NAME,
)
from homeassistant.const import ATTR_COMMAND
from homeassistant.core import HomeAssistant
from homeassistant.helpers import config_validation as cv, entity_platform
from homeassistant.helpers.device_registry import CONNECTION_NETWORK_MAC, DeviceInfo
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.util import Throttle
from . import ADB_PYTHON_EXCEPTIONS, ADB_TCP_EXCEPTIONS, get_androidtv_mac
from .const import (
ANDROID_DEV,
ANDROID_DEV_OPT,
@ -54,10 +41,7 @@ from .const import (
DOMAIN,
SIGNAL_CONFIG_ENTITY,
)
_ADBDeviceT = TypeVar("_ADBDeviceT", bound="ADBDevice")
_R = TypeVar("_R")
_P = ParamSpec("_P")
from .entity import AndroidTVEntity, adb_decorator
_LOGGER = logging.getLogger(__name__)
@ -73,9 +57,6 @@ SERVICE_DOWNLOAD = "download"
SERVICE_LEARN_SENDEVENT = "learn_sendevent"
SERVICE_UPLOAD = "upload"
PREFIX_ANDROIDTV = "Android TV"
PREFIX_FIRETV = "Fire TV"
# Translate from `AndroidTV` / `FireTV` reported state to HA state.
ANDROIDTV_STATES = {
"off": MediaPlayerState.OFF,
@ -92,25 +73,11 @@ async def async_setup_entry(
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up the Android Debug Bridge entity."""
aftv: AndroidTVAsync | FireTVAsync = hass.data[DOMAIN][entry.entry_id][ANDROID_DEV]
entry_data = hass.data[DOMAIN][entry.entry_id]
aftv: AndroidTVAsync | FireTVAsync = entry_data[ANDROID_DEV]
device_class = aftv.DEVICE_CLASS
device_type = (
PREFIX_ANDROIDTV if device_class == DEVICE_ANDROIDTV else PREFIX_FIRETV
)
# CONF_NAME may be present in entry.data for configuration imported from YAML
device_name: str = entry.data.get(
CONF_NAME, f"{device_type} {entry.data[CONF_HOST]}"
)
device_args = [
aftv,
device_name,
device_type,
entry.unique_id,
entry.entry_id,
hass.data[DOMAIN][entry.entry_id],
]
device_args = [aftv, entry, entry_data]
async_add_entities(
[
AndroidTVDevice(*device_args)
@ -146,108 +113,25 @@ async def async_setup_entry(
)
_FuncType = Callable[Concatenate[_ADBDeviceT, _P], Awaitable[_R]]
_ReturnFuncType = Callable[Concatenate[_ADBDeviceT, _P], Coroutine[Any, Any, _R | None]]
def adb_decorator(
override_available: bool = False,
) -> Callable[[_FuncType[_ADBDeviceT, _P, _R]], _ReturnFuncType[_ADBDeviceT, _P, _R]]:
"""Wrap ADB methods and catch exceptions.
Allows for overriding the available status of the ADB connection via the
`override_available` parameter.
"""
def _adb_decorator(
func: _FuncType[_ADBDeviceT, _P, _R],
) -> _ReturnFuncType[_ADBDeviceT, _P, _R]:
"""Wrap the provided ADB method and catch exceptions."""
@functools.wraps(func)
async def _adb_exception_catcher(
self: _ADBDeviceT, *args: _P.args, **kwargs: _P.kwargs
) -> _R | None:
"""Call an ADB-related method and catch exceptions."""
if not self.available and not override_available:
return None
try:
return await func(self, *args, **kwargs)
except LockNotAcquiredException:
# If the ADB lock could not be acquired, skip this command
_LOGGER.info(
(
"ADB command %s not executed because the connection is"
" currently in use"
),
func.__name__,
)
return None
except self.exceptions as err:
_LOGGER.error(
(
"Failed to execute an ADB command. ADB connection re-"
"establishing attempt in the next update. Error: %s"
),
err,
)
await self.aftv.adb_close()
# pylint: disable-next=protected-access
self._attr_available = False
return None
except Exception:
# An unforeseen exception occurred. Close the ADB connection so that
# it doesn't happen over and over again, then raise the exception.
await self.aftv.adb_close()
# pylint: disable-next=protected-access
self._attr_available = False
raise
return _adb_exception_catcher
return _adb_decorator
class ADBDevice(MediaPlayerEntity):
class ADBDevice(AndroidTVEntity, MediaPlayerEntity):
"""Representation of an Android or Fire TV device."""
_attr_device_class = MediaPlayerDeviceClass.TV
_attr_has_entity_name = True
_attr_name = None
def __init__(
self,
aftv: AndroidTVAsync | FireTVAsync,
name: str,
dev_type: str,
unique_id: str,
entry_id: str,
entry: ConfigEntry,
entry_data: dict[str, Any],
) -> None:
"""Initialize the Android / Fire TV device."""
self.aftv = aftv
self._attr_unique_id = unique_id
self._entry_id = entry_id
self._entry_data = entry_data
super().__init__(aftv, entry, entry_data)
self._entry_id = entry.entry_id
self._media_image: tuple[bytes | None, str | None] = None, None
self._attr_media_image_hash = None
info = aftv.device_properties
model = info.get(ATTR_MODEL)
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, unique_id)},
model=f"{model} ({dev_type})" if model else dev_type,
name=name,
)
if manufacturer := info.get(ATTR_MANUFACTURER):
self._attr_device_info[ATTR_MANUFACTURER] = manufacturer
if sw_version := info.get(ATTR_SW_VERSION):
self._attr_device_info[ATTR_SW_VERSION] = sw_version
if mac := get_androidtv_mac(info):
self._attr_device_info[ATTR_CONNECTIONS] = {(CONNECTION_NETWORK_MAC, mac)}
self._app_id_to_name: dict[str, str] = {}
self._app_name_to_id: dict[str, str] = {}
self._get_sources = DEFAULT_GET_SOURCES
@ -256,14 +140,6 @@ class ADBDevice(MediaPlayerEntity):
self.turn_on_command: str | None = None
self.turn_off_command: str | None = None
# ADB exceptions to catch
if not aftv.adb_server_ip:
# Using "adb_shell" (Python ADB implementation)
self.exceptions = ADB_PYTHON_EXCEPTIONS
else:
# Using "pure-python-adb" (communicate with ADB server)
self.exceptions = ADB_TCP_EXCEPTIONS
# Property attributes
self._attr_extra_state_attributes = {
ATTR_ADB_RESPONSE: None,

7
tests/components/androidtv/test_media_player.py
View file

@ -25,11 +25,10 @@ from homeassistant.components.androidtv.const import (
DEVICE_FIRETV,
DOMAIN,
)
from homeassistant.components.androidtv.entity import PREFIX_ANDROIDTV, PREFIX_FIRETV
from homeassistant.components.androidtv.media_player import (
ATTR_DEVICE_PATH,
ATTR_LOCAL_PATH,
PREFIX_ANDROIDTV,
PREFIX_FIRETV,
SERVICE_ADB_COMMAND,
SERVICE_DOWNLOAD,
SERVICE_LEARN_SENDEVENT,
@ -47,8 +46,6 @@ from homeassistant.components.media_player import (
SERVICE_MEDIA_PREVIOUS_TRACK,
SERVICE_MEDIA_STOP,
SERVICE_SELECT_SOURCE,
SERVICE_TURN_OFF,
SERVICE_TURN_ON,
SERVICE_VOLUME_DOWN,
SERVICE_VOLUME_MUTE,
SERVICE_VOLUME_SET,
@ -63,6 +60,8 @@ from homeassistant.const import (
CONF_NAME,
CONF_PORT,
EVENT_HOMEASSISTANT_STOP,
SERVICE_TURN_OFF,
SERVICE_TURN_ON,
STATE_OFF,
STATE_PLAYING,
STATE_STANDBY,