diff --git a/homeassistant/components/august/__init__.py b/homeassistant/components/august/__init__.py index 4e6c2a11b06..a1547778f81 100644 --- a/homeassistant/components/august/__init__.py +++ b/homeassistant/components/august/__init__.py @@ -7,7 +7,7 @@ from collections.abc import Callable, Coroutine, Iterable, ValuesView from datetime import datetime from itertools import chain import logging -from typing import Any, ParamSpec, TypeVar +from typing import Any from aiohttp import ClientError, ClientResponseError from yalexs.activity import ActivityTypes @@ -36,9 +36,6 @@ from .gateway import AugustGateway from .subscriber import AugustSubscriberMixin from .util import async_create_august_clientsession -_R = TypeVar("_R") -_P = ParamSpec("_P") - _LOGGER = logging.getLogger(__name__) API_CACHED_ATTRS = { @@ -403,7 +400,7 @@ class AugustData(AugustSubscriberMixin): hyper_bridge, ) - async def _async_call_api_op_requires_bridge( + async def _async_call_api_op_requires_bridge[**_P, _R]( self, device_id: str, func: Callable[_P, Coroutine[Any, Any, _R]], diff --git a/homeassistant/components/counter/__init__.py b/homeassistant/components/counter/__init__.py index a607a7bdebe..3d68d70e575 100644 --- a/homeassistant/components/counter/__init__.py +++ b/homeassistant/components/counter/__init__.py @@ -3,7 +3,7 @@ from __future__ import annotations import logging -from typing import Any, Self, TypeVar +from typing import Any, Self import voluptuous as vol @@ -23,8 +23,6 @@ from homeassistant.helpers.restore_state import RestoreEntity from homeassistant.helpers.storage import Store from homeassistant.helpers.typing import ConfigType -_T = TypeVar("_T") - _LOGGER = logging.getLogger(__name__) ATTR_INITIAL = "initial" @@ -62,7 +60,7 @@ STORAGE_FIELDS = { } -def _none_to_empty_dict(value: _T | None) -> _T | dict[str, Any]: +def _none_to_empty_dict[_T](value: _T | None) -> _T | dict[str, Any]: if value is None: return {} return value diff --git a/homeassistant/components/cover/__init__.py b/homeassistant/components/cover/__init__.py index ac9c0384dea..9e3184b4822 100644 --- a/homeassistant/components/cover/__init__.py +++ b/homeassistant/components/cover/__init__.py @@ -8,7 +8,7 @@ from enum import IntFlag, StrEnum import functools as ft from functools import cached_property import logging -from typing import Any, ParamSpec, TypeVar, final +from typing import Any, final import voluptuous as vol @@ -54,9 +54,6 @@ SCAN_INTERVAL = timedelta(seconds=15) ENTITY_ID_FORMAT = DOMAIN + ".{}" -_P = ParamSpec("_P") -_R = TypeVar("_R") - class CoverDeviceClass(StrEnum): """Device class for cover.""" @@ -477,7 +474,7 @@ class CoverEntity(Entity, cached_properties=CACHED_PROPERTIES_WITH_ATTR_): else: await self.async_close_cover_tilt(**kwargs) - def _get_toggle_function( + def _get_toggle_function[**_P, _R]( self, fns: dict[str, Callable[_P, _R]] ) -> Callable[_P, _R]: # If we are opening or closing and we support stopping, then we should stop diff --git a/homeassistant/components/diagnostics/util.py b/homeassistant/components/diagnostics/util.py index 9b33b33f1ed..989433e15b2 100644 --- a/homeassistant/components/diagnostics/util.py +++ b/homeassistant/components/diagnostics/util.py @@ -3,14 +3,12 @@ from __future__ import annotations from collections.abc import Iterable, Mapping -from typing import Any, TypeVar, cast, overload +from typing import Any, cast, overload from homeassistant.core import callback from .const import REDACTED -_T = TypeVar("_T") - @overload def async_redact_data(data: Mapping, to_redact: Iterable[Any]) -> dict: # type: ignore[overload-overlap] @@ -18,11 +16,11 @@ def async_redact_data(data: Mapping, to_redact: Iterable[Any]) -> dict: # type: @overload -def async_redact_data(data: _T, to_redact: Iterable[Any]) -> _T: ... +def async_redact_data[_T](data: _T, to_redact: Iterable[Any]) -> _T: ... @callback -def async_redact_data(data: _T, to_redact: Iterable[Any]) -> _T: +def async_redact_data[_T](data: _T, to_redact: Iterable[Any]) -> _T: """Redact sensitive data in a dict.""" if not isinstance(data, (Mapping, list)): return data diff --git a/homeassistant/components/fitbit/api.py b/homeassistant/components/fitbit/api.py index 0f49c0858f5..1eed5acbcca 100644 --- a/homeassistant/components/fitbit/api.py +++ b/homeassistant/components/fitbit/api.py @@ -3,7 +3,7 @@ from abc import ABC, abstractmethod from collections.abc import Callable import logging -from typing import Any, TypeVar, cast +from typing import Any, cast from fitbit import Fitbit from fitbit.exceptions import HTTPException, HTTPUnauthorized @@ -24,9 +24,6 @@ CONF_REFRESH_TOKEN = "refresh_token" CONF_EXPIRES_AT = "expires_at" -_T = TypeVar("_T") - - class FitbitApi(ABC): """Fitbit client library wrapper base class. @@ -129,7 +126,7 @@ class FitbitApi(ABC): dated_results: list[dict[str, Any]] = response[key] return dated_results[-1] - async def _run(self, func: Callable[[], _T]) -> _T: + async def _run[_T](self, func: Callable[[], _T]) -> _T: """Run client command.""" try: return await self._hass.async_add_executor_job(func) diff --git a/homeassistant/components/fronius/__init__.py b/homeassistant/components/fronius/__init__.py index 18129ab0bcc..07271b91f28 100644 --- a/homeassistant/components/fronius/__init__.py +++ b/homeassistant/components/fronius/__init__.py @@ -5,7 +5,7 @@ from __future__ import annotations import asyncio from datetime import datetime, timedelta import logging -from typing import Final, TypeVar +from typing import Final from pyfronius import Fronius, FroniusError @@ -39,8 +39,6 @@ from .coordinator import ( _LOGGER: Final = logging.getLogger(__name__) PLATFORMS: Final = [Platform.SENSOR] -_FroniusCoordinatorT = TypeVar("_FroniusCoordinatorT", bound=FroniusCoordinatorBase) - type FroniusConfigEntry = ConfigEntry[FroniusSolarNet] @@ -255,7 +253,7 @@ class FroniusSolarNet: return inverter_infos @staticmethod - async def _init_optional_coordinator( + async def _init_optional_coordinator[_FroniusCoordinatorT: FroniusCoordinatorBase]( coordinator: _FroniusCoordinatorT, ) -> _FroniusCoordinatorT | None: """Initialize an update coordinator and return it if devices are found.""" diff --git a/homeassistant/components/google_assistant/trait.py b/homeassistant/components/google_assistant/trait.py index 3efeabfa778..e39634a5dd6 100644 --- a/homeassistant/components/google_assistant/trait.py +++ b/homeassistant/components/google_assistant/trait.py @@ -5,7 +5,7 @@ from __future__ import annotations from abc import ABC, abstractmethod from datetime import datetime, timedelta import logging -from typing import Any, TypeVar +from typing import Any from homeassistant.components import ( alarm_control_panel, @@ -242,10 +242,8 @@ COVER_VALVE_DOMAINS = {cover.DOMAIN, valve.DOMAIN} FRIENDLY_DOMAIN = {cover.DOMAIN: "Cover", valve.DOMAIN: "Valve"} -_TraitT = TypeVar("_TraitT", bound="_Trait") - -def register_trait(trait: type[_TraitT]) -> type[_TraitT]: +def register_trait[_TraitT: _Trait](trait: type[_TraitT]) -> type[_TraitT]: """Decorate a class to register a trait.""" TRAITS.append(trait) return trait diff --git a/homeassistant/components/history_stats/sensor.py b/homeassistant/components/history_stats/sensor.py index 0134f4682a5..0b02ddb2a8e 100644 --- a/homeassistant/components/history_stats/sensor.py +++ b/homeassistant/components/history_stats/sensor.py @@ -4,7 +4,7 @@ from __future__ import annotations from abc import abstractmethod import datetime -from typing import Any, TypeVar +from typing import Any import voluptuous as vol @@ -55,10 +55,8 @@ UNITS: dict[str, str] = { } ICON = "mdi:chart-line" -_T = TypeVar("_T", bound=dict[str, Any]) - -def exactly_two_period_keys(conf: _T) -> _T: +def exactly_two_period_keys[_T: dict[str, Any]](conf: _T) -> _T: """Ensure exactly 2 of CONF_PERIOD_KEYS are provided.""" if sum(param in conf for param in CONF_PERIOD_KEYS) != 2: raise vol.Invalid( diff --git a/homeassistant/components/homeassistant/triggers/numeric_state.py b/homeassistant/components/homeassistant/triggers/numeric_state.py index 43cc3d0918e..bc2c95675ad 100644 --- a/homeassistant/components/homeassistant/triggers/numeric_state.py +++ b/homeassistant/components/homeassistant/triggers/numeric_state.py @@ -5,7 +5,7 @@ from __future__ import annotations from collections.abc import Callable from datetime import timedelta import logging -from typing import Any, TypeVar +from typing import Any import voluptuous as vol @@ -41,10 +41,8 @@ from homeassistant.helpers.event import ( from homeassistant.helpers.trigger import TriggerActionType, TriggerInfo from homeassistant.helpers.typing import ConfigType -_T = TypeVar("_T", bound=dict[str, Any]) - -def validate_above_below(value: _T) -> _T: +def validate_above_below[_T: dict[str, Any]](value: _T) -> _T: """Validate that above and below can co-exist.""" above = value.get(CONF_ABOVE) below = value.get(CONF_BELOW) diff --git a/homeassistant/components/improv_ble/config_flow.py b/homeassistant/components/improv_ble/config_flow.py index 370b244dac2..f38f4830ace 100644 --- a/homeassistant/components/improv_ble/config_flow.py +++ b/homeassistant/components/improv_ble/config_flow.py @@ -6,7 +6,7 @@ import asyncio from collections.abc import Callable, Coroutine from dataclasses import dataclass import logging -from typing import Any, TypeVar +from typing import Any from bleak import BleakError from improv_ble_client import ( @@ -30,8 +30,6 @@ from .const import DOMAIN _LOGGER = logging.getLogger(__name__) -_T = TypeVar("_T") - STEP_PROVISION_SCHEMA = vol.Schema( { vol.Required("ssid"): str, @@ -392,7 +390,7 @@ class ImprovBLEConfigFlow(ConfigFlow, domain=DOMAIN): return self.async_show_progress_done(next_step_id="provision") @staticmethod - async def _try_call(func: Coroutine[Any, Any, _T]) -> _T: + async def _try_call[_T](func: Coroutine[Any, Any, _T]) -> _T: """Call the library and abort flow on common errors.""" try: return await func diff --git a/homeassistant/components/linear_garage_door/coordinator.py b/homeassistant/components/linear_garage_door/coordinator.py index 91ff0165163..35ccced3274 100644 --- a/homeassistant/components/linear_garage_door/coordinator.py +++ b/homeassistant/components/linear_garage_door/coordinator.py @@ -6,7 +6,7 @@ from collections.abc import Awaitable, Callable from dataclasses import dataclass from datetime import timedelta import logging -from typing import Any, TypeVar +from typing import Any from linear_garage_door import Linear from linear_garage_door.errors import InvalidLoginError @@ -19,8 +19,6 @@ from homeassistant.helpers.update_coordinator import DataUpdateCoordinator _LOGGER = logging.getLogger(__name__) -_T = TypeVar("_T") - @dataclass class LinearDevice: @@ -63,7 +61,7 @@ class LinearUpdateCoordinator(DataUpdateCoordinator[dict[str, LinearDevice]]): return await self.execute(update_data) - async def execute(self, func: Callable[[Linear], Awaitable[_T]]) -> _T: + async def execute[_T](self, func: Callable[[Linear], Awaitable[_T]]) -> _T: """Execute an API call.""" linear = Linear() try: diff --git a/homeassistant/components/melnor/models.py b/homeassistant/components/melnor/models.py index 933b2972d6a..377a758a2be 100644 --- a/homeassistant/components/melnor/models.py +++ b/homeassistant/components/melnor/models.py @@ -1,7 +1,6 @@ """Melnor integration models.""" from collections.abc import Callable -from typing import TypeVar from melnor_bluetooth.device import Device, Valve @@ -77,14 +76,11 @@ class MelnorZoneEntity(MelnorBluetoothEntity): ) -T = TypeVar("T", bound=EntityDescription) - - -def get_entities_for_valves( +def get_entities_for_valves[_T: EntityDescription]( coordinator: MelnorDataUpdateCoordinator, - descriptions: list[T], + descriptions: list[_T], function: Callable[ - [Valve, T], + [Valve, _T], CoordinatorEntity[MelnorDataUpdateCoordinator], ], ) -> list[CoordinatorEntity[MelnorDataUpdateCoordinator]]: diff --git a/homeassistant/components/radiotherm/__init__.py b/homeassistant/components/radiotherm/__init__.py index d5f1e4c076c..7b2eaba52c4 100644 --- a/homeassistant/components/radiotherm/__init__.py +++ b/homeassistant/components/radiotherm/__init__.py @@ -3,7 +3,7 @@ from __future__ import annotations from collections.abc import Coroutine -from typing import Any, TypeVar +from typing import Any from urllib.error import URLError from radiotherm.validate import RadiothermTstatError @@ -20,10 +20,8 @@ from .util import async_set_time PLATFORMS: list[Platform] = [Platform.CLIMATE, Platform.SWITCH] -_T = TypeVar("_T") - -async def _async_call_or_raise_not_ready( +async def _async_call_or_raise_not_ready[_T]( coro: Coroutine[Any, Any, _T], host: str ) -> _T: """Call a coro or raise ConfigEntryNotReady.""" diff --git a/homeassistant/components/recorder/core.py b/homeassistant/components/recorder/core.py index 108cc721466..65ad5664846 100644 --- a/homeassistant/components/recorder/core.py +++ b/homeassistant/components/recorder/core.py @@ -12,7 +12,7 @@ import queue import sqlite3 import threading import time -from typing import TYPE_CHECKING, Any, TypeVar, cast +from typing import TYPE_CHECKING, Any, cast import psutil_home_assistant as ha_psutil from sqlalchemy import create_engine, event as sqlalchemy_event, exc, select, update @@ -138,8 +138,6 @@ from .util import ( _LOGGER = logging.getLogger(__name__) -T = TypeVar("T") - DEFAULT_URL = "sqlite:///{hass_config_path}" # Controls how often we clean up @@ -366,9 +364,9 @@ class Recorder(threading.Thread): self.queue_task(COMMIT_TASK) @callback - def async_add_executor_job( - self, target: Callable[..., T], *args: Any - ) -> asyncio.Future[T]: + def async_add_executor_job[_T]( + self, target: Callable[..., _T], *args: Any + ) -> asyncio.Future[_T]: """Add an executor job from within the event loop.""" return self.hass.loop.run_in_executor(self._db_executor, target, *args) diff --git a/homeassistant/components/rfxtrx/__init__.py b/homeassistant/components/rfxtrx/__init__.py index fb339f4ba5a..f3466aa704d 100644 --- a/homeassistant/components/rfxtrx/__init__.py +++ b/homeassistant/components/rfxtrx/__init__.py @@ -6,7 +6,7 @@ import binascii from collections.abc import Callable, Mapping import copy import logging -from typing import Any, NamedTuple, TypeVarTuple, cast +from typing import Any, NamedTuple, cast import RFXtrx as rfxtrxmod import voluptuous as vol @@ -55,8 +55,6 @@ DEFAULT_OFF_DELAY = 2.0 SIGNAL_EVENT = f"{DOMAIN}_event" CONNECT_TIMEOUT = 30.0 -_Ts = TypeVarTuple("_Ts") - _LOGGER = logging.getLogger(__name__) @@ -573,7 +571,7 @@ class RfxtrxCommandEntity(RfxtrxEntity): """Initialzie a switch or light device.""" super().__init__(device, device_id, event=event) - async def _async_send( + async def _async_send[*_Ts]( self, fun: Callable[[rfxtrxmod.PySerialTransport, *_Ts], None], *args: *_Ts ) -> None: rfx_object: rfxtrxmod.Connect = self.hass.data[DOMAIN][DATA_RFXOBJECT] diff --git a/homeassistant/components/ring/coordinator.py b/homeassistant/components/ring/coordinator.py index a10f9317bab..1a52fc78988 100644 --- a/homeassistant/components/ring/coordinator.py +++ b/homeassistant/components/ring/coordinator.py @@ -3,7 +3,6 @@ from asyncio import TaskGroup from collections.abc import Callable import logging -from typing import TypeVar, TypeVarTuple from ring_doorbell import AuthenticationError, Ring, RingDevices, RingError, RingTimeout @@ -15,11 +14,8 @@ from .const import NOTIFICATIONS_SCAN_INTERVAL, SCAN_INTERVAL _LOGGER = logging.getLogger(__name__) -_R = TypeVar("_R") -_Ts = TypeVarTuple("_Ts") - -async def _call_api( +async def _call_api[*_Ts, _R]( hass: HomeAssistant, target: Callable[[*_Ts], _R], *args: *_Ts, msg_suffix: str = "" ) -> _R: try: diff --git a/homeassistant/components/soma/__init__.py b/homeassistant/components/soma/__init__.py index cd282a9f276..7b14aaa3c81 100644 --- a/homeassistant/components/soma/__init__.py +++ b/homeassistant/components/soma/__init__.py @@ -4,7 +4,7 @@ from __future__ import annotations from collections.abc import Callable, Coroutine import logging -from typing import Any, TypeVar +from typing import Any from api.soma_api import SomaApi from requests import RequestException @@ -22,8 +22,6 @@ from homeassistant.helpers.typing import ConfigType from .const import API, DOMAIN, HOST, PORT from .utils import is_api_response_success -_SomaEntityT = TypeVar("_SomaEntityT", bound="SomaEntity") - _LOGGER = logging.getLogger(__name__) DEVICES = "devices" @@ -76,7 +74,7 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: return await hass.config_entries.async_unload_platforms(entry, PLATFORMS) -def soma_api_call( +def soma_api_call[_SomaEntityT: SomaEntity]( api_call: Callable[[_SomaEntityT], Coroutine[Any, Any, dict]], ) -> Callable[[_SomaEntityT], Coroutine[Any, Any, dict]]: """Soma api call decorator.""" diff --git a/homeassistant/components/timer/__init__.py b/homeassistant/components/timer/__init__.py index 5da68d99dd6..8927439a6cc 100644 --- a/homeassistant/components/timer/__init__.py +++ b/homeassistant/components/timer/__init__.py @@ -5,7 +5,7 @@ from __future__ import annotations from collections.abc import Callable from datetime import datetime, timedelta import logging -from typing import Any, Self, TypeVar +from typing import Any, Self import voluptuous as vol @@ -29,7 +29,6 @@ from homeassistant.helpers.storage import Store from homeassistant.helpers.typing import ConfigType import homeassistant.util.dt as dt_util -_T = TypeVar("_T") _LOGGER = logging.getLogger(__name__) DOMAIN = "timer" @@ -82,7 +81,7 @@ def _format_timedelta(delta: timedelta) -> str: return f"{int(hours)}:{int(minutes):02}:{int(seconds):02}" -def _none_to_empty_dict(value: _T | None) -> _T | dict[Any, Any]: +def _none_to_empty_dict[_T](value: _T | None) -> _T | dict[Any, Any]: if value is None: return {} return value diff --git a/homeassistant/components/zha/core/helpers.py b/homeassistant/components/zha/core/helpers.py index a47d8ec8bf0..2508dd34fd4 100644 --- a/homeassistant/components/zha/core/helpers.py +++ b/homeassistant/components/zha/core/helpers.py @@ -14,7 +14,7 @@ from dataclasses import dataclass import enum import logging import re -from typing import TYPE_CHECKING, Any, TypeVar, overload +from typing import TYPE_CHECKING, Any, overload import voluptuous as vol import zigpy.exceptions @@ -62,7 +62,6 @@ if TYPE_CHECKING: from .device import ZHADevice from .gateway import ZHAGateway -_T = TypeVar("_T") _LOGGER = logging.getLogger(__name__) @@ -228,7 +227,7 @@ def async_is_bindable_target(source_zha_device, target_zha_device): @callback -def async_get_zha_config_value( +def async_get_zha_config_value[_T]( config_entry: ConfigEntry, section: str, config_key: str, default: _T ) -> _T: """Get the value for the specified configuration from the ZHA config entry.""" diff --git a/homeassistant/components/zha/websocket_api.py b/homeassistant/components/zha/websocket_api.py index 6e34ea01355..70be438bf24 100644 --- a/homeassistant/components/zha/websocket_api.py +++ b/homeassistant/components/zha/websocket_api.py @@ -4,7 +4,7 @@ from __future__ import annotations import asyncio import logging -from typing import TYPE_CHECKING, Any, Literal, NamedTuple, TypeVar, cast +from typing import TYPE_CHECKING, Any, Literal, NamedTuple, cast import voluptuous as vol import zigpy.backups @@ -118,11 +118,8 @@ IEEE_SERVICE = "ieee_based_service" IEEE_SCHEMA = vol.All(cv.string, EUI64.convert) -# typing typevar -_T = TypeVar("_T") - -def _ensure_list_if_present(value: _T | None) -> list[_T] | list[Any] | None: +def _ensure_list_if_present[_T](value: _T | None) -> list[_T] | list[Any] | None: """Wrap value in list if it is provided and not one.""" if value is None: return None