scene.py

"""Support for scene platform for Hue scenes (V2 only)."""

from __future__ import annotations

from typing import Any

from aiohue.v2 import HueBridgeV2
from aiohue.v2.controllers.events import EventType
from aiohue.v2.controllers.scenes import ScenesController
from aiohue.v2.models.scene import Scene as HueScene, ScenePut as HueScenePut
from aiohue.v2.models.smart_scene import SmartScene as HueSmartScene, SmartSceneState
import voluptuous as vol

from homeassistant.components.scene import ATTR_TRANSITION, Scene as SceneEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity_platform import (
    AddEntitiesCallback,
    async_get_current_platform,
)

from .bridge import HueBridge
from .const import DOMAIN
from .v2.entity import HueBaseEntity
from .v2.helpers import normalize_hue_brightness, normalize_hue_transition

SERVICE_ACTIVATE_SCENE = "activate_scene"
ATTR_DYNAMIC = "dynamic"
ATTR_SPEED = "speed"
ATTR_BRIGHTNESS = "brightness"


async def async_setup_entry(
    hass: HomeAssistant,
    config_entry: ConfigEntry,
    async_add_entities: AddEntitiesCallback,
) -> None:
    """Set up scene platform from Hue group scenes."""
    bridge: HueBridge = hass.data[DOMAIN][config_entry.entry_id]
    api: HueBridgeV2 = bridge.api

    if bridge.api_version == 1:
        # should not happen, but just in case
        raise NotImplementedError("Scene support is only available for V2 bridges")

    # add entities for all scenes
    @callback
    def async_add_entity(
        event_type: EventType, resource: HueScene | HueSmartScene
    ) -> None:
        """Add entity from Hue resource."""
        if isinstance(resource, HueSmartScene):
            async_add_entities([HueSmartSceneEntity(bridge, api.scenes, resource)])
        else:
            async_add_entities([HueSceneEntity(bridge, api.scenes, resource)])

    # add all current items in controller
    for item in api.scenes:
        async_add_entity(EventType.RESOURCE_ADDED, item)

    # register listener for new items only
    config_entry.async_on_unload(
        api.scenes.subscribe(async_add_entity, event_filter=EventType.RESOURCE_ADDED)
    )

    # add platform service to turn_on/activate scene with advanced options
    platform = async_get_current_platform()
    platform.async_register_entity_service(
        SERVICE_ACTIVATE_SCENE,
        {
            vol.Optional(ATTR_DYNAMIC): vol.Coerce(bool),
            vol.Optional(ATTR_SPEED): vol.All(
                vol.Coerce(int), vol.Range(min=0, max=100)
            ),
            vol.Optional(ATTR_TRANSITION): vol.All(
                vol.Coerce(float), vol.Range(min=0, max=3600)
            ),
            vol.Optional(ATTR_BRIGHTNESS): vol.All(
                vol.Coerce(int), vol.Range(min=1, max=255)
            ),
        },
        "_async_activate",
    )


class HueSceneEntityBase(HueBaseEntity, SceneEntity):
    """Base Representation of a Scene entity from Hue Scenes."""

    _attr_has_entity_name = True

    def __init__(
        self,
        bridge: HueBridge,
        controller: ScenesController,
        resource: HueScene | HueSmartScene,
    ) -> None:
        """Initialize the entity."""
        super().__init__(bridge, controller, resource)
        self.resource = resource
        self.controller = controller
        self.group = self.controller.get_group(self.resource.id)
        # we create a virtual service/device for Hue zones/rooms
        # so we have a parent for grouped lights and scenes
        self._attr_device_info = DeviceInfo(
            identifiers={(DOMAIN, self.group.id)},
        )

    async def async_added_to_hass(self) -> None:
        """Call when entity is added."""
        await super().async_added_to_hass()
        # Add value_changed callback for group to catch name changes.
        self.async_on_remove(
            self.bridge.api.groups.subscribe(
                self._handle_event,
                self.group.id,
                (EventType.RESOURCE_UPDATED),
            )
        )

    @property
    def name(self) -> str:
        """Return name of the scene."""
        return self.resource.metadata.name


class HueSceneEntity(HueSceneEntityBase):
    """Representation of a Scene entity from Hue Scenes."""

    @property
    def is_dynamic(self) -> bool:
        """Return if this scene has a dynamic color palette."""
        if self.resource.palette.color and len(self.resource.palette.color) > 1:
            return True
        if (
            self.resource.palette.color_temperature
            and len(self.resource.palette.color_temperature) > 1
        ):
            return True
        return False

    async def async_activate(self, **kwargs: Any) -> None:
        """Activate Hue scene."""
        transition = normalize_hue_transition(kwargs.get(ATTR_TRANSITION))
        # the options below are advanced only
        # as we're not allowed to override the default scene turn_on service
        # we've implemented a `activate_scene` entity service
        dynamic = kwargs.get(ATTR_DYNAMIC, False)
        speed = kwargs.get(ATTR_SPEED)
        brightness = normalize_hue_brightness(kwargs.get(ATTR_BRIGHTNESS))

        if speed is not None:
            await self.bridge.async_request_call(
                self.controller.scene.update,
                self.resource.id,
                HueScenePut(speed=speed / 100),
            )

        await self.bridge.async_request_call(
            self.controller.scene.recall,
            self.resource.id,
            dynamic=dynamic,
            duration=transition,
            brightness=brightness,
        )

    @property
    def extra_state_attributes(self) -> dict[str, Any] | None:
        """Return the optional state attributes."""
        brightness = None
        if palette := self.resource.palette:
            if palette.dimming:
                brightness = palette.dimming[0].brightness
        if brightness is None:
            # get brightness from actions
            for action in self.resource.actions:
                if action.action.dimming:
                    brightness = action.action.dimming.brightness
                    break
        if brightness is not None:
            # Hue uses a range of [0, 100] to control brightness.
            brightness = round((brightness / 100) * 255)
        return {
            "group_name": self.group.metadata.name,
            "group_type": self.group.type.value,
            "name": self.resource.metadata.name,
            "speed": self.resource.speed,
            "brightness": brightness,
            "is_dynamic": self.is_dynamic,
        }


class HueSmartSceneEntity(HueSceneEntityBase):
    """Representation of a Smart Scene entity from Hue Scenes."""

    @property
    def is_active(self) -> bool:
        """Return if this smart scene is currently active."""
        return self.resource.state == SmartSceneState.ACTIVE

    async def async_activate(self, **kwargs: Any) -> None:
        """Activate Hue Smart scene."""

        await self.bridge.async_request_call(
            self.controller.smart_scene.recall,
            self.resource.id,
        )

    @property
    def extra_state_attributes(self) -> dict[str, Any] | None:
        """Return the optional state attributes."""
        res = {
            "group_name": self.group.metadata.name,
            "group_type": self.group.type.value,
            "name": self.resource.metadata.name,
            "is_active": self.is_active,
        }
        if self.is_active and self.resource.active_timeslot:
            res["active_timeslot_id"] = self.resource.active_timeslot.timeslot_id
            res["active_timeslot_name"] = self.resource.active_timeslot.weekday.value
            # lookup active scene in timeslot
            active_scene = None
            count = 0
            for day_timeslot in self.resource.week_timeslots:
                for timeslot in day_timeslot.timeslots:
                    if count != self.resource.active_timeslot.timeslot_id:
                        count += 1
                        continue
                    active_scene = self.controller.get(timeslot.target.rid)
                    break
            if active_scene is not None:
                res["active_scene"] = active_scene.metadata.name
        return res