Improve type hints in rfxtrx siren entity (#76459)
This commit is contained in:
parent
6eb1dbdb74
commit
7d0a4ee00a
1 changed files with 31 additions and 14 deletions
|
@ -1,6 +1,7 @@
|
|||
"""Support for RFXtrx sirens."""
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime
|
||||
from typing import Any
|
||||
|
||||
import RFXtrx as rfxtrxmod
|
||||
|
@ -26,7 +27,7 @@ SECURITY_PANIC_OFF = "End Panic"
|
|||
SECURITY_PANIC_ALL = {SECURITY_PANIC_ON, SECURITY_PANIC_OFF}
|
||||
|
||||
|
||||
def supported(event: rfxtrxmod.RFXtrxEvent):
|
||||
def supported(event: rfxtrxmod.RFXtrxEvent) -> bool:
|
||||
"""Return whether an event supports sirens."""
|
||||
device = event.device
|
||||
|
||||
|
@ -104,16 +105,16 @@ class RfxtrxOffDelayMixin(Entity):
|
|||
_timeout: CALLBACK_TYPE | None = None
|
||||
_off_delay: float | None = None
|
||||
|
||||
def _setup_timeout(self):
|
||||
def _setup_timeout(self) -> None:
|
||||
@callback
|
||||
def _done(_):
|
||||
def _done(_: datetime) -> None:
|
||||
self._timeout = None
|
||||
self.async_write_ha_state()
|
||||
|
||||
if self._off_delay:
|
||||
self._timeout = async_call_later(self.hass, self._off_delay, _done)
|
||||
|
||||
def _cancel_timeout(self):
|
||||
def _cancel_timeout(self) -> None:
|
||||
if self._timeout:
|
||||
self._timeout()
|
||||
self._timeout = None
|
||||
|
@ -125,7 +126,13 @@ class RfxtrxChime(RfxtrxCommandEntity, SirenEntity, RfxtrxOffDelayMixin):
|
|||
_attr_supported_features = SirenEntityFeature.TURN_ON | SirenEntityFeature.TONES
|
||||
_device: rfxtrxmod.ChimeDevice
|
||||
|
||||
def __init__(self, device, device_id, off_delay=None, event=None):
|
||||
def __init__(
|
||||
self,
|
||||
device: rfxtrxmod.RFXtrxDevice,
|
||||
device_id: DeviceTuple,
|
||||
off_delay: float | None = None,
|
||||
event: rfxtrxmod.RFXtrxEvent | None = None,
|
||||
) -> None:
|
||||
"""Initialize the entity."""
|
||||
super().__init__(device, device_id, event)
|
||||
self._attr_available_tones = list(self._device.COMMANDS.values())
|
||||
|
@ -133,11 +140,11 @@ class RfxtrxChime(RfxtrxCommandEntity, SirenEntity, RfxtrxOffDelayMixin):
|
|||
self._off_delay = off_delay
|
||||
|
||||
@property
|
||||
def is_on(self):
|
||||
def is_on(self) -> bool:
|
||||
"""Return true if device is on."""
|
||||
return self._timeout is not None
|
||||
|
||||
async def async_turn_on(self, **kwargs):
|
||||
async def async_turn_on(self, **kwargs: Any) -> None:
|
||||
"""Turn the device on."""
|
||||
self._cancel_timeout()
|
||||
|
||||
|
@ -152,7 +159,7 @@ class RfxtrxChime(RfxtrxCommandEntity, SirenEntity, RfxtrxOffDelayMixin):
|
|||
|
||||
self.async_write_ha_state()
|
||||
|
||||
def _apply_event(self, event: rfxtrxmod.ControlEvent):
|
||||
def _apply_event(self, event: rfxtrxmod.ControlEvent) -> None:
|
||||
"""Apply a received event."""
|
||||
super()._apply_event(event)
|
||||
|
||||
|
@ -162,7 +169,9 @@ class RfxtrxChime(RfxtrxCommandEntity, SirenEntity, RfxtrxOffDelayMixin):
|
|||
self._setup_timeout()
|
||||
|
||||
@callback
|
||||
def _handle_event(self, event, device_id):
|
||||
def _handle_event(
|
||||
self, event: rfxtrxmod.RFXtrxEvent, device_id: DeviceTuple
|
||||
) -> None:
|
||||
"""Check if event applies to me and update."""
|
||||
if self._event_applies(event, device_id):
|
||||
self._apply_event(event)
|
||||
|
@ -176,7 +185,13 @@ class RfxtrxSecurityPanic(RfxtrxCommandEntity, SirenEntity, RfxtrxOffDelayMixin)
|
|||
_attr_supported_features = SirenEntityFeature.TURN_ON | SirenEntityFeature.TURN_OFF
|
||||
_device: rfxtrxmod.SecurityDevice
|
||||
|
||||
def __init__(self, device, device_id, off_delay=None, event=None):
|
||||
def __init__(
|
||||
self,
|
||||
device: rfxtrxmod.RFXtrxDevice,
|
||||
device_id: DeviceTuple,
|
||||
off_delay: float | None = None,
|
||||
event: rfxtrxmod.RFXtrxEvent | None = None,
|
||||
) -> None:
|
||||
"""Initialize the entity."""
|
||||
super().__init__(device, device_id, event)
|
||||
self._on_value = get_first_key(self._device.STATUS, SECURITY_PANIC_ON)
|
||||
|
@ -184,11 +199,11 @@ class RfxtrxSecurityPanic(RfxtrxCommandEntity, SirenEntity, RfxtrxOffDelayMixin)
|
|||
self._off_delay = off_delay
|
||||
|
||||
@property
|
||||
def is_on(self):
|
||||
def is_on(self) -> bool:
|
||||
"""Return true if device is on."""
|
||||
return self._timeout is not None
|
||||
|
||||
async def async_turn_on(self, **kwargs: Any):
|
||||
async def async_turn_on(self, **kwargs: Any) -> None:
|
||||
"""Turn the device on."""
|
||||
self._cancel_timeout()
|
||||
|
||||
|
@ -206,7 +221,7 @@ class RfxtrxSecurityPanic(RfxtrxCommandEntity, SirenEntity, RfxtrxOffDelayMixin)
|
|||
|
||||
self.async_write_ha_state()
|
||||
|
||||
def _apply_event(self, event: rfxtrxmod.SensorEvent):
|
||||
def _apply_event(self, event: rfxtrxmod.SensorEvent) -> None:
|
||||
"""Apply a received event."""
|
||||
super()._apply_event(event)
|
||||
|
||||
|
@ -219,7 +234,9 @@ class RfxtrxSecurityPanic(RfxtrxCommandEntity, SirenEntity, RfxtrxOffDelayMixin)
|
|||
self._cancel_timeout()
|
||||
|
||||
@callback
|
||||
def _handle_event(self, event, device_id):
|
||||
def _handle_event(
|
||||
self, event: rfxtrxmod.RFXtrxEvent, device_id: DeviceTuple
|
||||
) -> None:
|
||||
"""Check if event applies to me and update."""
|
||||
if self._event_applies(event, device_id):
|
||||
self._apply_event(event)
|
||||
|
|
Loading…
Add table
Add a link
Reference in a new issue