From f0f3773858a189821f6f447b8184b792a74bd219 Mon Sep 17 00:00:00 2001 From: ollo69 <60491700+ollo69@users.noreply.github.com> Date: Thu, 22 Feb 2024 23:38:38 +0100 Subject: [PATCH] Add base class AndroidTVEntity to AndroidTV (#105945) --- homeassistant/components/androidtv/entity.py | 145 +++++++++++++++++ .../components/androidtv/media_player.py | 146 ++---------------- .../components/androidtv/test_media_player.py | 7 +- 3 files changed, 159 insertions(+), 139 deletions(-) create mode 100644 homeassistant/components/androidtv/entity.py diff --git a/homeassistant/components/androidtv/entity.py b/homeassistant/components/androidtv/entity.py new file mode 100644 index 00000000000..e9cbd435d9b --- /dev/null +++ b/homeassistant/components/androidtv/entity.py @@ -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 diff --git a/homeassistant/components/androidtv/media_player.py b/homeassistant/components/androidtv/media_player.py index bd058ac769e..5e97396b369 100644 --- a/homeassistant/components/androidtv/media_player.py +++ b/homeassistant/components/androidtv/media_player.py @@ -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, diff --git a/tests/components/androidtv/test_media_player.py b/tests/components/androidtv/test_media_player.py index 847bc5c7d2f..da56133abb0 100644 --- a/tests/components/androidtv/test_media_player.py +++ b/tests/components/androidtv/test_media_player.py @@ -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,