hass_odido_zyxel_5g/custom_components/msp_integration_101_intermediate/light.py

"""Light setup for our Integration."""

from datetime import timedelta
import logging
from typing import Any

import voluptuous as vol

from homeassistant.components.light import ATTR_BRIGHTNESS, ColorMode, LightEntity
from homeassistant.core import HomeAssistant
from homeassistant.helpers import config_validation as cv, entity_platform
from homeassistant.helpers.entity_platform import AddEntitiesCallback

from . import MyConfigEntry
from .base import ExampleBaseEntity
from .const import CONF_OFF_TIME, SET_OFF_TIMER_ENTITY_SERVICE_NAME
from .coordinator import ExampleCoordinator

_LOGGER = logging.getLogger(__name__)


async def async_setup_entry(
    hass: HomeAssistant,
    config_entry: MyConfigEntry,
    async_add_entities: AddEntitiesCallback,
):
    """Set up the Binary Sensors."""
    # This gets the data update coordinator from the config entry runtime data as specified in your __init__.py
    coordinator: ExampleCoordinator = config_entry.runtime_data.coordinator

    # ----------------------------------------------------------------------------
    # Here we are going to add some lights entities for the lights in our mock data.
    # We have an on/off light and a dimmable light in our mock data, so add each
    # specific class based on the light type.
    # ----------------------------------------------------------------------------
    lights = []

    # On/Off lights
    lights.extend(
        [
            ExampleOnOffLight(coordinator, device, "state")
            for device in coordinator.data
            if device.get("device_type") == "ON_OFF_LIGHT"
        ]
    )

    # Dimmable lights
    lights.extend(
        [
            ExampleDimmableLight(coordinator, device, "state")
            for device in coordinator.data
            if device.get("device_type") == "DIMMABLE_LIGHT"
        ]
    )

    # Create the lights.
    async_add_entities(lights)

    # ----------------------------------------------------------------------------
    # Add an Entity Service
    #
    # This example creates a service to set an off timer for our mock device.
    # You can have your schema as anything you need and must relate to the entry
    # in services.yaml.  Best to set your target filters in serivces.yaml to match
    # what this can be called on, so your users cannot pick inappropriate entities.
    # If an entity is supplied that does not support this service, nothing will
    # happen.
    #
    # The function async_set_off_timer has to be part of your entity class and
    # is shown below.
    #
    # You will see the off timer sensor on a light update to reflect the time you
    # set.
    # ----------------------------------------------------------------------------
    platform = entity_platform.async_get_current_platform()
    platform.async_register_entity_service(
        SET_OFF_TIMER_ENTITY_SERVICE_NAME,
        {
            vol.Required(CONF_OFF_TIME): cv.time_period,
        },
        "async_set_off_timer",
    )


class ExampleOnOffLight(ExampleBaseEntity, LightEntity):
    """Implementation of an on/off light.

    This inherits our ExampleBaseEntity to set common properties.
    See base.py for this class.

    https://developers.home-assistant.io/docs/core/entity/light/
    """

    _attr_supported_color_modes = {ColorMode.ONOFF}
    _attr_color_mode = ColorMode.ONOFF

    @property
    def is_on(self) -> bool | None:
        """Return if the binary sensor is on."""
        # This needs to enumerate to true or false
        return (
            self.coordinator.get_device_parameter(self.device_id, self.parameter)
            == "ON"
        )

    async def async_turn_on(self, **kwargs: Any) -> None:
        """Turn the entity on."""
        await self.hass.async_add_executor_job(
            self.coordinator.api.set_data, self.device_id, self.parameter, "ON"
        )
        # ----------------------------------------------------------------------------
        # Use async_refresh on the DataUpdateCoordinator to perform immediate update.
        # Using self.async_update or self.coordinator.async_request_refresh may delay update due
        # to trying to batch requests.
        # ----------------------------------------------------------------------------
        await self.coordinator.async_refresh()

    async def async_turn_off(self, **kwargs: Any) -> None:
        """Turn the entity off."""
        await self.hass.async_add_executor_job(
            self.coordinator.api.set_data, self.device_id, self.parameter, "OFF"
        )
        # ----------------------------------------------------------------------------
        # Use async_refresh on the DataUpdateCoordinator to perform immediate update.
        # Using self.async_update or self.coordinator.async_request_refresh may delay update due
        # to trying to batch requests.
        # ----------------------------------------------------------------------------
        await self.coordinator.async_refresh()

    async def async_set_off_timer(self, off_time: timedelta) -> None:
        """Handle the set off timer service call.

        Important here to have your service parameters included in your
        function as they are passed as named parameters.
        """
        await self.hass.async_add_executor_job(
            self.coordinator.api.set_data,
            self.device_id,
            "off_timer",
            ":".join(str(off_time).split(":")[:2]),
        )
        # We have made a change to our device, so call a refresh to get updated data.
        # We use async_request_refresh here to batch the updates in case you select
        # multiple entities.
        await self.coordinator.async_request_refresh()


class ExampleDimmableLight(ExampleOnOffLight):
    """Implementation of a dimmable light."""

    _attr_supported_color_modes = {ColorMode.BRIGHTNESS}
    _attr_color_mode = ColorMode.BRIGHTNESS

    @property
    def brightness(self) -> int:
        """Return the brightness of this light between 0..255."""
        # Our light is in range 0..100, so convert
        return int(
            self.coordinator.get_device_parameter(self.device_id, "brightness")
            * (255 / 100)
        )

    async def async_turn_on(self, **kwargs: Any) -> None:
        """Turn the entity on."""
        if ATTR_BRIGHTNESS in kwargs:
            brightness = int(kwargs[ATTR_BRIGHTNESS] * (100 / 255))
            await self.hass.async_add_executor_job(
                self.coordinator.api.set_data, self.device_id, "brightness", brightness
            )
        else:
            await self.hass.async_add_executor_job(
                self.coordinator.api.set_data, self.device_id, self.parameter, "ON"
            )
        # ----------------------------------------------------------------------------
        # Use async_refresh on the DataUpdateCoordinator to perform immediate update.
        # Using self.async_update or self.coordinator.async_request_refresh may delay update due
        # to trying to batch requests and cause wierd UI behaviour.
        # ----------------------------------------------------------------------------
        await self.coordinator.async_refresh()