Use PEP 695 for function annotations (3) (#117660)

homeassistant/config_entries.py

@@ -125,7 +125,6 @@ SAVE_DELAY = 1
 DISCOVERY_COOLDOWN = 1
 
 _DataT = TypeVar("_DataT", default=Any)
-_R = TypeVar("_R")
 
 
 class ConfigEntryState(Enum):
@@ -1108,7 +1107,7 @@ class ConfigEntry(Generic[_DataT]):
         )
 
     @callback
-    def async_create_task(
+    def async_create_task[_R](
         self,
         hass: HomeAssistant,
         target: Coroutine[Any, Any, _R],
@@ -1132,7 +1131,7 @@ class ConfigEntry(Generic[_DataT]):
         return task
 
     @callback
-    def async_create_background_task(
+    def async_create_background_task[_R](
         self,
         hass: HomeAssistant,
         target: Coroutine[Any, Any, _R],

homeassistant/core.py

@@ -41,7 +41,6 @@ from typing import (
     ParamSpec,
     Self,
     TypedDict,
-    TypeVarTuple,
     cast,
     overload,
 )
@@ -131,15 +130,12 @@ FINAL_WRITE_STAGE_SHUTDOWN_TIMEOUT = 60
 CLOSE_STAGE_SHUTDOWN_TIMEOUT = 30
 
 
-_T = TypeVar("_T")
 _R = TypeVar("_R")
 _R_co = TypeVar("_R_co", covariant=True)
 _P = ParamSpec("_P")
-_Ts = TypeVarTuple("_Ts")
 # Internal; not helpers.typing.UNDEFINED due to circular dependency
 _UNDEF: dict[Any, Any] = {}
 _SENTINEL = object()
-_CallableT = TypeVar("_CallableT", bound=Callable[..., Any])
 _DataT = TypeVar("_DataT", bound=Mapping[str, Any], default=Mapping[str, Any])
 type CALLBACK_TYPE = Callable[[], None]
 
@@ -234,7 +230,7 @@ def validate_state(state: str) -> str:
     return state
 
 
-def callback(func: _CallableT) -> _CallableT:
+def callback[_CallableT: Callable[..., Any]](func: _CallableT) -> _CallableT:
     """Annotation to mark method as safe to call from within the event loop."""
     setattr(func, "_hass_callback", True)
     return func
@@ -562,7 +558,7 @@ class HomeAssistant:
         self.bus.async_fire_internal(EVENT_CORE_CONFIG_UPDATE)
         self.bus.async_fire_internal(EVENT_HOMEASSISTANT_STARTED)
 
-    def add_job(
+    def add_job[*_Ts](
         self, target: Callable[[*_Ts], Any] | Coroutine[Any, Any, Any], *args: *_Ts
     ) -> None:
         """Add a job to be executed by the event loop or by an executor.
@@ -586,7 +582,7 @@ class HomeAssistant:
 
     @overload
     @callback
-    def async_add_job(
+    def async_add_job[_R, *_Ts](
         self,
         target: Callable[[*_Ts], Coroutine[Any, Any, _R]],
         *args: *_Ts,
@@ -595,7 +591,7 @@ class HomeAssistant:
 
     @overload
     @callback
-    def async_add_job(
+    def async_add_job[_R, *_Ts](
         self,
         target: Callable[[*_Ts], Coroutine[Any, Any, _R] | _R],
         *args: *_Ts,
@@ -604,7 +600,7 @@ class HomeAssistant:
 
     @overload
     @callback
-    def async_add_job(
+    def async_add_job[_R](
         self,
         target: Coroutine[Any, Any, _R],
         *args: Any,
@@ -612,7 +608,7 @@ class HomeAssistant:
     ) -> asyncio.Future[_R] | None: ...
 
     @callback
-    def async_add_job(
+    def async_add_job[_R, *_Ts](
         self,
         target: Callable[[*_Ts], Coroutine[Any, Any, _R] | _R]
         | Coroutine[Any, Any, _R],
@@ -650,7 +646,7 @@ class HomeAssistant:
 
     @overload
     @callback
-    def async_add_hass_job(
+    def async_add_hass_job[_R](
         self,
         hassjob: HassJob[..., Coroutine[Any, Any, _R]],
         *args: Any,
@@ -660,7 +656,7 @@ class HomeAssistant:
 
     @overload
     @callback
-    def async_add_hass_job(
+    def async_add_hass_job[_R](
         self,
         hassjob: HassJob[..., Coroutine[Any, Any, _R] | _R],
         *args: Any,
@@ -669,7 +665,7 @@ class HomeAssistant:
     ) -> asyncio.Future[_R] | None: ...
 
     @callback
-    def async_add_hass_job(
+    def async_add_hass_job[_R](
         self,
         hassjob: HassJob[..., Coroutine[Any, Any, _R] | _R],
         *args: Any,
@@ -775,7 +771,7 @@ class HomeAssistant:
         )
 
     @callback
-    def async_create_task(
+    def async_create_task[_R](
         self,
         target: Coroutine[Any, Any, _R],
         name: str | None = None,
@@ -801,7 +797,7 @@ class HomeAssistant:
         return self.async_create_task_internal(target, name, eager_start)
 
     @callback
-    def async_create_task_internal(
+    def async_create_task_internal[_R](
         self,
         target: Coroutine[Any, Any, _R],
         name: str | None = None,
@@ -832,7 +828,7 @@ class HomeAssistant:
         return task
 
     @callback
-    def async_create_background_task(
+    def async_create_background_task[_R](
         self, target: Coroutine[Any, Any, _R], name: str, eager_start: bool = True
     ) -> asyncio.Task[_R]:
         """Create a task from within the event loop.
@@ -864,7 +860,7 @@ class HomeAssistant:
         return task
 
     @callback
-    def async_add_executor_job(
+    def async_add_executor_job[_T, *_Ts](
         self, target: Callable[[*_Ts], _T], *args: *_Ts
     ) -> asyncio.Future[_T]:
         """Add an executor job from within the event loop."""
@@ -878,7 +874,7 @@ class HomeAssistant:
         return task
 
     @callback
-    def async_add_import_executor_job(
+    def async_add_import_executor_job[_T, *_Ts](
         self, target: Callable[[*_Ts], _T], *args: *_Ts
     ) -> asyncio.Future[_T]:
         """Add an import executor job from within the event loop.
@@ -935,24 +931,24 @@ class HomeAssistant:
 
     @overload
     @callback
-    def async_run_job(
+    def async_run_job[_R, *_Ts](
         self, target: Callable[[*_Ts], Coroutine[Any, Any, _R]], *args: *_Ts
     ) -> asyncio.Future[_R] | None: ...
 
     @overload
     @callback
-    def async_run_job(
+    def async_run_job[_R, *_Ts](
         self, target: Callable[[*_Ts], Coroutine[Any, Any, _R] | _R], *args: *_Ts
     ) -> asyncio.Future[_R] | None: ...
 
     @overload
     @callback
-    def async_run_job(
+    def async_run_job[_R](
         self, target: Coroutine[Any, Any, _R], *args: Any
     ) -> asyncio.Future[_R] | None: ...
 
     @callback
-    def async_run_job(
+    def async_run_job[_R, *_Ts](
         self,
         target: Callable[[*_Ts], Coroutine[Any, Any, _R] | _R]
         | Coroutine[Any, Any, _R],

homeassistant/loader.py

@@ -19,7 +19,7 @@ import pathlib
 import sys
 import time
 from types import ModuleType
-from typing import TYPE_CHECKING, Any, Literal, Protocol, TypedDict, TypeVar, cast
+from typing import TYPE_CHECKING, Any, Literal, Protocol, TypedDict, cast
 
 from awesomeversion import (
     AwesomeVersion,
@@ -49,8 +49,6 @@ if TYPE_CHECKING:
     from .helpers import device_registry as dr
     from .helpers.typing import ConfigType
 
-_CallableT = TypeVar("_CallableT", bound=Callable[..., Any])
-
 _LOGGER = logging.getLogger(__name__)
 
 #
@@ -1574,7 +1572,7 @@ class Helpers:
         return wrapped
 
 
-def bind_hass(func: _CallableT) -> _CallableT:
+def bind_hass[_CallableT: Callable[..., Any]](func: _CallableT) -> _CallableT:
     """Decorate function to indicate that first argument is hass.
 
     The use of this decorator is discouraged, and it should not be used

homeassistant/scripts/benchmark/__init__.py

@@ -10,7 +10,6 @@ from contextlib import suppress
 import json
 import logging
 from timeit import default_timer as timer
-from typing import TypeVar
 
 from homeassistant import core
 from homeassistant.const import EVENT_STATE_CHANGED
@@ -24,8 +23,6 @@ from homeassistant.helpers.json import JSON_DUMP, JSONEncoder
 # mypy: allow-untyped-calls, allow-untyped-defs, no-check-untyped-defs
 # mypy: no-warn-return-any
 
-_CallableT = TypeVar("_CallableT", bound=Callable)
-
 BENCHMARKS: dict[str, Callable] = {}
 
 
@@ -56,7 +53,7 @@ async def run_benchmark(bench):
     await hass.async_stop()
 
 
-def benchmark(func: _CallableT) -> _CallableT:
+def benchmark[_CallableT: Callable](func: _CallableT) -> _CallableT:
     """Decorate to mark a benchmark."""
     BENCHMARKS[func.__name__] = func
     return func

homeassistant/util/__init__.py

@@ -10,15 +10,12 @@ import random
 import re
 import string
 import threading
-from typing import Any, TypeVar
+from typing import Any
 
 import slugify as unicode_slug
 
 from .dt import as_local, utcnow
 
-_T = TypeVar("_T")
-_U = TypeVar("_U")
-
 RE_SANITIZE_FILENAME = re.compile(r"(~|\.\.|/|\\)")
 RE_SANITIZE_PATH = re.compile(r"(~|\.(\.)+)")
 
@@ -61,7 +58,7 @@ def repr_helper(inp: Any) -> str:
     return str(inp)
 
 
-def convert(
+def convert[_T, _U](
     value: _T | None, to_type: Callable[[_T], _U], default: _U | None = None
 ) -> _U | None:
     """Convert value to to_type, returns default if fails."""

homeassistant/util/async_.py

@@ -7,17 +7,14 @@ from collections.abc import Awaitable, Callable, Coroutine
 import concurrent.futures
 import logging
 import threading
-from typing import Any, TypeVar, TypeVarTuple
+from typing import Any
 
 _LOGGER = logging.getLogger(__name__)
 
 _SHUTDOWN_RUN_CALLBACK_THREADSAFE = "_shutdown_run_callback_threadsafe"
 
-_T = TypeVar("_T")
-_Ts = TypeVarTuple("_Ts")
 
-
+def create_eager_task[_T](
-def create_eager_task(
     coro: Coroutine[Any, Any, _T],
     *,
     name: str | None = None,
@@ -45,7 +42,7 @@ def cancelling(task: Future[Any]) -> bool:
     return bool((cancelling_ := getattr(task, "cancelling", None)) and cancelling_())
 
 
-def run_callback_threadsafe(
+def run_callback_threadsafe[_T, *_Ts](
     loop: AbstractEventLoop, callback: Callable[[*_Ts], _T], *args: *_Ts
 ) -> concurrent.futures.Future[_T]:
     """Submit a callback object to a given event loop.

homeassistant/util/enum.py

@@ -15,11 +15,9 @@ if TYPE_CHECKING:
 else:
     from functools import lru_cache
 
-_EnumT = TypeVar("_EnumT", bound=Enum)
-
 
 @lru_cache
-def try_parse_enum(cls: type[_EnumT], value: Any) -> _EnumT | None:
+def try_parse_enum[_EnumT: Enum](cls: type[_EnumT], value: Any) -> _EnumT | None:
     """Try to parse the value into an Enum.
 
     Return None if parsing fails.

homeassistant/util/logging.py

@@ -9,7 +9,7 @@ import logging
 import logging.handlers
 import queue
 import traceback
-from typing import Any, TypeVar, TypeVarTuple, cast, overload
+from typing import Any, cast, overload
 
 from homeassistant.core import (
     HassJobType,
@@ -18,9 +18,6 @@ from homeassistant.core import (
     get_hassjob_callable_job_type,
 )
 
-_T = TypeVar("_T")
-_Ts = TypeVarTuple("_Ts")
-
 
 class HomeAssistantQueueHandler(logging.handlers.QueueHandler):
     """Process the log in another thread."""
@@ -80,7 +77,7 @@ def async_activate_log_queue_handler(hass: HomeAssistant) -> None:
     listener.start()
 
 
-def log_exception(format_err: Callable[[*_Ts], Any], *args: *_Ts) -> None:
+def log_exception[*_Ts](format_err: Callable[[*_Ts], Any], *args: *_Ts) -> None:
     """Log an exception with additional context."""
     module = inspect.getmodule(inspect.stack(context=0)[1].frame)
     if module is not None:
@@ -98,7 +95,7 @@ def log_exception(format_err: Callable[[*_Ts], Any], *args: *_Ts) -> None:
     logging.getLogger(module_name).error("%s\n%s", friendly_msg, exc_msg)
 
 
-async def _async_wrapper(
+async def _async_wrapper[*_Ts](
     async_func: Callable[[*_Ts], Coroutine[Any, Any, None]],
     format_err: Callable[[*_Ts], Any],
     *args: *_Ts,
@@ -110,7 +107,7 @@ async def _async_wrapper(
         log_exception(format_err, *args)
 
 
-def _sync_wrapper(
+def _sync_wrapper[*_Ts](
     func: Callable[[*_Ts], Any], format_err: Callable[[*_Ts], Any], *args: *_Ts
 ) -> None:
     """Catch and log exception."""
@@ -121,7 +118,7 @@ def _sync_wrapper(
 
 
 @callback
-def _callback_wrapper(
+def _callback_wrapper[*_Ts](
     func: Callable[[*_Ts], Any], format_err: Callable[[*_Ts], Any], *args: *_Ts
 ) -> None:
     """Catch and log exception."""
@@ -132,7 +129,7 @@ def _callback_wrapper(
 
 
 @overload
-def catch_log_exception(
+def catch_log_exception[*_Ts](
     func: Callable[[*_Ts], Coroutine[Any, Any, Any]],
     format_err: Callable[[*_Ts], Any],
     job_type: HassJobType | None = None,
@@ -140,14 +137,14 @@ def catch_log_exception(
 
 
 @overload
-def catch_log_exception(
+def catch_log_exception[*_Ts](
     func: Callable[[*_Ts], Any],
     format_err: Callable[[*_Ts], Any],
     job_type: HassJobType | None = None,
 ) -> Callable[[*_Ts], None] | Callable[[*_Ts], Coroutine[Any, Any, None]]: ...
 
 
-def catch_log_exception(
+def catch_log_exception[*_Ts](
     func: Callable[[*_Ts], Any],
     format_err: Callable[[*_Ts], Any],
     job_type: HassJobType | None = None,
@@ -170,7 +167,7 @@ def catch_log_exception(
     return wraps(func)(partial(_sync_wrapper, func, format_err))  # type: ignore[return-value]
 
 
-def catch_log_coro_exception(
+def catch_log_coro_exception[_T, *_Ts](
     target: Coroutine[Any, Any, _T], format_err: Callable[[*_Ts], Any], *args: *_Ts
 ) -> Coroutine[Any, Any, _T | None]:
     """Decorate a coroutine to catch and log exceptions."""
@@ -186,7 +183,7 @@ def catch_log_coro_exception(
     return coro_wrapper(*args)
 
 
-def async_create_catching_coro(
+def async_create_catching_coro[_T](
     target: Coroutine[Any, Any, _T],
 ) -> Coroutine[Any, Any, _T | None]:
     """Wrap a coroutine to catch and log exceptions.

homeassistant/util/percentage.py

@@ -2,8 +2,6 @@
 
 from __future__ import annotations
 
-from typing import TypeVar
-
 from .scaling import (  # noqa: F401
     int_states_in_range,
     scale_ranged_value_to_int_range,
@@ -11,10 +9,8 @@ from .scaling import (  # noqa: F401
     states_in_range,
 )
 
-_T = TypeVar("_T")
 
-
+def ordered_list_item_to_percentage[_T](ordered_list: list[_T], item: _T) -> int:
-def ordered_list_item_to_percentage(ordered_list: list[_T], item: _T) -> int:
     """Determine the percentage of an item in an ordered list.
 
     When using this utility for fan speeds, do not include "off"
@@ -37,7 +33,7 @@ def ordered_list_item_to_percentage(ordered_list: list[_T], item: _T) -> int:
     return (list_position * 100) // list_len
 
 
-def percentage_to_ordered_list_item(ordered_list: list[_T], percentage: int) -> _T:
+def percentage_to_ordered_list_item[_T](ordered_list: list[_T], percentage: int) -> _T:
     """Find the item that most closely matches the percentage in an ordered list.
 
     When using this utility for fan speeds, do not include "off"

homeassistant/util/variance.py

@@ -5,31 +5,30 @@ from __future__ import annotations
 from collections.abc import Callable
 from datetime import datetime, timedelta
 import functools
-from typing import Any, ParamSpec, TypeVar, overload
+from typing import Any, overload
-
-_R = TypeVar("_R", int, float, datetime)
-_P = ParamSpec("_P")
 
 
 @overload
-def ignore_variance(
+def ignore_variance[**_P](
     func: Callable[_P, int], ignored_variance: int
 ) -> Callable[_P, int]: ...
 
 
 @overload
-def ignore_variance(
+def ignore_variance[**_P](
     func: Callable[_P, float], ignored_variance: float
 ) -> Callable[_P, float]: ...
 
 
 @overload
-def ignore_variance(
+def ignore_variance[**_P](
     func: Callable[_P, datetime], ignored_variance: timedelta
 ) -> Callable[_P, datetime]: ...
 
 
-def ignore_variance(func: Callable[_P, _R], ignored_variance: Any) -> Callable[_P, _R]:
+def ignore_variance[**_P, _R: (int, float, datetime)](
+    func: Callable[_P, _R], ignored_variance: Any
+) -> Callable[_P, _R]:
     """Wrap a function that returns old result if new result does not vary enough."""
     last_value: _R | None = None