Improve debouncer typing (#75436)

.strict-typing

@@ -15,6 +15,7 @@ homeassistant.auth.auth_store
 homeassistant.auth.providers.*
 homeassistant.helpers.area_registry
 homeassistant.helpers.condition
+homeassistant.helpers.debounce
 homeassistant.helpers.discovery
 homeassistant.helpers.entity
 homeassistant.helpers.entity_values

homeassistant/components/flux_led/number.py

@@ -2,8 +2,9 @@
 from __future__ import annotations
 
 from abc import abstractmethod
+from collections.abc import Coroutine
 import logging
-from typing import cast
+from typing import Any, cast
 
 from flux_led.protocol import (
     MUSIC_PIXELS_MAX,
@@ -143,7 +144,7 @@ class FluxConfigNumber(
     ) -> None:
         """Initialize the flux number."""
         super().__init__(coordinator, base_unique_id, name, key)
-        self._debouncer: Debouncer | None = None
+        self._debouncer: Debouncer[Coroutine[Any, Any, None]] | None = None
         self._pending_value: int | None = None
 
     async def async_added_to_hass(self) -> None:

homeassistant/components/plex/__init__.py

@@ -95,7 +95,7 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
         _LOGGER.debug("Scanning for GDM clients")
         gdm.scan(scan_for_clients=True)
 
-    hass.data[PLEX_DOMAIN][GDM_DEBOUNCER] = Debouncer(
+    hass.data[PLEX_DOMAIN][GDM_DEBOUNCER] = Debouncer[None](
         hass,
         _LOGGER,
         cooldown=10,

homeassistant/components/samsungtv/__init__.py

@@ -1,7 +1,7 @@
 """The Samsung TV integration."""
 from __future__ import annotations
 
-from collections.abc import Mapping
+from collections.abc import Coroutine, Mapping
 from functools import partial
 import socket
 from typing import Any
@@ -131,7 +131,7 @@ class DebouncedEntryReloader:
         self.hass = hass
         self.entry = entry
         self.token = self.entry.data.get(CONF_TOKEN)
-        self._debounced_reload = Debouncer(
+        self._debounced_reload: Debouncer[Coroutine[Any, Any, None]] = Debouncer(
             hass,
             LOGGER,
             cooldown=ENTRY_RELOAD_COOLDOWN,

homeassistant/components/shelly/__init__.py

@@ -2,6 +2,7 @@
 from __future__ import annotations
 
 import asyncio
+from collections.abc import Coroutine
 from datetime import timedelta
 from typing import Any, Final, cast
 
@@ -296,7 +297,7 @@ class BlockDeviceWrapper(update_coordinator.DataUpdateCoordinator):
         self.entry = entry
         self.device = device
 
-        self._debounced_reload = Debouncer(
+        self._debounced_reload: Debouncer[Coroutine[Any, Any, None]] = Debouncer(
             hass,
             LOGGER,
             cooldown=ENTRY_RELOAD_COOLDOWN,
@@ -636,7 +637,7 @@ class RpcDeviceWrapper(update_coordinator.DataUpdateCoordinator):
         self.entry = entry
         self.device = device
 
-        self._debounced_reload = Debouncer(
+        self._debounced_reload: Debouncer[Coroutine[Any, Any, None]] = Debouncer(
             hass,
             LOGGER,
             cooldown=ENTRY_RELOAD_COOLDOWN,

homeassistant/components/sonos/household_coordinator.py

@@ -4,6 +4,7 @@ from __future__ import annotations
 import asyncio
 from collections.abc import Callable, Coroutine
 import logging
+from typing import Any
 
 from soco import SoCo
 
@@ -35,7 +36,7 @@ class SonosHouseholdCoordinator:
     async def _async_setup(self) -> None:
         """Finish setup in async context."""
         self.cache_update_lock = asyncio.Lock()
-        self.async_poll = Debouncer(
+        self.async_poll = Debouncer[Coroutine[Any, Any, None]](
             self.hass,
             _LOGGER,
             cooldown=3,

homeassistant/components/usb/__init__.py

@@ -1,12 +1,13 @@
 """The USB Discovery integration."""
 from __future__ import annotations
 
+from collections.abc import Coroutine
 import dataclasses
 import fnmatch
 import logging
 import os
 import sys
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Any
 
 from serial.tools.list_ports import comports
 from serial.tools.list_ports_common import ListPortInfo
@@ -109,7 +110,7 @@ class USBDiscovery:
         self.usb = usb
         self.seen: set[tuple[str, ...]] = set()
         self.observer_active = False
-        self._request_debouncer: Debouncer | None = None
+        self._request_debouncer: Debouncer[Coroutine[Any, Any, None]] | None = None
 
     async def async_setup(self) -> None:
         """Set up USB Discovery."""

homeassistant/helpers/debounce.py

@@ -2,14 +2,16 @@
 from __future__ import annotations
 
 import asyncio
-from collections.abc import Awaitable, Callable
+from collections.abc import Callable
 from logging import Logger
-from typing import Any
+from typing import Generic, TypeVar
 
 from homeassistant.core import HassJob, HomeAssistant, callback
 
+_R_co = TypeVar("_R_co", covariant=True)
 
-class Debouncer:
+
+class Debouncer(Generic[_R_co]):
     """Class to rate limit calls to a specific command."""
 
     def __init__(
@@ -19,7 +21,7 @@ class Debouncer:
         *,
         cooldown: float,
         immediate: bool,
-        function: Callable[..., Awaitable[Any]] | None = None,
+        function: Callable[[], _R_co] | None = None,
     ) -> None:
         """Initialize debounce.
 
@@ -35,15 +37,17 @@ class Debouncer:
         self._timer_task: asyncio.TimerHandle | None = None
         self._execute_at_end_of_timer: bool = False
         self._execute_lock = asyncio.Lock()
-        self._job: HassJob | None = None if function is None else HassJob(function)
+        self._job: HassJob[[], _R_co] | None = (
+            None if function is None else HassJob(function)
+        )
 
     @property
-    def function(self) -> Callable[..., Awaitable[Any]] | None:
+    def function(self) -> Callable[[], _R_co] | None:
         """Return the function being wrapped by the Debouncer."""
         return self._function
 
     @function.setter
-    def function(self, function: Callable[..., Awaitable[Any]]) -> None:
+    def function(self, function: Callable[[], _R_co]) -> None:
         """Update the function being wrapped by the Debouncer."""
         self._function = function
         if self._job is None or function != self._job.target:

homeassistant/helpers/device_registry.py

@@ -2,6 +2,7 @@
 from __future__ import annotations
 
 from collections import OrderedDict
+from collections.abc import Coroutine
 import logging
 import time
 from typing import TYPE_CHECKING, Any, NamedTuple, cast
@@ -832,7 +833,7 @@ def async_setup_cleanup(hass: HomeAssistant, dev_reg: DeviceRegistry) -> None:
         ent_reg = entity_registry.async_get(hass)
         async_cleanup(hass, dev_reg, ent_reg)
 
-    debounced_cleanup = Debouncer(
+    debounced_cleanup: Debouncer[Coroutine[Any, Any, None]] = Debouncer(
         hass, _LOGGER, cooldown=CLEANUP_DELAY, immediate=False, function=cleanup
     )
 

homeassistant/helpers/update_coordinator.py

@@ -2,7 +2,7 @@
 from __future__ import annotations
 
 import asyncio
-from collections.abc import Awaitable, Callable, Generator
+from collections.abc import Awaitable, Callable, Coroutine, Generator
 from datetime import datetime, timedelta
 import logging
 from time import monotonic
@@ -44,7 +44,7 @@ class DataUpdateCoordinator(Generic[_T]):
         name: str,
         update_interval: timedelta | None = None,
         update_method: Callable[[], Awaitable[_T]] | None = None,
-        request_refresh_debouncer: Debouncer | None = None,
+        request_refresh_debouncer: Debouncer[Coroutine[Any, Any, None]] | None = None,
     ) -> None:
         """Initialize global data updater."""
         self.hass = hass

mypy.ini

@@ -57,6 +57,9 @@ disallow_any_generics = true
 [mypy-homeassistant.helpers.condition]
 disallow_any_generics = true
 
+[mypy-homeassistant.helpers.debounce]
+disallow_any_generics = true
+
 [mypy-homeassistant.helpers.discovery]
 disallow_any_generics = true