entity.py

Go to the documentation of this file.

"""Module for SIA Base Entity."""
 
from __future__ import annotations
 
from abc import abstractmethod
from dataclasses import dataclass
import logging
 
from pysiaalarm import SIAEvent
 
from homeassistant.components.alarm_control_panel import AlarmControlPanelState
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_PORT
from homeassistant.core import CALLBACK_TYPE, State, callback
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity import EntityDescription
from homeassistant.helpers.event import async_call_later
from homeassistant.helpers.restore_state import RestoreEntity
from homeassistant.helpers.typing import StateType
 
from .const import (
    AVAILABILITY_EVENT_CODE,
    CONF_ACCOUNT,
    CONF_ACCOUNTS,
    CONF_PING_INTERVAL,
    DOMAIN,
    SIA_EVENT,
    SIA_HUB_ZONE,
)
from .utils import (
    get_attr_from_sia_event,
    get_unavailability_interval,
    get_unique_id_and_name,
)
 
_LOGGER = logging.getLogger(__name__)
 
 
@dataclass(frozen=True)
class SIARequiredKeysMixin:
    """Required keys for SIA entities."""
 
    code_consequences: dict[str, StateType | bool | AlarmControlPanelState]
 
 
@dataclass(frozen=True)
class SIAEntityDescription(EntityDescription, SIARequiredKeysMixin):
    """Entity Description for SIA entities."""
 
 
class SIABaseEntity(RestoreEntity):
    """Base class for SIA entities."""
 
    entity_description: SIAEntityDescription
 
    def __init__(
        self,
        entry: ConfigEntry,
        account: str,
        zone: int,
        entity_description: SIAEntityDescription,
    ) -> None:
        """Create SIABaseEntity object."""
        self.portport = entry.data[CONF_PORT]
        self.accountaccount = account
        self.zonezone = zone
        self.entity_descriptionentity_description = entity_description
 
        self.ping_interval: int = next(
            acc[CONF_PING_INTERVAL]
            for acc in entry.data[CONF_ACCOUNTS]
            if acc[CONF_ACCOUNT] == account
        )
        self._attr_unique_id, self._attr_name_attr_name = get_unique_id_and_name(
            entry.entry_id, entry.data[CONF_PORT], account, zone, entity_description.key
        )
        self._attr_device_info_attr_device_info = DeviceInfo(
            name=self._attr_name_attr_name,
            identifiers={(DOMAIN, self._attr_unique_id)},
            via_device=(DOMAIN, f"{entry.data[CONF_PORT]}_{account}"),
        )
 
        self._post_interval_update_cb_canceller_post_interval_update_cb_canceller: CALLBACK_TYPE | None = None
        self._attr_extra_state_attributes_attr_extra_state_attributes = {}
        self._attr_should_poll_attr_should_poll = False
 
    async def async_added_to_hass(self) -> None:
        """Run when entity about to be added to hass.
 
        Overridden from Entity.
 
        1. register the dispatcher and add the callback to on_remove
        2. get previous state from storage and pass to entity specific function
        3. if available: create availability cb
        """
        self.async_on_removeasync_on_remove(
            async_dispatcher_connect(
                self.hasshass,
                SIA_EVENT.format(self.portport, self.accountaccount),
                self.async_handle_eventasync_handle_event,
            )
        )
        self.handle_last_statehandle_last_state(await self.async_get_last_stateasync_get_last_state())
        if self._attr_available_attr_available:
            self.async_create_post_interval_update_cbasync_create_post_interval_update_cb()
 
    @abstractmethod
    def handle_last_state(self, last_state: State | None) -> None:
        """Handle the last state."""
 
    async def async_will_remove_from_hass(self) -> None:
        """Run when entity will be removed from hass.
 
        Overridden from Entity.
        """
        self._cancel_post_interval_update_cb_cancel_post_interval_update_cb()
 
    @callback
    def async_handle_event(self, sia_event: SIAEvent) -> None:
        """Listen to dispatcher events for this port and account and update state and attributes.
 
        If the event is for either the zone or the 0 zone (hub zone),
        then handle it further.
 
        If the event had a code that was relevant for the entity,
        then update the attributes.
        If the event had a code that was relevant or it was a availability event
        then update the availability and schedule the next unavailability check.
        """
        _LOGGER.debug("Received event: %s", sia_event)
        if (int(sia_event.ri) if sia_event.ri else 0) not in (self.zonezone, SIA_HUB_ZONE):
            return
 
        relevant_event = self.update_stateupdate_state(sia_event)
 
        if relevant_event:
            self._attr_extra_state_attributes_attr_extra_state_attributes.update(get_attr_from_sia_event(sia_event))
 
        if relevant_event or sia_event.code == AVAILABILITY_EVENT_CODE:
            self._attr_available_attr_available = True
            self._cancel_post_interval_update_cb_cancel_post_interval_update_cb()
            self.async_create_post_interval_update_cbasync_create_post_interval_update_cb()
 
        self.async_write_ha_stateasync_write_ha_state()
 
    @abstractmethod
    def update_state(self, sia_event: SIAEvent) -> bool:
        """Do the entity specific state updates.
 
        Return True if the event was relevant for this entity.
        """
 
    @callback
    def async_create_post_interval_update_cb(self) -> None:
        """Create a port interval update cb and store the callback."""
        self._post_interval_update_cb_canceller_post_interval_update_cb_canceller = async_call_later(
            self.hasshass,
            get_unavailability_interval(self.ping_interval),
            self.async_post_interval_updateasync_post_interval_update,
        )
 
    @callback
    def async_post_interval_update(self, _) -> None:
        """Set unavailable after a ping interval."""
        self._attr_available_attr_available = False
        self.async_write_ha_stateasync_write_ha_state()
 
    @callback
    def _cancel_post_interval_update_cb(self) -> None:
        """Cancel the callback."""
        if self._post_interval_update_cb_canceller_post_interval_update_cb_canceller:
            self._post_interval_update_cb_canceller_post_interval_update_cb_canceller()
            self._post_interval_update_cb_canceller_post_interval_update_cb_canceller = None