Improve callable annotations (#118024)

homeassistant/components/guardian/coordinator.py

@@ -34,7 +34,7 @@ class GuardianDataUpdateCoordinator(DataUpdateCoordinator[dict[str, Any]]):
         entry: ConfigEntry,
         client: Client,
         api_name: str,
-        api_coro: Callable[..., Coroutine[Any, Any, dict[str, Any]]],
+        api_coro: Callable[[], Coroutine[Any, Any, dict[str, Any]]],
         api_lock: asyncio.Lock,
         valve_controller_uid: str,
     ) -> None:

homeassistant/components/iqvia/__init__.py

@@ -58,7 +58,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
     client.disable_request_retries()
 
     async def async_get_data_from_api(
-        api_coro: Callable[..., Coroutine[Any, Any, dict[str, Any]]],
+        api_coro: Callable[[], Coroutine[Any, Any, dict[str, Any]]],
     ) -> dict[str, Any]:
         """Get data from a particular API coroutine."""
         try:

homeassistant/components/jewish_calendar/binary_sensor.py

@@ -27,7 +27,7 @@ from . import DOMAIN
 class JewishCalendarBinarySensorMixIns(BinarySensorEntityDescription):
     """Binary Sensor description mixin class for Jewish Calendar."""
 
-    is_on: Callable[..., bool] = lambda _: False
+    is_on: Callable[[Zmanim], bool] = lambda _: False
 
 
 @dataclass(frozen=True)

homeassistant/components/lamarzocco/coordinator.py

@@ -3,7 +3,6 @@
 from collections.abc import Callable, Coroutine
 from datetime import timedelta
 import logging
-from typing import Any
 
 from bleak.backends.device import BLEDevice
 from lmcloud import LMCloud as LaMarzoccoClient
@@ -132,11 +131,11 @@ class LaMarzoccoUpdateCoordinator(DataUpdateCoordinator[None]):
 
         self.lm.initialized = True
 
-    async def _async_handle_request(
+    async def _async_handle_request[**_P](
         self,
-        func: Callable[..., Coroutine[None, None, None]],
+        func: Callable[_P, Coroutine[None, None, None]],
-        *args: Any,
+        *args: _P.args,
-        **kwargs: Any,
+        **kwargs: _P.kwargs,
     ) -> None:
         """Handle a request to the API."""
         try:

homeassistant/components/mqtt/device_tracker.py

@@ -103,7 +103,7 @@ class MqttDeviceTracker(MqttEntity, TrackerEntity):
     _default_name = None
     _entity_id_format = device_tracker.ENTITY_ID_FORMAT
     _location_name: str | None = None
-    _value_template: Callable[..., ReceivePayloadType]
+    _value_template: Callable[[ReceivePayloadType], ReceivePayloadType]
 
     @staticmethod
     def config_schema() -> vol.Schema:
@@ -124,7 +124,7 @@ class MqttDeviceTracker(MqttEntity, TrackerEntity):
         @write_state_on_attr_change(self, {"_location_name"})
         def message_received(msg: ReceiveMessage) -> None:
             """Handle new MQTT messages."""
-            payload: ReceivePayloadType = self._value_template(msg.payload)
+            payload = self._value_template(msg.payload)
             if payload == self._config[CONF_PAYLOAD_HOME]:
                 self._location_name = STATE_HOME
             elif payload == self._config[CONF_PAYLOAD_NOT_HOME]:

homeassistant/components/rainmachine/coordinator.py

@@ -2,8 +2,9 @@
 
 from __future__ import annotations
 
-from collections.abc import Awaitable, Callable
+from collections.abc import Callable, Coroutine
 from datetime import timedelta
+from typing import Any
 
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.core import HomeAssistant, callback
@@ -32,7 +33,7 @@ class RainMachineDataUpdateCoordinator(DataUpdateCoordinator[dict]):
         name: str,
         api_category: str,
         update_interval: timedelta,
-        update_method: Callable[..., Awaitable],
+        update_method: Callable[[], Coroutine[Any, Any, dict]],
     ) -> None:
         """Initialize."""
         super().__init__(
@@ -45,7 +46,7 @@ class RainMachineDataUpdateCoordinator(DataUpdateCoordinator[dict]):
         )
 
         self._rebooting = False
-        self._signal_handler_unsubs: list[Callable[..., None]] = []
+        self._signal_handler_unsubs: list[Callable[[], None]] = []
         self.config_entry = entry
         self.signal_reboot_completed = SIGNAL_REBOOT_COMPLETED.format(
             self.config_entry.entry_id

homeassistant/helpers/httpx_client.py

@@ -2,7 +2,7 @@
 
 from __future__ import annotations
 
-from collections.abc import Callable
+from collections.abc import Callable, Coroutine
 import sys
 from typing import Any, Self
 
@@ -105,7 +105,7 @@ def create_async_httpx_client(
 def _async_register_async_client_shutdown(
     hass: HomeAssistant,
     client: httpx.AsyncClient,
-    original_aclose: Callable[..., Any],
+    original_aclose: Callable[[], Coroutine[Any, Any, None]],
 ) -> None:
     """Register httpx AsyncClient aclose on Home Assistant shutdown.
 

homeassistant/helpers/script.py

@@ -1372,7 +1372,7 @@ class Script:
         domain: str,
         *,
         # Used in "Running <running_description>" log message
-        change_listener: Callable[..., Any] | None = None,
+        change_listener: Callable[[], Any] | None = None,
         copy_variables: bool = False,
         log_exceptions: bool = True,
         logger: logging.Logger | None = None,
@@ -1438,7 +1438,7 @@ class Script:
         return self._change_listener
 
     @change_listener.setter
-    def change_listener(self, change_listener: Callable[..., Any]) -> None:
+    def change_listener(self, change_listener: Callable[[], Any]) -> None:
         """Update the change_listener."""
         self._change_listener = change_listener
         if (