entity.py

Go to the documentation of this file.

"""Generic Hue Entity Model."""
 
from __future__ import annotations
 
from typing import TYPE_CHECKING
 
from aiohue.v2.controllers.base import BaseResourcesController
from aiohue.v2.controllers.events import EventType
from aiohue.v2.models.resource import ResourceTypes
from aiohue.v2.models.zigbee_connectivity import ConnectivityServiceStatus
 
from homeassistant.core import callback
from homeassistant.helpers import entity_registry as er
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity import Entity
 
from ..bridge import HueBridge
from ..const import CONF_IGNORE_AVAILABILITY, DOMAIN
 
if TYPE_CHECKING:
    from aiohue.v2.models.device_power import DevicePower
    from aiohue.v2.models.grouped_light import GroupedLight
    from aiohue.v2.models.light import Light
    from aiohue.v2.models.light_level import LightLevel
    from aiohue.v2.models.motion import Motion
 
    type HueResource = Light | DevicePower | GroupedLight | LightLevel | Motion
 
 
RESOURCE_TYPE_NAMES = {
    # a simple mapping of hue resource type to Hass name
    ResourceTypes.LIGHT_LEVEL: "Illuminance",
    ResourceTypes.DEVICE_POWER: "Battery",
}
 
 
class HueBaseEntity(Entity):  # pylint: disable=hass-enforce-class-module
    """Generic Entity Class for a Hue resource."""
 
    _attr_should_poll = False
 
    def __init__(
        self,
        bridge: HueBridge,
        controller: BaseResourcesController,
        resource: HueResource,
    ) -> None:
        """Initialize a generic Hue resource entity."""
        self.bridgebridge = bridge
        self.controllercontroller = controller
        self.resourceresource = resource
        self.devicedevice = controller.get_device(resource.id)
        self.loggerlogger = bridge.logger.getChild(resource.type.value)
 
        # Entity class attributes
        self._attr_unique_id_attr_unique_id = resource.id
        # device is precreated in main handler
        # this attaches the entity to the precreated device
        if self.devicedevice is None:
            # attach all device-less entities to the bridge itself
            # e.g. config based sensors like entertainment area
            self._attr_device_info_attr_device_info = DeviceInfo(
                identifiers={(DOMAIN, bridge.api.config.bridge.bridge_id)},
            )
        else:
            self._attr_device_info_attr_device_info = DeviceInfo(
                identifiers={(DOMAIN, self.devicedevice.id)},
            )
        # used for availability workaround
        self._ignore_availability_ignore_availability = None
        self._last_state_last_state = None
 
    async def async_added_to_hass(self) -> None:
        """Call when entity is added."""
        self._check_availability_check_availability()
        # Add value_changed callbacks.
        self.async_on_removeasync_on_remove(
            self.controllercontroller.subscribe(
                self._handle_event_handle_event,
                self.resourceresource.id,
                (EventType.RESOURCE_UPDATED, EventType.RESOURCE_DELETED),
            )
        )
        # also subscribe to device update event to catch device changes (e.g. name)
        if self.devicedevice is None:
            return
        self.async_on_removeasync_on_remove(
            self.bridgebridge.api.devices.subscribe(
                self._handle_event_handle_event,
                self.devicedevice.id,
                EventType.RESOURCE_UPDATED,
            )
        )
        # subscribe to zigbee_connectivity to catch availability changes
        if zigbee := self.bridgebridge.api.devices.get_zigbee_connectivity(self.devicedevice.id):
            self.async_on_removeasync_on_remove(
                self.bridgebridge.api.sensors.zigbee_connectivity.subscribe(
                    self._handle_event_handle_event,
                    zigbee.id,
                    EventType.RESOURCE_UPDATED,
                )
            )
 
    @property
    def available(self) -> bool:
        """Return entity availability."""
        # entities without a device attached should be always available
        if self.devicedevice is None:
            return True
        # the zigbee connectivity sensor itself should be always available
        if self.resourceresource.type == ResourceTypes.ZIGBEE_CONNECTIVITY:
            return True
        if self._ignore_availability_ignore_availability:
            return True
        # all device-attached entities get availability from the zigbee connectivity
        if zigbee := self.bridgebridge.api.devices.get_zigbee_connectivity(self.devicedevice.id):
            return zigbee.status == ConnectivityServiceStatus.CONNECTED
        return True
 
    @callback
    def on_update(self) -> None:
        """Call on update event."""
        # used in subclasses
 
    @callback
    def _handle_event(self, event_type: EventType, resource: HueResource) -> None:
        """Handle status event for this resource (or it's parent)."""
        if event_type == EventType.RESOURCE_DELETED:
            # cleanup entities that are not strictly device-bound and have the bridge as parent
            if self.devicedevice is None and resource.id == self.resourceresource.id:
                ent_reg = er.async_get(self.hasshass)
                ent_reg.async_remove(self.entity_identity_id)
            return
 
        self.loggerlogger.debug("Received status update for %s", self.entity_identity_id)
        self._check_availability_check_availability()
        self.on_updateon_update()
        self.async_write_ha_stateasync_write_ha_state()
 
    @callback
    def _check_availability(self):
        """Check availability of the device."""
        # return if we already processed this entity
        if self._ignore_availability_ignore_availability is not None:
            return
        # only do the availability check for entities connected to a device (with `on` feature)
        if self.devicedevice is None or not hasattr(self.resourceresource, "on"):
            self._ignore_availability_ignore_availability = False
            return
        # ignore availability if user added device to ignore list
        if self.devicedevice.id in self.bridgebridge.config_entry.options.get(
            CONF_IGNORE_AVAILABILITY, []
        ):
            self._ignore_availability_ignore_availability = True
            self.loggerlogger.info(
                "Device %s is configured to ignore availability status. ",
                self.namename,
            )
            return
        # certified products (normally) report their state correctly
        # no need for workaround/reporting
        if self.devicedevice.product_data.certified:
            self._ignore_availability_ignore_availability = False
            return
        # some (3th party) Hue lights report their connection status incorrectly
        # causing the zigbee availability to report as disconnected while in fact
        # it can be controlled. If the light is reported unavailable
        # by the zigbee connectivity but the state changes its considered as a
        # malfunctioning device and we report it.
        # While the user should actually fix this issue, we allow to
        # ignore the availability for this light/device from the config options.
        cur_state = self.resourceresource.on.on
        if self._last_state_last_state is None:
            self._last_state_last_state = cur_state
            return
        if zigbee := self.bridgebridge.api.devices.get_zigbee_connectivity(self.devicedevice.id):
            if (
                self._last_state_last_state != cur_state
                and zigbee.status != ConnectivityServiceStatus.CONNECTED
            ):
                # the device state changed from on->off or off->on
                # while it was reported as not connected!
                self.loggerlogger.warning(
                    (
                        "Device %s changed state while reported as disconnected. This"
                        " might be an indicator that routing is not working for this"
                        " device or the device is having connectivity issues. You can"
                        " disable availability reporting for this device in the Hue"
                        " options. Device details: %s - %s (%s) fw: %s"
                    ),
                    self.namename,
                    self.devicedevice.product_data.manufacturer_name,
                    self.devicedevice.product_data.product_name,
                    self.devicedevice.product_data.model_id,
                    self.devicedevice.product_data.software_version,
                )
                # set attribute to false because we only want to log once per light/device.
                # a user must opt-in to ignore availability through integration options
                self._ignore_availability_ignore_availability = False
        self._last_state_last_state = cur_state