root
239 lines
10 KiB
Python
239 lines
10 KiB
Python
import logging
|
|
from typing import Optional, Dict
|
|
|
|
from homeassistant.core import HomeAssistant
|
|
from meross_iot.controller.device import BaseDevice
|
|
from meross_iot.controller.mixins.light import LightMixin
|
|
from meross_iot.controller.mixins.diffuser_light import DiffuserLightMixin
|
|
from meross_iot.manager import MerossManager
|
|
from meross_iot.model.http.device import HttpDeviceInfo
|
|
from meross_iot.model.enums import DiffuserLightMode
|
|
import homeassistant.util.color as color_util
|
|
from homeassistant.components.light import LightEntity
|
|
from homeassistant.components.light import ColorMode, \
|
|
ATTR_HS_COLOR, ATTR_COLOR_TEMP, ATTR_BRIGHTNESS, ATTR_RGB_COLOR
|
|
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
|
|
from . import MerossDevice
|
|
from .common import (DOMAIN, MANAGER, HA_LIGHT, DEVICE_LIST_COORDINATOR)
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
|
|
class MerossOilDiffuserLightDevice(DiffuserLightMixin, BaseDevice):
|
|
"""
|
|
Type hints helper
|
|
"""
|
|
pass
|
|
|
|
|
|
class MerossLightDevice(LightMixin, BaseDevice):
|
|
"""
|
|
Type hints helper
|
|
"""
|
|
pass
|
|
|
|
|
|
class DiffuserLightEntityWrapper(MerossDevice, LightEntity):
|
|
"""Wrapper class to adapt the Meross OilDiffuserLight"""
|
|
_device: MerossOilDiffuserLightDevice
|
|
# For now, we assume OilDiffuserLight supports all the following features.
|
|
# From Meross API it is in fact impossible to determine which exact features are supported by the device.
|
|
_attr_supported_color_modes = {ColorMode.WHITE, ColorMode.RGB, ColorMode.COLOR_TEMP}
|
|
|
|
def __init__(self,
|
|
channel: int,
|
|
device: MerossOilDiffuserLightDevice,
|
|
device_list_coordinator: DataUpdateCoordinator[Dict[str, HttpDeviceInfo]]):
|
|
|
|
super().__init__(
|
|
device=device,
|
|
channel=channel,
|
|
device_list_coordinator=device_list_coordinator,
|
|
platform=HA_LIGHT)
|
|
|
|
async def async_turn_off(self, **kwargs) -> None:
|
|
await self._device.async_turn_off(channel=self._channel_id, skip_rate_limits=True)
|
|
|
|
async def async_turn_on(self, **kwargs) -> None:
|
|
if not self.is_on:
|
|
await self._device.async_turn_on(channel=self._channel_id, skip_rate_limits=True)
|
|
|
|
if ATTR_HS_COLOR in kwargs:
|
|
h, s = kwargs[ATTR_HS_COLOR]
|
|
rgb = color_util.color_hsv_to_RGB(h, s, 100)
|
|
_LOGGER.debug("color change: rgb=%r -- h=%r s=%r" % (rgb, h, s))
|
|
await self._device.async_set_light_mode(channel=self._channel_id, mode=DiffuserLightMode.FIXED_RGB, rgb=rgb, onoff=True, skip_rate_limits=True)
|
|
elif ATTR_COLOR_TEMP in kwargs:
|
|
mired = kwargs[ATTR_COLOR_TEMP]
|
|
norm_value = (mired - self.min_mireds) / (self.max_mireds - self.min_mireds)
|
|
temperature = 100 - (norm_value * 100)
|
|
_LOGGER.debug("temperature change: mired=%r meross=%r" % (mired, temperature))
|
|
await self._device.async_set_light_mode(channel=self._channel_id, mode=DiffuserLightMode.FIXED_LUMINANCE, onoff=True, rgb=65293, brightness=temperature, skip_rate_limits=True)
|
|
|
|
# Brightness must always be set, so take previous luminance if not explicitly set now.
|
|
if ATTR_BRIGHTNESS in kwargs:
|
|
brightness = kwargs[ATTR_BRIGHTNESS] * 100 / 255
|
|
_LOGGER.debug("brightness change: %r" % brightness)
|
|
await self._device.async_set_light_mode(channel=self._channel_id, luminance=brightness, skip_rate_limits=True)
|
|
|
|
@property
|
|
def is_on(self) -> Optional[bool]:
|
|
return self._device.get_light_is_on(channel=self._channel_id)
|
|
|
|
@property
|
|
def hs_color(self):
|
|
rgb = self._device.get_light_rgb_color(channel=self._channel_id)
|
|
if rgb is not None and isinstance(rgb, tuple) and len(rgb) == 3:
|
|
return color_util.color_RGB_to_hs(*rgb)
|
|
else:
|
|
return None # Return None if RGB value is not available
|
|
|
|
@property
|
|
def brightness(self):
|
|
luminance = self._device.get_light_brightness()
|
|
if luminance is not None:
|
|
return float(luminance) / 100 * 255
|
|
return None
|
|
|
|
|
|
class LightEntityWrapper(MerossDevice, LightEntity):
|
|
"""Wrapper class to adapt the Meross bulbs into the Homeassistant platform"""
|
|
_device: MerossLightDevice
|
|
|
|
def __init__(self,
|
|
channel: int,
|
|
device: MerossLightDevice,
|
|
device_list_coordinator: DataUpdateCoordinator[Dict[str, HttpDeviceInfo]]):
|
|
|
|
super().__init__(
|
|
device=device,
|
|
channel=channel,
|
|
device_list_coordinator=device_list_coordinator,
|
|
platform=HA_LIGHT)
|
|
|
|
async def async_turn_off(self, **kwargs) -> None:
|
|
await self._device.async_turn_off(channel=self._channel_id, skip_rate_limits=True)
|
|
|
|
async def async_turn_on(self, **kwargs) -> None:
|
|
if not self.is_on:
|
|
await self._device.async_turn_on(channel=self._channel_id, skip_rate_limits=True)
|
|
|
|
# Color is taken from either of these 2 values, but not both.
|
|
if ATTR_RGB_COLOR in kwargs:
|
|
await self._device.async_set_light_color(channel=self._channel_id, rgb=kwargs[ATTR_RGB_COLOR], onoff=True, skip_rate_limits=True)
|
|
elif ATTR_HS_COLOR in kwargs:
|
|
h, s = kwargs[ATTR_HS_COLOR]
|
|
rgb = color_util.color_hsv_to_RGB(h, s, 100)
|
|
_LOGGER.debug("color change: rgb=%r -- h=%r s=%r" % (rgb, h, s))
|
|
await self._device.async_set_light_color(channel=self._channel_id, rgb=rgb, onoff=True, skip_rate_limits=True)
|
|
elif ATTR_COLOR_TEMP in kwargs:
|
|
mired = kwargs[ATTR_COLOR_TEMP]
|
|
norm_value = (mired - self.min_mireds) / (self.max_mireds - self.min_mireds)
|
|
temperature = 100 - (norm_value * 100)
|
|
_LOGGER.debug("temperature change: mired=%r meross=%r" % (mired, temperature))
|
|
await self._device.async_set_light_color(channel=self._channel_id, temperature=temperature, skip_rate_limits=True)
|
|
|
|
# Brightness must always be set, so take previous luminance if not explicitly set now.
|
|
if ATTR_BRIGHTNESS in kwargs:
|
|
brightness = kwargs[ATTR_BRIGHTNESS] * 100 / 255
|
|
_LOGGER.debug("brightness change: %r" % brightness)
|
|
await self._device.async_set_light_color(channel=self._channel_id, luminance=brightness, skip_rate_limits=True)
|
|
|
|
@property
|
|
def supported_color_modes(self) -> set[ColorMode] | set[str] | None:
|
|
res = set()
|
|
if self._device.get_supports_luminance(channel=self._channel_id):
|
|
res.add(ColorMode.WHITE)
|
|
if self._device.get_supports_rgb(channel=self._channel_id):
|
|
res.add(ColorMode.RGB)
|
|
if self._device.get_supports_temperature(channel=self._channel_id):
|
|
res.add(ColorMode.COLOR_TEMP)
|
|
if len(res) < 1:
|
|
res.add(ColorMode.ONOFF)
|
|
return res
|
|
|
|
@property
|
|
def is_on(self) -> Optional[bool]:
|
|
return self._device.get_light_is_on(channel=self._channel_id)
|
|
|
|
@property
|
|
def brightness(self):
|
|
if not self._device.get_supports_luminance(self._channel_id):
|
|
return None
|
|
|
|
luminance = self._device.get_luminance()
|
|
if luminance is not None:
|
|
return float(luminance) / 100 * 255
|
|
|
|
return None
|
|
|
|
@property
|
|
def color_mode(self) -> ColorMode | str | None:
|
|
"""Return the color mode of the light."""
|
|
# TODO: we need support from low-level library in order to keep track of mode that haas been set.
|
|
if self._device.get_supports_rgb(channel=self._channel_id):
|
|
return ColorMode.RGB
|
|
elif self._device.get_supports_luminance(channel=self._channel_id):
|
|
return ColorMode.WHITE
|
|
if self._device.get_supports_temperature(channel=self._channel_id):
|
|
return ColorMode.COLOR_TEMP
|
|
return ColorMode.ONOFF
|
|
|
|
@property
|
|
def hs_color(self):
|
|
rgb = self._device.get_rgb_color(channel=self._channel_id)
|
|
if rgb is not None and isinstance(rgb, tuple) and len(rgb) == 3:
|
|
return color_util.color_RGB_to_hs(*rgb)
|
|
else:
|
|
return None # Return None if RGB value is not available
|
|
|
|
@property
|
|
def color_temp(self):
|
|
if self._device.get_supports_temperature(channel=self._channel_id):
|
|
value = self._device.get_color_temperature()
|
|
norm_value = (100 - value) / 100.0
|
|
return self.min_mireds + (norm_value * (self.max_mireds - self.min_mireds))
|
|
return None
|
|
|
|
|
|
async def async_setup_entry(hass: HomeAssistant, config_entry, async_add_entities):
|
|
def entity_adder_callback():
|
|
"""Discover and adds new Meross entities"""
|
|
manager: MerossManager = hass.data[DOMAIN][MANAGER] # type
|
|
coordinator = hass.data[DOMAIN][DEVICE_LIST_COORDINATOR]
|
|
devices = manager.find_devices()
|
|
|
|
new_entities = []
|
|
|
|
light_devs = filter(lambda d: isinstance(d, LightMixin), devices)
|
|
for d in light_devs:
|
|
channels = [c.index for c in d.channels] if len(d.channels) > 0 else [0]
|
|
for channel_index in channels:
|
|
w = LightEntityWrapper(device=d, channel=channel_index, device_list_coordinator=coordinator)
|
|
if w.unique_id not in hass.data[DOMAIN]["ADDED_ENTITIES_IDS"]:
|
|
new_entities.append(w)
|
|
|
|
diffuser_devs = filter(lambda d: isinstance(d, DiffuserLightMixin), devices)
|
|
for d in diffuser_devs:
|
|
channels = [c.index for c in d.channels] if len(d.channels) > 0 else [0]
|
|
for channel_index in channels:
|
|
w = DiffuserLightEntityWrapper(device=d, channel=channel_index, device_list_coordinator=coordinator)
|
|
if w.unique_id not in hass.data[DOMAIN]["ADDED_ENTITIES_IDS"]:
|
|
new_entities.append(w)
|
|
|
|
async_add_entities(new_entities, True)
|
|
|
|
coordinator = hass.data[DOMAIN][DEVICE_LIST_COORDINATOR]
|
|
coordinator.async_add_listener(entity_adder_callback)
|
|
# Run the entity adder a first time during setup
|
|
entity_adder_callback()
|
|
|
|
|
|
# TODO: Implement entry unload
|
|
# TODO: Unload entry
|
|
# TODO: Remove entry
|
|
|
|
|
|
def setup_platform(hass, config, async_add_entities, discovery_info=None):
|
|
pass
|