diff --git a/.strict-typing b/.strict-typing index aa911dd81d9..f133199a753 100644 --- a/.strict-typing +++ b/.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 diff --git a/homeassistant/components/flux_led/number.py b/homeassistant/components/flux_led/number.py index 65c8a955dcf..18237c97e94 100644 --- a/homeassistant/components/flux_led/number.py +++ b/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: diff --git a/homeassistant/components/plex/__init__.py b/homeassistant/components/plex/__init__.py index 148e6a906bd..9ff8bcf7b54 100644 --- a/homeassistant/components/plex/__init__.py +++ b/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, diff --git a/homeassistant/components/samsungtv/__init__.py b/homeassistant/components/samsungtv/__init__.py index b870aab62d4..0a7fd4b3378 100644 --- a/homeassistant/components/samsungtv/__init__.py +++ b/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, diff --git a/homeassistant/components/shelly/__init__.py b/homeassistant/components/shelly/__init__.py index 012f692c579..125e63449ef 100644 --- a/homeassistant/components/shelly/__init__.py +++ b/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, diff --git a/homeassistant/components/sonos/household_coordinator.py b/homeassistant/components/sonos/household_coordinator.py index 0d76feae461..51d7e9cec8c 100644 --- a/homeassistant/components/sonos/household_coordinator.py +++ b/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, diff --git a/homeassistant/components/usb/__init__.py b/homeassistant/components/usb/__init__.py index 7a56a659d07..5783401df13 100644 --- a/homeassistant/components/usb/__init__.py +++ b/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.""" diff --git a/homeassistant/helpers/debounce.py b/homeassistant/helpers/debounce.py index 7937459b50c..2fbdefd7ec0 100644 --- a/homeassistant/helpers/debounce.py +++ b/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: diff --git a/homeassistant/helpers/device_registry.py b/homeassistant/helpers/device_registry.py index ca5e6e1aefa..a19f476495a 100644 --- a/homeassistant/helpers/device_registry.py +++ b/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 ) diff --git a/homeassistant/helpers/update_coordinator.py b/homeassistant/helpers/update_coordinator.py index 30da847642b..768b8040729 100644 --- a/homeassistant/helpers/update_coordinator.py +++ b/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 diff --git a/mypy.ini b/mypy.ini index 2333c20c4d8..3355680069e 100644 --- a/mypy.ini +++ b/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