diff --git a/homeassistant/components/aquostv/media_player.py b/homeassistant/components/aquostv/media_player.py index 7160810e0dc..64631ed1948 100644 --- a/homeassistant/components/aquostv/media_player.py +++ b/homeassistant/components/aquostv/media_player.py @@ -4,7 +4,7 @@ from __future__ import annotations from collections.abc import Callable import logging -from typing import Any, Concatenate, ParamSpec, TypeVar +from typing import Any, Concatenate import sharp_aquos_rc import voluptuous as vol @@ -28,9 +28,6 @@ import homeassistant.helpers.config_validation as cv from homeassistant.helpers.entity_platform import AddEntitiesCallback from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType -_SharpAquosTVDeviceT = TypeVar("_SharpAquosTVDeviceT", bound="SharpAquosTVDevice") -_P = ParamSpec("_P") - _LOGGER = logging.getLogger(__name__) DEFAULT_NAME = "Sharp Aquos TV" @@ -85,7 +82,7 @@ def setup_platform( add_entities([SharpAquosTVDevice(name, remote, power_on_enabled)]) -def _retry( +def _retry[_SharpAquosTVDeviceT: SharpAquosTVDevice, **_P]( func: Callable[Concatenate[_SharpAquosTVDeviceT, _P], Any], ) -> Callable[Concatenate[_SharpAquosTVDeviceT, _P], None]: """Handle query retries.""" diff --git a/homeassistant/components/arcam_fmj/media_player.py b/homeassistant/components/arcam_fmj/media_player.py index ca08a2b4d16..9865b459497 100644 --- a/homeassistant/components/arcam_fmj/media_player.py +++ b/homeassistant/components/arcam_fmj/media_player.py @@ -5,7 +5,7 @@ from __future__ import annotations from collections.abc import Callable, Coroutine import functools import logging -from typing import Any, ParamSpec, TypeVar +from typing import Any from arcam.fmj import ConnectionFailed, SourceCodes from arcam.fmj.state import State @@ -36,9 +36,6 @@ from .const import ( SIGNAL_CLIENT_STOPPED, ) -_R = TypeVar("_R") -_P = ParamSpec("_P") - _LOGGER = logging.getLogger(__name__) @@ -64,7 +61,7 @@ async def async_setup_entry( ) -def convert_exception( +def convert_exception[**_P, _R]( func: Callable[_P, Coroutine[Any, Any, _R]], ) -> Callable[_P, Coroutine[Any, Any, _R]]: """Return decorator to convert a connection error into a home assistant error.""" diff --git a/homeassistant/components/braviatv/coordinator.py b/homeassistant/components/braviatv/coordinator.py index 15e6744ceb8..e08e88073f3 100644 --- a/homeassistant/components/braviatv/coordinator.py +++ b/homeassistant/components/braviatv/coordinator.py @@ -7,7 +7,7 @@ from datetime import datetime, timedelta from functools import wraps import logging from types import MappingProxyType -from typing import Any, Concatenate, Final, ParamSpec, TypeVar +from typing import Any, Concatenate, Final from pybravia import ( BraviaAuthError, @@ -35,14 +35,12 @@ from .const import ( SourceType, ) -_BraviaTVCoordinatorT = TypeVar("_BraviaTVCoordinatorT", bound="BraviaTVCoordinator") -_P = ParamSpec("_P") _LOGGER = logging.getLogger(__name__) SCAN_INTERVAL: Final = timedelta(seconds=10) -def catch_braviatv_errors( +def catch_braviatv_errors[_BraviaTVCoordinatorT: BraviaTVCoordinator, **_P]( func: Callable[Concatenate[_BraviaTVCoordinatorT, _P], Awaitable[None]], ) -> Callable[Concatenate[_BraviaTVCoordinatorT, _P], Coroutine[Any, Any, None]]: """Catch Bravia errors.""" diff --git a/homeassistant/components/cloud/http_api.py b/homeassistant/components/cloud/http_api.py index 2d8974ad6a3..e14ee7da7c2 100644 --- a/homeassistant/components/cloud/http_api.py +++ b/homeassistant/components/cloud/http_api.py @@ -9,7 +9,7 @@ import dataclasses from functools import wraps from http import HTTPStatus import logging -from typing import Any, Concatenate, ParamSpec, TypeVar +from typing import Any, Concatenate import aiohttp from aiohttp import web @@ -116,11 +116,7 @@ def async_setup(hass: HomeAssistant) -> None: ) -_HassViewT = TypeVar("_HassViewT", bound=HomeAssistantView) -_P = ParamSpec("_P") - - -def _handle_cloud_errors( +def _handle_cloud_errors[_HassViewT: HomeAssistantView, **_P]( handler: Callable[ Concatenate[_HassViewT, web.Request, _P], Awaitable[web.Response] ], diff --git a/homeassistant/components/decora/light.py b/homeassistant/components/decora/light.py index d598e3e01c9..3f8118a6e5d 100644 --- a/homeassistant/components/decora/light.py +++ b/homeassistant/components/decora/light.py @@ -7,7 +7,7 @@ import copy from functools import wraps import logging import time -from typing import TYPE_CHECKING, Any, Concatenate, ParamSpec, TypeVar +from typing import TYPE_CHECKING, Any, Concatenate from bluepy.btle import BTLEException import decora @@ -29,10 +29,6 @@ if TYPE_CHECKING: from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType -_DecoraLightT = TypeVar("_DecoraLightT", bound="DecoraLight") -_R = TypeVar("_R") -_P = ParamSpec("_P") - _LOGGER = logging.getLogger(__name__) @@ -60,7 +56,7 @@ PLATFORM_SCHEMA = vol.Schema( ) -def retry( +def retry[_DecoraLightT: DecoraLight, **_P, _R]( method: Callable[Concatenate[_DecoraLightT, _P], _R], ) -> Callable[Concatenate[_DecoraLightT, _P], _R | None]: """Retry bluetooth commands.""" diff --git a/homeassistant/components/denonavr/media_player.py b/homeassistant/components/denonavr/media_player.py index 970cd605d2d..8d6df72a67e 100644 --- a/homeassistant/components/denonavr/media_player.py +++ b/homeassistant/components/denonavr/media_player.py @@ -6,7 +6,7 @@ from collections.abc import Awaitable, Callable, Coroutine from datetime import timedelta from functools import wraps import logging -from typing import Any, Concatenate, ParamSpec, TypeVar +from typing import Any, Concatenate from denonavr import DenonAVR from denonavr.const import ( @@ -100,11 +100,6 @@ TELNET_EVENTS = { "Z3", } -_DenonDeviceT = TypeVar("_DenonDeviceT", bound="DenonDevice") -_R = TypeVar("_R") -_P = ParamSpec("_P") - - DENON_STATE_MAPPING = { STATE_ON: MediaPlayerState.ON, STATE_OFF: MediaPlayerState.OFF, @@ -164,7 +159,7 @@ async def async_setup_entry( async_add_entities(entities, update_before_add=True) -def async_log_errors( +def async_log_errors[_DenonDeviceT: DenonDevice, **_P, _R]( func: Callable[Concatenate[_DenonDeviceT, _P], Awaitable[_R]], ) -> Callable[Concatenate[_DenonDeviceT, _P], Coroutine[Any, Any, _R | None]]: """Log errors occurred when calling a Denon AVR receiver. diff --git a/homeassistant/components/dlna_dmr/media_player.py b/homeassistant/components/dlna_dmr/media_player.py index e6348546d7a..443c2101302 100644 --- a/homeassistant/components/dlna_dmr/media_player.py +++ b/homeassistant/components/dlna_dmr/media_player.py @@ -7,7 +7,7 @@ from collections.abc import Awaitable, Callable, Coroutine, Sequence import contextlib from datetime import datetime, timedelta import functools -from typing import Any, Concatenate, ParamSpec, TypeVar +from typing import Any, Concatenate from async_upnp_client.client import UpnpService, UpnpStateVariable from async_upnp_client.const import NotificationSubType @@ -52,11 +52,6 @@ from .data import EventListenAddr, get_domain_data PARALLEL_UPDATES = 0 -_DlnaDmrEntityT = TypeVar("_DlnaDmrEntityT", bound="DlnaDmrEntity") -_R = TypeVar("_R") -_P = ParamSpec("_P") - - _TRANSPORT_STATE_TO_MEDIA_PLAYER_STATE = { TransportState.PLAYING: MediaPlayerState.PLAYING, TransportState.TRANSITIONING: MediaPlayerState.PLAYING, @@ -68,7 +63,7 @@ _TRANSPORT_STATE_TO_MEDIA_PLAYER_STATE = { } -def catch_request_errors( +def catch_request_errors[_DlnaDmrEntityT: DlnaDmrEntity, **_P, _R]( func: Callable[Concatenate[_DlnaDmrEntityT, _P], Awaitable[_R]], ) -> Callable[Concatenate[_DlnaDmrEntityT, _P], Coroutine[Any, Any, _R | None]]: """Catch UpnpError errors.""" diff --git a/homeassistant/components/dlna_dms/dms.py b/homeassistant/components/dlna_dms/dms.py index 2312c7d2e3d..afff1152cca 100644 --- a/homeassistant/components/dlna_dms/dms.py +++ b/homeassistant/components/dlna_dms/dms.py @@ -8,7 +8,7 @@ from dataclasses import dataclass from enum import StrEnum import functools from functools import cached_property -from typing import Any, TypeVar, cast +from typing import Any, cast from async_upnp_client.aiohttp import AiohttpSessionRequester from async_upnp_client.client import UpnpRequester @@ -43,9 +43,6 @@ from .const import ( STREAMABLE_PROTOCOLS, ) -_DlnaDmsDeviceMethod = TypeVar("_DlnaDmsDeviceMethod", bound="DmsDeviceSource") -_R = TypeVar("_R") - class DlnaDmsData: """Storage class for domain global data.""" @@ -124,7 +121,7 @@ class ActionError(DlnaDmsDeviceError): """Error when calling a UPnP Action on the device.""" -def catch_request_errors( +def catch_request_errors[_DlnaDmsDeviceMethod: DmsDeviceSource, _R]( func: Callable[[_DlnaDmsDeviceMethod, str], Coroutine[Any, Any, _R]], ) -> Callable[[_DlnaDmsDeviceMethod, str], Coroutine[Any, Any, _R]]: """Catch UpnpError errors.""" diff --git a/homeassistant/components/duotecno/entity.py b/homeassistant/components/duotecno/entity.py index 7661080f231..3908440a182 100644 --- a/homeassistant/components/duotecno/entity.py +++ b/homeassistant/components/duotecno/entity.py @@ -4,7 +4,7 @@ from __future__ import annotations from collections.abc import Awaitable, Callable, Coroutine from functools import wraps -from typing import Any, Concatenate, ParamSpec, TypeVar +from typing import Any, Concatenate from duotecno.unit import BaseUnit @@ -47,11 +47,7 @@ class DuotecnoEntity(Entity): return self._unit.is_available() -_T = TypeVar("_T", bound="DuotecnoEntity") -_P = ParamSpec("_P") - - -def api_call( +def api_call[_T: DuotecnoEntity, **_P]( func: Callable[Concatenate[_T, _P], Awaitable[None]], ) -> Callable[Concatenate[_T, _P], Coroutine[Any, Any, None]]: """Catch command exceptions.""" diff --git a/homeassistant/components/evil_genius_labs/util.py b/homeassistant/components/evil_genius_labs/util.py index db07cf46918..f3c86f2666f 100644 --- a/homeassistant/components/evil_genius_labs/util.py +++ b/homeassistant/components/evil_genius_labs/util.py @@ -4,16 +4,12 @@ from __future__ import annotations from collections.abc import Awaitable, Callable, Coroutine from functools import wraps -from typing import Any, Concatenate, ParamSpec, TypeVar +from typing import Any, Concatenate from . import EvilGeniusEntity -_EvilGeniusEntityT = TypeVar("_EvilGeniusEntityT", bound=EvilGeniusEntity) -_R = TypeVar("_R") -_P = ParamSpec("_P") - -def update_when_done( +def update_when_done[_EvilGeniusEntityT: EvilGeniusEntity, **_P, _R]( func: Callable[Concatenate[_EvilGeniusEntityT, _P], Awaitable[_R]], ) -> Callable[Concatenate[_EvilGeniusEntityT, _P], Coroutine[Any, Any, _R]]: """Decorate function to trigger update when function is done.""" diff --git a/homeassistant/components/guardian/util.py b/homeassistant/components/guardian/util.py index 6d407f9c7cc..4b9a2835474 100644 --- a/homeassistant/components/guardian/util.py +++ b/homeassistant/components/guardian/util.py @@ -6,7 +6,7 @@ from collections.abc import Callable, Coroutine, Iterable from dataclasses import dataclass from datetime import timedelta from functools import wraps -from typing import TYPE_CHECKING, Any, Concatenate, ParamSpec, TypeVar +from typing import TYPE_CHECKING, Any, Concatenate from aioguardian.errors import GuardianError @@ -20,14 +20,10 @@ from .const import LOGGER if TYPE_CHECKING: from . import GuardianEntity - _GuardianEntityT = TypeVar("_GuardianEntityT", bound=GuardianEntity) - DEFAULT_UPDATE_INTERVAL = timedelta(seconds=30) SIGNAL_REBOOT_REQUESTED = "guardian_reboot_requested_{0}" -_P = ParamSpec("_P") - @dataclass class EntityDomainReplacementStrategy: @@ -64,7 +60,7 @@ def async_finish_entity_domain_replacements( @callback -def convert_exceptions_to_homeassistant_error( +def convert_exceptions_to_homeassistant_error[_GuardianEntityT: GuardianEntity, **_P]( func: Callable[Concatenate[_GuardianEntityT, _P], Coroutine[Any, Any, Any]], ) -> Callable[Concatenate[_GuardianEntityT, _P], Coroutine[Any, Any, None]]: """Decorate to handle exceptions from the Guardian API.""" diff --git a/homeassistant/components/hassio/handler.py b/homeassistant/components/hassio/handler.py index ff34aa06cf3..a7c8d8774de 100644 --- a/homeassistant/components/hassio/handler.py +++ b/homeassistant/components/hassio/handler.py @@ -7,7 +7,7 @@ from collections.abc import Callable, Coroutine from http import HTTPStatus import logging import os -from typing import Any, ParamSpec +from typing import Any import aiohttp from yarl import URL @@ -24,8 +24,6 @@ from homeassistant.loader import bind_hass from .const import ATTR_DISCOVERY, ATTR_MESSAGE, ATTR_RESULT, DOMAIN, X_HASS_SOURCE -_P = ParamSpec("_P") - _LOGGER = logging.getLogger(__name__) @@ -33,7 +31,7 @@ class HassioAPIError(RuntimeError): """Return if a API trow a error.""" -def _api_bool( +def _api_bool[**_P]( funct: Callable[_P, Coroutine[Any, Any, dict[str, Any]]], ) -> Callable[_P, Coroutine[Any, Any, bool]]: """Return a boolean.""" @@ -49,7 +47,7 @@ def _api_bool( return _wrapper -def api_data( +def api_data[**_P]( funct: Callable[_P, Coroutine[Any, Any, dict[str, Any]]], ) -> Callable[_P, Coroutine[Any, Any, Any]]: """Return data of an api.""" diff --git a/homeassistant/components/hive/__init__.py b/homeassistant/components/hive/__init__.py index fb2733223eb..4001215d90e 100644 --- a/homeassistant/components/hive/__init__.py +++ b/homeassistant/components/hive/__init__.py @@ -5,7 +5,7 @@ from __future__ import annotations from collections.abc import Awaitable, Callable, Coroutine from functools import wraps import logging -from typing import Any, Concatenate, ParamSpec, TypeVar +from typing import Any, Concatenate from aiohttp.web_exceptions import HTTPException from apyhiveapi import Auth, Hive @@ -28,9 +28,6 @@ from homeassistant.helpers.typing import ConfigType from .const import DOMAIN, PLATFORM_LOOKUP, PLATFORMS -_HiveEntityT = TypeVar("_HiveEntityT", bound="HiveEntity") -_P = ParamSpec("_P") - _LOGGER = logging.getLogger(__name__) CONFIG_SCHEMA = vol.Schema( @@ -131,7 +128,7 @@ async def async_remove_config_entry_device( return True -def refresh_system( +def refresh_system[_HiveEntityT: HiveEntity, **_P]( func: Callable[Concatenate[_HiveEntityT, _P], Awaitable[Any]], ) -> Callable[Concatenate[_HiveEntityT, _P], Coroutine[Any, Any, None]]: """Force update all entities after state change.""" diff --git a/homeassistant/components/homematicip_cloud/helpers.py b/homeassistant/components/homematicip_cloud/helpers.py index 43edca4774a..4ac9af48ee1 100644 --- a/homeassistant/components/homematicip_cloud/helpers.py +++ b/homeassistant/components/homematicip_cloud/helpers.py @@ -6,17 +6,12 @@ from collections.abc import Callable, Coroutine from functools import wraps import json import logging -from typing import Any, Concatenate, ParamSpec, TypeGuard, TypeVar +from typing import Any, Concatenate, TypeGuard from homeassistant.exceptions import HomeAssistantError from . import HomematicipGenericEntity -_HomematicipGenericEntityT = TypeVar( - "_HomematicipGenericEntityT", bound=HomematicipGenericEntity -) -_P = ParamSpec("_P") - _LOGGER = logging.getLogger(__name__) @@ -28,7 +23,7 @@ def is_error_response(response: Any) -> TypeGuard[dict[str, Any]]: return False -def handle_errors( +def handle_errors[_HomematicipGenericEntityT: HomematicipGenericEntity, **_P]( func: Callable[ Concatenate[_HomematicipGenericEntityT, _P], Coroutine[Any, Any, Any] ], diff --git a/homeassistant/components/homewizard/helpers.py b/homeassistant/components/homewizard/helpers.py index a3eda4ad565..c4160b0bbb0 100644 --- a/homeassistant/components/homewizard/helpers.py +++ b/homeassistant/components/homewizard/helpers.py @@ -3,7 +3,7 @@ from __future__ import annotations from collections.abc import Callable, Coroutine -from typing import Any, Concatenate, ParamSpec, TypeVar +from typing import Any, Concatenate from homewizard_energy.errors import DisabledError, RequestError @@ -12,11 +12,8 @@ from homeassistant.exceptions import HomeAssistantError from .const import DOMAIN from .entity import HomeWizardEntity -_HomeWizardEntityT = TypeVar("_HomeWizardEntityT", bound=HomeWizardEntity) -_P = ParamSpec("_P") - -def homewizard_exception_handler( +def homewizard_exception_handler[_HomeWizardEntityT: HomeWizardEntity, **_P]( func: Callable[Concatenate[_HomeWizardEntityT, _P], Coroutine[Any, Any, Any]], ) -> Callable[Concatenate[_HomeWizardEntityT, _P], Coroutine[Any, Any, None]]: """Decorate HomeWizard Energy calls to handle HomeWizardEnergy exceptions. diff --git a/homeassistant/components/http/ban.py b/homeassistant/components/http/ban.py index b4e949514b8..dd5f1ed1b05 100644 --- a/homeassistant/components/http/ban.py +++ b/homeassistant/components/http/ban.py @@ -10,7 +10,7 @@ from http import HTTPStatus from ipaddress import IPv4Address, IPv6Address, ip_address import logging from socket import gethostbyaddr, herror -from typing import Any, Concatenate, Final, ParamSpec, TypeVar +from typing import Any, Concatenate, Final from aiohttp.web import ( AppKey, @@ -32,9 +32,6 @@ from homeassistant.util import dt as dt_util, yaml from .const import KEY_HASS from .view import HomeAssistantView -_HassViewT = TypeVar("_HassViewT", bound=HomeAssistantView) -_P = ParamSpec("_P") - _LOGGER: Final = logging.getLogger(__name__) KEY_BAN_MANAGER = AppKey["IpBanManager"]("ha_banned_ips_manager") @@ -91,7 +88,7 @@ async def ban_middleware( raise -def log_invalid_auth( +def log_invalid_auth[_HassViewT: HomeAssistantView, **_P]( func: Callable[Concatenate[_HassViewT, Request, _P], Awaitable[Response]], ) -> Callable[Concatenate[_HassViewT, Request, _P], Coroutine[Any, Any, Response]]: """Decorate function to handle invalid auth or failed login attempts."""