diff --git a/homeassistant/components/iaqualink/__init__.py b/homeassistant/components/iaqualink/__init__.py index 33697dfb2cc..fd03168714d 100644 --- a/homeassistant/components/iaqualink/__init__.py +++ b/homeassistant/components/iaqualink/__init__.py @@ -6,7 +6,7 @@ from collections.abc import Awaitable, Callable, Coroutine from datetime import datetime from functools import wraps import logging -from typing import Any, Concatenate, ParamSpec, TypeVar +from typing import Any, Concatenate import httpx from iaqualink.client import AqualinkClient @@ -39,9 +39,6 @@ from homeassistant.helpers.event import async_track_time_interval from .const import DOMAIN, UPDATE_INTERVAL -_AqualinkEntityT = TypeVar("_AqualinkEntityT", bound="AqualinkEntity") -_P = ParamSpec("_P") - _LOGGER = logging.getLogger(__name__) ATTR_CONFIG = "config" @@ -182,7 +179,7 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: return await hass.config_entries.async_unload_platforms(entry, platforms_to_unload) -def refresh_system( +def refresh_system[_AqualinkEntityT: AqualinkEntity, **_P]( func: Callable[Concatenate[_AqualinkEntityT, _P], Awaitable[Any]], ) -> Callable[Concatenate[_AqualinkEntityT, _P], Coroutine[Any, Any, None]]: """Force update all entities after state change.""" diff --git a/homeassistant/components/kodi/media_player.py b/homeassistant/components/kodi/media_player.py index 27b2d3e0199..46d3d614bfa 100644 --- a/homeassistant/components/kodi/media_player.py +++ b/homeassistant/components/kodi/media_player.py @@ -7,7 +7,7 @@ from datetime import timedelta from functools import wraps import logging import re -from typing import Any, Concatenate, ParamSpec, TypeVar +from typing import Any, Concatenate from jsonrpc_base.jsonrpc import ProtocolError, TransportError from pykodi import CannotConnectError @@ -71,9 +71,6 @@ from .const import ( EVENT_TURN_ON, ) -_KodiEntityT = TypeVar("_KodiEntityT", bound="KodiEntity") -_P = ParamSpec("_P") - _LOGGER = logging.getLogger(__name__) EVENT_KODI_CALL_METHOD_RESULT = "kodi_call_method_result" @@ -231,7 +228,7 @@ async def async_setup_entry( async_add_entities([entity]) -def cmd( +def cmd[_KodiEntityT: KodiEntity, **_P]( func: Callable[Concatenate[_KodiEntityT, _P], Awaitable[Any]], ) -> Callable[Concatenate[_KodiEntityT, _P], Coroutine[Any, Any, None]]: """Catch command exceptions.""" diff --git a/homeassistant/components/lametric/helpers.py b/homeassistant/components/lametric/helpers.py index 24c028da78c..8620b0c7cd9 100644 --- a/homeassistant/components/lametric/helpers.py +++ b/homeassistant/components/lametric/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 demetriek import LaMetricConnectionError, LaMetricError @@ -15,11 +15,8 @@ from .const import DOMAIN from .coordinator import LaMetricDataUpdateCoordinator from .entity import LaMetricEntity -_LaMetricEntityT = TypeVar("_LaMetricEntityT", bound=LaMetricEntity) -_P = ParamSpec("_P") - -def lametric_exception_handler( +def lametric_exception_handler[_LaMetricEntityT: LaMetricEntity, **_P]( func: Callable[Concatenate[_LaMetricEntityT, _P], Coroutine[Any, Any, Any]], ) -> Callable[Concatenate[_LaMetricEntityT, _P], Coroutine[Any, Any, None]]: """Decorate LaMetric calls to handle LaMetric exceptions. diff --git a/homeassistant/components/limitlessled/light.py b/homeassistant/components/limitlessled/light.py index 423cfac4144..182c12eb395 100644 --- a/homeassistant/components/limitlessled/light.py +++ b/homeassistant/components/limitlessled/light.py @@ -4,7 +4,7 @@ from __future__ import annotations from collections.abc import Callable import logging -from typing import Any, Concatenate, ParamSpec, TypeVar, cast +from typing import Any, Concatenate, cast from limitlessled import Color from limitlessled.bridge import Bridge @@ -40,9 +40,6 @@ from homeassistant.helpers.restore_state import RestoreEntity from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType from homeassistant.util.color import color_hs_to_RGB, color_temperature_mired_to_kelvin -_LimitlessLEDGroupT = TypeVar("_LimitlessLEDGroupT", bound="LimitlessLEDGroup") -_P = ParamSpec("_P") - _LOGGER = logging.getLogger(__name__) CONF_BRIDGES = "bridges" @@ -176,7 +173,7 @@ def setup_platform( add_entities(lights) -def state( +def state[_LimitlessLEDGroupT: LimitlessLEDGroup, **_P]( new_state: bool, ) -> Callable[ [Callable[Concatenate[_LimitlessLEDGroupT, int, Pipeline, _P], Any]], diff --git a/homeassistant/components/matter/api.py b/homeassistant/components/matter/api.py index e6a2a6c54d5..39597bc2ab2 100644 --- a/homeassistant/components/matter/api.py +++ b/homeassistant/components/matter/api.py @@ -4,7 +4,7 @@ from __future__ import annotations from collections.abc import Callable, Coroutine from functools import wraps -from typing import Any, Concatenate, ParamSpec +from typing import Any, Concatenate from matter_server.client.models.node import MatterNode from matter_server.common.errors import MatterError @@ -18,8 +18,6 @@ from homeassistant.core import HomeAssistant, callback from .adapter import MatterAdapter from .helpers import MissingNode, get_matter, node_from_ha_device_id -_P = ParamSpec("_P") - ID = "id" TYPE = "type" DEVICE_ID = "device_id" @@ -93,7 +91,7 @@ def async_get_matter_adapter( return _get_matter -def async_handle_failed_command( +def async_handle_failed_command[**_P]( func: Callable[ Concatenate[HomeAssistant, ActiveConnection, dict[str, Any], _P], Coroutine[Any, Any, None], diff --git a/homeassistant/components/modern_forms/__init__.py b/homeassistant/components/modern_forms/__init__.py index a190eb26837..dea7d4fadea 100644 --- a/homeassistant/components/modern_forms/__init__.py +++ b/homeassistant/components/modern_forms/__init__.py @@ -4,7 +4,7 @@ from __future__ import annotations from collections.abc import Callable, Coroutine import logging -from typing import Any, Concatenate, ParamSpec, TypeVar +from typing import Any, Concatenate from aiomodernforms import ModernFormsConnectionError, ModernFormsError @@ -17,11 +17,6 @@ from homeassistant.helpers.update_coordinator import CoordinatorEntity from .const import DOMAIN from .coordinator import ModernFormsDataUpdateCoordinator -_ModernFormsDeviceEntityT = TypeVar( - "_ModernFormsDeviceEntityT", bound="ModernFormsDeviceEntity" -) -_P = ParamSpec("_P") - PLATFORMS = [ Platform.BINARY_SENSOR, Platform.FAN, @@ -61,7 +56,10 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: return unload_ok -def modernforms_exception_handler( +def modernforms_exception_handler[ + _ModernFormsDeviceEntityT: ModernFormsDeviceEntity, + **_P, +]( func: Callable[Concatenate[_ModernFormsDeviceEntityT, _P], Any], ) -> Callable[Concatenate[_ModernFormsDeviceEntityT, _P], Coroutine[Any, Any, None]]: """Decorate Modern Forms calls to handle Modern Forms exceptions. diff --git a/homeassistant/components/otbr/util.py b/homeassistant/components/otbr/util.py index 4374412b8c1..16cf3b60e37 100644 --- a/homeassistant/components/otbr/util.py +++ b/homeassistant/components/otbr/util.py @@ -7,7 +7,7 @@ import dataclasses from functools import wraps import logging import random -from typing import Any, Concatenate, ParamSpec, TypeVar, cast +from typing import Any, Concatenate, cast import python_otbr_api from python_otbr_api import PENDING_DATASET_DELAY_TIMER, tlv_parser @@ -27,9 +27,6 @@ from homeassistant.helpers import issue_registry as ir from .const import DOMAIN -_R = TypeVar("_R") -_P = ParamSpec("_P") - _LOGGER = logging.getLogger(__name__) INFO_URL_SKY_CONNECT = ( @@ -61,7 +58,7 @@ def generate_random_pan_id() -> int: return random.randint(0, 0xFFFE) -def _handle_otbr_error( +def _handle_otbr_error[**_P, _R]( func: Callable[Concatenate[OTBRData, _P], Coroutine[Any, Any, _R]], ) -> Callable[Concatenate[OTBRData, _P], Coroutine[Any, Any, _R]]: """Handle OTBR errors.""" diff --git a/homeassistant/components/plex/media_player.py b/homeassistant/components/plex/media_player.py index 21e52171fe8..1dd79ad27a5 100644 --- a/homeassistant/components/plex/media_player.py +++ b/homeassistant/components/plex/media_player.py @@ -5,7 +5,7 @@ from __future__ import annotations from collections.abc import Callable from functools import wraps import logging -from typing import Any, Concatenate, ParamSpec, TypeVar, cast +from typing import Any, Concatenate, cast import plexapi.exceptions import requests.exceptions @@ -46,14 +46,10 @@ from .helpers import get_plex_data, get_plex_server from .media_browser import browse_media from .services import process_plex_payload -_PlexMediaPlayerT = TypeVar("_PlexMediaPlayerT", bound="PlexMediaPlayer") -_R = TypeVar("_R") -_P = ParamSpec("_P") - _LOGGER = logging.getLogger(__name__) -def needs_session( +def needs_session[_PlexMediaPlayerT: PlexMediaPlayer, **_P, _R]( func: Callable[Concatenate[_PlexMediaPlayerT, _P], _R], ) -> Callable[Concatenate[_PlexMediaPlayerT, _P], _R | None]: """Ensure session is available for certain attributes.""" diff --git a/homeassistant/components/plugwise/util.py b/homeassistant/components/plugwise/util.py index df1069cbbc3..d998711f2b9 100644 --- a/homeassistant/components/plugwise/util.py +++ b/homeassistant/components/plugwise/util.py @@ -1,7 +1,7 @@ """Utilities for Plugwise.""" from collections.abc import Awaitable, Callable, Coroutine -from typing import Any, Concatenate, ParamSpec, TypeVar +from typing import Any, Concatenate from plugwise.exceptions import PlugwiseException @@ -9,12 +9,8 @@ from homeassistant.exceptions import HomeAssistantError from .entity import PlugwiseEntity -_PlugwiseEntityT = TypeVar("_PlugwiseEntityT", bound=PlugwiseEntity) -_R = TypeVar("_R") -_P = ParamSpec("_P") - -def plugwise_command( +def plugwise_command[_PlugwiseEntityT: PlugwiseEntity, **_P, _R]( func: Callable[Concatenate[_PlugwiseEntityT, _P], Awaitable[_R]], ) -> Callable[Concatenate[_PlugwiseEntityT, _P], Coroutine[Any, Any, _R]]: """Decorate Plugwise calls that send commands/make changes to the device. diff --git a/homeassistant/components/rainmachine/switch.py b/homeassistant/components/rainmachine/switch.py index f7be08d71d3..9bb7c4e7448 100644 --- a/homeassistant/components/rainmachine/switch.py +++ b/homeassistant/components/rainmachine/switch.py @@ -6,7 +6,7 @@ import asyncio from collections.abc import Awaitable, Callable, Coroutine from dataclasses import dataclass from datetime import datetime -from typing import Any, Concatenate, ParamSpec, TypeVar +from typing import Any, Concatenate from regenmaschine.errors import RainMachineError import voluptuous as vol @@ -110,11 +110,7 @@ VEGETATION_MAP = { } -_T = TypeVar("_T", bound="RainMachineBaseSwitch") -_P = ParamSpec("_P") - - -def raise_on_request_error( +def raise_on_request_error[_T: RainMachineBaseSwitch, **_P]( func: Callable[Concatenate[_T, _P], Awaitable[None]], ) -> Callable[Concatenate[_T, _P], Coroutine[Any, Any, None]]: """Define a decorator to raise on a request error.""" diff --git a/homeassistant/components/renault/renault_vehicle.py b/homeassistant/components/renault/renault_vehicle.py index 59e1826ce1b..d5c4f78126c 100644 --- a/homeassistant/components/renault/renault_vehicle.py +++ b/homeassistant/components/renault/renault_vehicle.py @@ -8,7 +8,7 @@ from dataclasses import dataclass from datetime import datetime, timedelta from functools import wraps import logging -from typing import Any, Concatenate, ParamSpec, TypeVar, cast +from typing import Any, Concatenate, cast from renault_api.exceptions import RenaultException from renault_api.kamereon import models @@ -22,13 +22,11 @@ from .const import DOMAIN from .coordinator import RenaultDataUpdateCoordinator LOGGER = logging.getLogger(__name__) -_T = TypeVar("_T") -_P = ParamSpec("_P") -def with_error_wrapping( - func: Callable[Concatenate[RenaultVehicleProxy, _P], Awaitable[_T]], -) -> Callable[Concatenate[RenaultVehicleProxy, _P], Coroutine[Any, Any, _T]]: +def with_error_wrapping[**_P, _R]( + func: Callable[Concatenate[RenaultVehicleProxy, _P], Awaitable[_R]], +) -> Callable[Concatenate[RenaultVehicleProxy, _P], Coroutine[Any, Any, _R]]: """Catch Renault errors.""" @wraps(func) @@ -36,7 +34,7 @@ def with_error_wrapping( self: RenaultVehicleProxy, *args: _P.args, **kwargs: _P.kwargs, - ) -> _T: + ) -> _R: """Catch RenaultException errors and raise HomeAssistantError.""" try: return await func(self, *args, **kwargs) diff --git a/homeassistant/components/ring/entity.py b/homeassistant/components/ring/entity.py index 65ccbb8ece4..a4275815450 100644 --- a/homeassistant/components/ring/entity.py +++ b/homeassistant/components/ring/entity.py @@ -1,7 +1,7 @@ """Base class for Ring entity.""" from collections.abc import Callable -from typing import Any, Concatenate, Generic, ParamSpec, cast +from typing import Any, Concatenate, Generic, cast from ring_doorbell import ( AuthenticationError, @@ -26,12 +26,9 @@ _RingCoordinatorT = TypeVar( "_RingCoordinatorT", bound=(RingDataCoordinator | RingNotificationsCoordinator), ) -_RingBaseEntityT = TypeVar("_RingBaseEntityT", bound="RingBaseEntity[Any, Any]") -_R = TypeVar("_R") -_P = ParamSpec("_P") -def exception_wrap( +def exception_wrap[_RingBaseEntityT: RingBaseEntity[Any, Any], **_P, _R]( func: Callable[Concatenate[_RingBaseEntityT, _P], _R], ) -> Callable[Concatenate[_RingBaseEntityT, _P], _R]: """Define a wrapper to catch exceptions and raise HomeAssistant errors.""" diff --git a/homeassistant/components/sensibo/entity.py b/homeassistant/components/sensibo/entity.py index 97ef4dffca7..b13a5f82111 100644 --- a/homeassistant/components/sensibo/entity.py +++ b/homeassistant/components/sensibo/entity.py @@ -4,7 +4,7 @@ from __future__ import annotations import asyncio from collections.abc import Callable, Coroutine -from typing import TYPE_CHECKING, Any, Concatenate, ParamSpec, TypeVar +from typing import TYPE_CHECKING, Any, Concatenate from pysensibo.model import MotionSensor, SensiboDevice @@ -15,11 +15,8 @@ from homeassistant.helpers.update_coordinator import CoordinatorEntity from .const import DOMAIN, LOGGER, SENSIBO_ERRORS, TIMEOUT from .coordinator import SensiboDataUpdateCoordinator -_T = TypeVar("_T", bound="SensiboDeviceBaseEntity") -_P = ParamSpec("_P") - -def async_handle_api_call( +def async_handle_api_call[_T: SensiboDeviceBaseEntity, **_P]( function: Callable[Concatenate[_T, _P], Coroutine[Any, Any, Any]], ) -> Callable[Concatenate[_T, _P], Coroutine[Any, Any, Any]]: """Decorate api calls.""" diff --git a/homeassistant/components/sfr_box/button.py b/homeassistant/components/sfr_box/button.py index 6dc91149d86..f6d3100d692 100644 --- a/homeassistant/components/sfr_box/button.py +++ b/homeassistant/components/sfr_box/button.py @@ -5,7 +5,7 @@ from __future__ import annotations from collections.abc import Awaitable, Callable, Coroutine from dataclasses import dataclass from functools import wraps -from typing import Any, Concatenate, ParamSpec, TypeVar +from typing import Any, Concatenate from sfrbox_api.bridge import SFRBox from sfrbox_api.exceptions import SFRBoxError @@ -26,13 +26,10 @@ from homeassistant.helpers.entity_platform import AddEntitiesCallback from .const import DOMAIN from .models import DomainData -_T = TypeVar("_T") -_P = ParamSpec("_P") - -def with_error_wrapping( - func: Callable[Concatenate[SFRBoxButton, _P], Awaitable[_T]], -) -> Callable[Concatenate[SFRBoxButton, _P], Coroutine[Any, Any, _T]]: +def with_error_wrapping[**_P, _R]( + func: Callable[Concatenate[SFRBoxButton, _P], Awaitable[_R]], +) -> Callable[Concatenate[SFRBoxButton, _P], Coroutine[Any, Any, _R]]: """Catch SFR errors.""" @wraps(func) @@ -40,7 +37,7 @@ def with_error_wrapping( self: SFRBoxButton, *args: _P.args, **kwargs: _P.kwargs, - ) -> _T: + ) -> _R: """Catch SFRBoxError errors and raise HomeAssistantError.""" try: return await func(self, *args, **kwargs) diff --git a/homeassistant/components/spotify/media_player.py b/homeassistant/components/spotify/media_player.py index 1fb7a614049..40bdd19a3eb 100644 --- a/homeassistant/components/spotify/media_player.py +++ b/homeassistant/components/spotify/media_player.py @@ -6,7 +6,7 @@ from asyncio import run_coroutine_threadsafe from collections.abc import Callable from datetime import timedelta import logging -from typing import Any, Concatenate, ParamSpec, TypeVar +from typing import Any, Concatenate import requests from spotipy import SpotifyException @@ -35,10 +35,6 @@ from .browse_media import async_browse_media_internal from .const import DOMAIN, MEDIA_PLAYER_PREFIX, PLAYABLE_MEDIA_TYPES, SPOTIFY_SCOPES from .util import fetch_image_url -_SpotifyMediaPlayerT = TypeVar("_SpotifyMediaPlayerT", bound="SpotifyMediaPlayer") -_R = TypeVar("_R") -_P = ParamSpec("_P") - _LOGGER = logging.getLogger(__name__) SCAN_INTERVAL = timedelta(seconds=30) @@ -86,7 +82,7 @@ async def async_setup_entry( async_add_entities([spotify], True) -def spotify_exception_handler( +def spotify_exception_handler[_SpotifyMediaPlayerT: SpotifyMediaPlayer, **_P, _R]( func: Callable[Concatenate[_SpotifyMediaPlayerT, _P], _R], ) -> Callable[Concatenate[_SpotifyMediaPlayerT, _P], _R | None]: """Decorate Spotify calls to handle Spotify exception.