coordinator.py

Go to the documentation of this file.

"""Data coordinator for receiving LD2410B updates."""
 
from datetime import datetime
import logging
import time
 
from ld2410_ble import LD2410BLE, LD2410BLEState
 
from homeassistant.core import CALLBACK_TYPE, HassJob, HomeAssistant, callback
from homeassistant.helpers.event import async_call_later
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
 
from .const import DOMAIN
 
_LOGGER = logging.getLogger(__name__)
 
NEVER_TIME = -86400.0
DEBOUNCE_SECONDS = 1.0
 
 
class LD2410BLECoordinator(DataUpdateCoordinator[None]):
    """Data coordinator for receiving LD2410B updates."""
 
    def __init__(self, hass: HomeAssistant, ld2410_ble: LD2410BLE) -> None:
        """Initialise the coordinator."""
        super().__init__(
            hass,
            _LOGGER,
            name=DOMAIN,
        )
        self._ld2410_ble_ld2410_ble = ld2410_ble
        ld2410_ble.register_callback(self._async_handle_update_async_handle_update)
        ld2410_ble.register_disconnected_callback(self._async_handle_disconnect_async_handle_disconnect)
        self.connectedconnected = False
        self._last_update_time_last_update_time = NEVER_TIME
        self._debounce_cancel_debounce_cancel: CALLBACK_TYPE | None = None
        self._debounced_update_job_debounced_update_job = HassJob(
            self._async_handle_debounced_update_async_handle_debounced_update,
            f"LD2410 {ld2410_ble.address} BLE debounced update",
        )
 
    @callback
    def _async_handle_debounced_update(self, _now: datetime) -> None:
        """Handle debounced update."""
        self._debounce_cancel_debounce_cancel = None
        self._last_update_time_last_update_time = time.monotonic()
        self.async_set_updated_dataasync_set_updated_data(None)
 
    @callback
    def _async_handle_update(self, state: LD2410BLEState) -> None:
        """Just trigger the callbacks."""
        self.connectedconnected = True
        previous_last_updated_time = self._last_update_time_last_update_time
        self._last_update_time_last_update_time = time.monotonic()
        if self._last_update_time_last_update_time - previous_last_updated_time >= DEBOUNCE_SECONDS:
            self.async_set_updated_dataasync_set_updated_data(None)
            return
        if self._debounce_cancel_debounce_cancel is None:
            self._debounce_cancel_debounce_cancel = async_call_later(
                self.hasshass, DEBOUNCE_SECONDS, self._debounced_update_job_debounced_update_job
            )
 
    @callback
    def _async_handle_disconnect(self) -> None:
        """Trigger the callbacks for disconnected."""
        self.connectedconnected = False
        self.async_update_listenersasync_update_listeners()
 
    async def async_shutdown(self) -> None:
        """Shutdown the coordinator."""
        if self._debounce_cancel_debounce_cancel is not None:
            self._debounce_cancel_debounce_cancel()
            self._debounce_cancel_debounce_cancel = None
        await super().async_shutdown()