From 4b036cbad91c4e6a8eba6952af4a01d39a22a4e2 Mon Sep 17 00:00:00 2001 From: Marc Mueller <30130371+cdce8p@users.noreply.github.com> Date: Tue, 19 Jul 2022 18:33:53 +0200 Subject: [PATCH] Add typing to pilight Throttle decorator (#75443) --- homeassistant/components/pilight/__init__.py | 15 +++++++++++---- 1 file changed, 11 insertions(+), 4 deletions(-) diff --git a/homeassistant/components/pilight/__init__.py b/homeassistant/components/pilight/__init__.py index 38d7a3e18ea..0386d267f04 100644 --- a/homeassistant/components/pilight/__init__.py +++ b/homeassistant/components/pilight/__init__.py @@ -1,11 +1,16 @@ """Component to create an interface to a Pilight daemon.""" +from __future__ import annotations + +from collections.abc import Callable from datetime import timedelta import functools import logging import socket import threading +from typing import Any from pilight import pilight +from typing_extensions import ParamSpec import voluptuous as vol from homeassistant.const import ( @@ -22,6 +27,8 @@ from homeassistant.helpers.event import track_point_in_utc_time from homeassistant.helpers.typing import ConfigType from homeassistant.util import dt as dt_util +_P = ParamSpec("_P") + _LOGGER = logging.getLogger(__name__) CONF_SEND_DELAY = "send_delay" @@ -138,23 +145,23 @@ class CallRateDelayThrottle: def __init__(self, hass, delay_seconds: float) -> None: """Initialize the delay handler.""" self._delay = timedelta(seconds=max(0.0, delay_seconds)) - self._queue: list = [] + self._queue: list[Callable[[Any], None]] = [] self._active = False self._lock = threading.Lock() self._next_ts = dt_util.utcnow() self._schedule = functools.partial(track_point_in_utc_time, hass) - def limited(self, method): + def limited(self, method: Callable[_P, Any]) -> Callable[_P, None]: """Decorate to delay calls on a certain method.""" @functools.wraps(method) - def decorated(*args, **kwargs): + def decorated(*args: _P.args, **kwargs: _P.kwargs) -> None: """Delay a call.""" if self._delay.total_seconds() == 0.0: method(*args, **kwargs) return - def action(event): + def action(event: Any) -> None: """Wrap an action that gets scheduled.""" method(*args, **kwargs)