jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-23 05:07:41 +00:00
Code Issues Packages Projects Releases Wiki Activity

Type androidtv error decorator (#70976)

Browse Source
This commit is contained in:
Marc Mueller 2022-04-28 20:51:21 +02:00 committed by GitHub
parent d0f1168ff0
commit a46b38d648
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 20 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

24
homeassistant/components/androidtv/media_player.py
View File

@ -1,9 +1,11 @@
"""Support for functionality to interact with Android TV / Fire TV devices.""" """Support for functionality to interact with Android TV / Fire TV devices."""
from __future__ import annotations from __future__ import annotations
from collections.abc import Awaitable, Callable, Coroutine
from datetime import datetime from datetime import datetime
import functools import functools
import logging import logging
from typing import Any, TypeVar
from adb_shell.exceptions import ( from adb_shell.exceptions import (
AdbTimeoutError, AdbTimeoutError,
@ -14,6 +16,7 @@ from adb_shell.exceptions import (
) )
from androidtv.constants import APPS, KEYS from androidtv.constants import APPS, KEYS
from androidtv.exceptions import LockNotAcquiredException from androidtv.exceptions import LockNotAcquiredException
from typing_extensions import Concatenate, ParamSpec
import voluptuous as vol import voluptuous as vol
from homeassistant.components import persistent_notification from homeassistant.components import persistent_notification
@ -61,6 +64,10 @@ from .const import (
SIGNAL_CONFIG_ENTITY, SIGNAL_CONFIG_ENTITY,
) )
_ADBDeviceT = TypeVar("_ADBDeviceT", bound="ADBDevice")
_R = TypeVar("_R")
_P = ParamSpec("_P")
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
ATTR_ADB_RESPONSE = "adb_response" ATTR_ADB_RESPONSE = "adb_response"
@ -144,18 +151,27 @@ async def async_setup_entry(
) )
def adb_decorator(override_available=False): def adb_decorator(
override_available: bool = False,
) -> Callable[
[Callable[Concatenate[_ADBDeviceT, _P], Awaitable[_R]]],
Callable[Concatenate[_ADBDeviceT, _P], Coroutine[Any, Any, _R | None]],
]:
"""Wrap ADB methods and catch exceptions. """Wrap ADB methods and catch exceptions.
Allows for overriding the available status of the ADB connection via the Allows for overriding the available status of the ADB connection via the
`override_available` parameter. `override_available` parameter.
""" """
def _adb_decorator(func): def _adb_decorator(
func: Callable[Concatenate[_ADBDeviceT, _P], Awaitable[_R]]
) -> Callable[Concatenate[_ADBDeviceT, _P], Coroutine[Any, Any, _R | None]]:
"""Wrap the provided ADB method and catch exceptions.""" """Wrap the provided ADB method and catch exceptions."""
@functools.wraps(func) @functools.wraps(func)
async def _adb_exception_catcher(self, *args, **kwargs): async def _adb_exception_catcher(
self: _ADBDeviceT, *args: _P.args, **kwargs: _P.kwargs
) -> _R | None:
"""Call an ADB-related method and catch exceptions.""" """Call an ADB-related method and catch exceptions."""
# pylint: disable=protected-access # pylint: disable=protected-access
if not self.available and not override_available: if not self.available and not override_available:
@ -168,7 +184,7 @@ def adb_decorator(override_available=False):
_LOGGER.info( _LOGGER.info(
"ADB command not executed because the connection is currently in use" "ADB command not executed because the connection is currently in use"
) )
return return None
except self.exceptions as err: except self.exceptions as err:
_LOGGER.error( _LOGGER.error(
"Failed to execute an ADB command. ADB connection re-" "Failed to execute an ADB command. ADB connection re-"