Add typing to pilight Throttle decorator (#75443)

This commit is contained in:
Marc Mueller 2022-07-19 18:33:53 +02:00 committed by GitHub
parent 6b60fb9541
commit 4b036cbad9
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23

View File

@ -1,11 +1,16 @@
"""Component to create an interface to a Pilight daemon.""" """Component to create an interface to a Pilight daemon."""
from __future__ import annotations
from collections.abc import Callable
from datetime import timedelta from datetime import timedelta
import functools import functools
import logging import logging
import socket import socket
import threading import threading
from typing import Any
from pilight import pilight from pilight import pilight
from typing_extensions import ParamSpec
import voluptuous as vol import voluptuous as vol
from homeassistant.const import ( 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.helpers.typing import ConfigType
from homeassistant.util import dt as dt_util from homeassistant.util import dt as dt_util
_P = ParamSpec("_P")
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
CONF_SEND_DELAY = "send_delay" CONF_SEND_DELAY = "send_delay"
@ -138,23 +145,23 @@ class CallRateDelayThrottle:
def __init__(self, hass, delay_seconds: float) -> None: def __init__(self, hass, delay_seconds: float) -> None:
"""Initialize the delay handler.""" """Initialize the delay handler."""
self._delay = timedelta(seconds=max(0.0, delay_seconds)) self._delay = timedelta(seconds=max(0.0, delay_seconds))
self._queue: list = [] self._queue: list[Callable[[Any], None]] = []
self._active = False self._active = False
self._lock = threading.Lock() self._lock = threading.Lock()
self._next_ts = dt_util.utcnow() self._next_ts = dt_util.utcnow()
self._schedule = functools.partial(track_point_in_utc_time, hass) 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.""" """Decorate to delay calls on a certain method."""
@functools.wraps(method) @functools.wraps(method)
def decorated(*args, **kwargs): def decorated(*args: _P.args, **kwargs: _P.kwargs) -> None:
"""Delay a call.""" """Delay a call."""
if self._delay.total_seconds() == 0.0: if self._delay.total_seconds() == 0.0:
method(*args, **kwargs) method(*args, **kwargs)
return return
def action(event): def action(event: Any) -> None:
"""Wrap an action that gets scheduled.""" """Wrap an action that gets scheduled."""
method(*args, **kwargs) method(*args, **kwargs)