Add support for Shelly CCT light (#126989)

* Initial support for cct lights

* Move properties to the RpcShellyCctLight class

* Fix entity names

* Add async_remove_orphaned_entities() function

* Do not return

* Fix tests

* Combine async_remove_orphaned_virtual_entities and async_remove_orphaned_entities

* Remove SHELLY_PLUS_RGBW_CHANNELS from const

* Add tests

* Use _attr*

* Check ColorMode.COLOR_TEMP

* Add sensors for CCT light

* Remove removal condition

* Remove orphaned sensors

* Cleaning

* Add device temperature sensor for CCT light

* Simplify async_remove_orphaned_entities()

* Comment

* Add COMPONENT_ID_PATTERN const

* Call async_add_entities once

* Suggested change

* Better type for keys

* Do not call keys()
This commit is contained in:
Maciej Bieniek 2024-10-04 08:49:35 +02:00 committed by GitHub
parent 49e634a62f
commit 1290f18ed4
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
10 changed files with 321 additions and 68 deletions

View File

@ -34,7 +34,7 @@ from .entity import (
async_setup_entry_rpc,
)
from .utils import (
async_remove_orphaned_virtual_entities,
async_remove_orphaned_entities,
get_device_entry_gen,
get_virtual_component_ids,
is_block_momentary_input,
@ -263,13 +263,13 @@ async def async_setup_entry(
virtual_binary_sensor_ids = get_virtual_component_ids(
coordinator.device.config, BINARY_SENSOR_PLATFORM
)
async_remove_orphaned_virtual_entities(
async_remove_orphaned_entities(
hass,
config_entry.entry_id,
coordinator.mac,
BINARY_SENSOR_PLATFORM,
"boolean",
virtual_binary_sensor_ids,
"boolean",
)
return

View File

@ -239,8 +239,6 @@ DEVICES_WITHOUT_FIRMWARE_CHANGELOG = (
CONF_GEN = "gen"
SHELLY_PLUS_RGBW_CHANNELS = 4
VIRTUAL_COMPONENTS_MAP = {
"binary_sensor": {"types": ["boolean"], "modes": ["label"]},
"number": {"types": ["number"], "modes": ["field", "slider"]},
@ -257,3 +255,5 @@ VIRTUAL_NUMBER_MODE_MAP = {
API_WS_URL = "/api/shelly/ws"
COMPONENT_ID_PATTERN = re.compile(r"[a-z\d]+:\d+")

View File

@ -34,14 +34,13 @@ from .const import (
RGBW_MODELS,
RPC_MIN_TRANSITION_TIME_SEC,
SHBLB_1_RGB_EFFECTS,
SHELLY_PLUS_RGBW_CHANNELS,
STANDARD_RGB_EFFECTS,
)
from .coordinator import ShellyBlockCoordinator, ShellyConfigEntry, ShellyRpcCoordinator
from .entity import ShellyBlockEntity, ShellyRpcEntity
from .utils import (
async_remove_orphaned_entities,
async_remove_shelly_entity,
async_remove_shelly_rpc_entities,
brightness_to_percentage,
get_device_entry_gen,
get_rpc_key_ids,
@ -119,30 +118,25 @@ def async_setup_rpc_entry(
)
return
entities: list[RpcShellyLightBase] = []
if light_key_ids := get_rpc_key_ids(coordinator.device.status, "light"):
# Light mode remove RGB & RGBW entities, add light entities
async_remove_shelly_rpc_entities(
hass, LIGHT_DOMAIN, coordinator.mac, ["rgb:0", "rgbw:0"]
)
async_add_entities(RpcShellyLight(coordinator, id_) for id_ in light_key_ids)
return
light_keys = [f"light:{i}" for i in range(SHELLY_PLUS_RGBW_CHANNELS)]
entities.extend(RpcShellyLight(coordinator, id_) for id_ in light_key_ids)
if cct_key_ids := get_rpc_key_ids(coordinator.device.status, "cct"):
entities.extend(RpcShellyCctLight(coordinator, id_) for id_ in cct_key_ids)
if rgb_key_ids := get_rpc_key_ids(coordinator.device.status, "rgb"):
# RGB mode remove light & RGBW entities, add RGB entity
async_remove_shelly_rpc_entities(
hass, LIGHT_DOMAIN, coordinator.mac, [*light_keys, "rgbw:0"]
)
async_add_entities(RpcShellyRgbLight(coordinator, id_) for id_ in rgb_key_ids)
return
entities.extend(RpcShellyRgbLight(coordinator, id_) for id_ in rgb_key_ids)
if rgbw_key_ids := get_rpc_key_ids(coordinator.device.status, "rgbw"):
# RGBW mode remove light & RGB entities, add RGBW entity
async_remove_shelly_rpc_entities(
hass, LIGHT_DOMAIN, coordinator.mac, [*light_keys, "rgb:0"]
)
async_add_entities(RpcShellyRgbwLight(coordinator, id_) for id_ in rgbw_key_ids)
entities.extend(RpcShellyRgbwLight(coordinator, id_) for id_ in rgbw_key_ids)
async_add_entities(entities)
async_remove_orphaned_entities(
hass,
config_entry.entry_id,
coordinator.mac,
LIGHT_DOMAIN,
coordinator.device.status,
)
class BlockShellyLight(ShellyBlockEntity, LightEntity):
@ -427,6 +421,9 @@ class RpcShellyLightBase(ShellyRpcEntity, LightEntity):
if ATTR_BRIGHTNESS in kwargs:
params["brightness"] = brightness_to_percentage(kwargs[ATTR_BRIGHTNESS])
if ATTR_COLOR_TEMP_KELVIN in kwargs:
params["ct"] = kwargs[ATTR_COLOR_TEMP_KELVIN]
if ATTR_TRANSITION in kwargs:
params["transition_duration"] = max(
kwargs[ATTR_TRANSITION], RPC_MIN_TRANSITION_TIME_SEC
@ -472,6 +469,29 @@ class RpcShellyLight(RpcShellyLightBase):
_attr_supported_features = LightEntityFeature.TRANSITION
class RpcShellyCctLight(RpcShellyLightBase):
"""Entity that controls a CCT light on RPC based Shelly devices."""
_component = "CCT"
_attr_color_mode = ColorMode.COLOR_TEMP
_attr_supported_color_modes = {ColorMode.COLOR_TEMP}
_attr_supported_features = LightEntityFeature.TRANSITION
def __init__(self, coordinator: ShellyRpcCoordinator, id_: int) -> None:
"""Initialize light."""
color_temp_range = coordinator.device.config[f"cct:{id_}"]["ct_range"]
self._attr_min_color_temp_kelvin = color_temp_range[0]
self._attr_max_color_temp_kelvin = color_temp_range[1]
super().__init__(coordinator, id_)
@property
def color_temp_kelvin(self) -> int:
"""Return the CT color value in Kelvin."""
return cast(int, self.status["ct"])
class RpcShellyRgbLight(RpcShellyLightBase):
"""Entity that controls a RGB light on RPC based Shelly devices."""

View File

@ -35,7 +35,7 @@ from .entity import (
async_setup_entry_rpc,
)
from .utils import (
async_remove_orphaned_virtual_entities,
async_remove_orphaned_entities,
get_device_entry_gen,
get_virtual_component_ids,
)
@ -115,13 +115,13 @@ async def async_setup_entry(
virtual_number_ids = get_virtual_component_ids(
coordinator.device.config, NUMBER_PLATFORM
)
async_remove_orphaned_virtual_entities(
async_remove_orphaned_entities(
hass,
config_entry.entry_id,
coordinator.mac,
NUMBER_PLATFORM,
"number",
virtual_number_ids,
"number",
)
return

View File

@ -22,7 +22,7 @@ from .entity import (
async_setup_entry_rpc,
)
from .utils import (
async_remove_orphaned_virtual_entities,
async_remove_orphaned_entities,
get_device_entry_gen,
get_virtual_component_ids,
)
@ -61,13 +61,13 @@ async def async_setup_entry(
virtual_text_ids = get_virtual_component_ids(
coordinator.device.config, SELECT_PLATFORM
)
async_remove_orphaned_virtual_entities(
async_remove_orphaned_entities(
hass,
config_entry.entry_id,
coordinator.mac,
SELECT_PLATFORM,
"enum",
virtual_text_ids,
"enum",
)

View File

@ -53,7 +53,7 @@ from .entity import (
async_setup_entry_rpc,
)
from .utils import (
async_remove_orphaned_virtual_entities,
async_remove_orphaned_entities,
get_device_entry_gen,
get_device_uptime,
get_virtual_component_ids,
@ -392,6 +392,14 @@ RPC_SENSORS: Final = {
device_class=SensorDeviceClass.POWER,
state_class=SensorStateClass.MEASUREMENT,
),
"power_cct": RpcSensorDescription(
key="cct",
sub_key="apower",
name="Power",
native_unit_of_measurement=UnitOfPower.WATT,
device_class=SensorDeviceClass.POWER,
state_class=SensorStateClass.MEASUREMENT,
),
"power_rgb": RpcSensorDescription(
key="rgb",
sub_key="apower",
@ -552,6 +560,17 @@ RPC_SENSORS: Final = {
state_class=SensorStateClass.MEASUREMENT,
entity_registry_enabled_default=False,
),
"voltage_cct": RpcSensorDescription(
key="cct",
sub_key="voltage",
name="Voltage",
native_unit_of_measurement=UnitOfElectricPotential.VOLT,
value=lambda status, _: None if status is None else float(status),
suggested_display_precision=1,
device_class=SensorDeviceClass.VOLTAGE,
state_class=SensorStateClass.MEASUREMENT,
entity_registry_enabled_default=False,
),
"voltage_rgb": RpcSensorDescription(
key="rgb",
sub_key="voltage",
@ -641,6 +660,16 @@ RPC_SENSORS: Final = {
state_class=SensorStateClass.MEASUREMENT,
entity_registry_enabled_default=False,
),
"current_cct": RpcSensorDescription(
key="cct",
sub_key="current",
name="Current",
native_unit_of_measurement=UnitOfElectricCurrent.AMPERE,
value=lambda status, _: None if status is None else float(status),
device_class=SensorDeviceClass.CURRENT,
state_class=SensorStateClass.MEASUREMENT,
entity_registry_enabled_default=False,
),
"current_rgb": RpcSensorDescription(
key="rgb",
sub_key="current",
@ -741,6 +770,17 @@ RPC_SENSORS: Final = {
device_class=SensorDeviceClass.ENERGY,
state_class=SensorStateClass.TOTAL_INCREASING,
),
"energy_cct": RpcSensorDescription(
key="cct",
sub_key="aenergy",
name="Energy",
native_unit_of_measurement=UnitOfEnergy.WATT_HOUR,
suggested_unit_of_measurement=UnitOfEnergy.KILO_WATT_HOUR,
value=lambda status, _: status["total"],
suggested_display_precision=2,
device_class=SensorDeviceClass.ENERGY,
state_class=SensorStateClass.TOTAL_INCREASING,
),
"energy_rgb": RpcSensorDescription(
key="rgb",
sub_key="aenergy",
@ -975,6 +1015,19 @@ RPC_SENSORS: Final = {
entity_category=EntityCategory.DIAGNOSTIC,
use_polling_coordinator=True,
),
"temperature_cct": RpcSensorDescription(
key="cct",
sub_key="temperature",
name="Device temperature",
native_unit_of_measurement=UnitOfTemperature.CELSIUS,
value=lambda status, _: status["tC"],
suggested_display_precision=1,
device_class=SensorDeviceClass.TEMPERATURE,
state_class=SensorStateClass.MEASUREMENT,
entity_registry_enabled_default=False,
entity_category=EntityCategory.DIAGNOSTIC,
use_polling_coordinator=True,
),
"temperature_rgb": RpcSensorDescription(
key="rgb",
sub_key="temperature",
@ -1174,19 +1227,27 @@ async def async_setup_entry(
hass, config_entry, async_add_entities, RPC_SENSORS, RpcSensor
)
async_remove_orphaned_entities(
hass,
config_entry.entry_id,
coordinator.mac,
SENSOR_PLATFORM,
coordinator.device.status,
)
# the user can remove virtual components from the device configuration, so
# we need to remove orphaned entities
virtual_component_ids = get_virtual_component_ids(
coordinator.device.config, SENSOR_PLATFORM
)
for component in ("enum", "number", "text"):
virtual_component_ids = get_virtual_component_ids(
coordinator.device.config, SENSOR_PLATFORM
)
async_remove_orphaned_virtual_entities(
async_remove_orphaned_entities(
hass,
config_entry.entry_id,
coordinator.mac,
SENSOR_PLATFORM,
component,
virtual_component_ids,
component,
)
return

View File

@ -32,7 +32,7 @@ from .entity import (
async_setup_rpc_attribute_entities,
)
from .utils import (
async_remove_orphaned_virtual_entities,
async_remove_orphaned_entities,
async_remove_shelly_entity,
get_device_entry_gen,
get_rpc_key_ids,
@ -181,13 +181,13 @@ def async_setup_rpc_entry(
virtual_switch_ids = get_virtual_component_ids(
coordinator.device.config, SWITCH_PLATFORM
)
async_remove_orphaned_virtual_entities(
async_remove_orphaned_entities(
hass,
config_entry.entry_id,
coordinator.mac,
SWITCH_PLATFORM,
"boolean",
virtual_switch_ids,
"boolean",
)
if not switch_ids:

View File

@ -22,7 +22,7 @@ from .entity import (
async_setup_entry_rpc,
)
from .utils import (
async_remove_orphaned_virtual_entities,
async_remove_orphaned_entities,
get_device_entry_gen,
get_virtual_component_ids,
)
@ -61,13 +61,13 @@ async def async_setup_entry(
virtual_text_ids = get_virtual_component_ids(
coordinator.device.config, TEXT_PLATFORM
)
async_remove_orphaned_virtual_entities(
async_remove_orphaned_entities(
hass,
config_entry.entry_id,
coordinator.mac,
TEXT_PLATFORM,
"text",
virtual_text_ids,
"text",
)

View File

@ -2,9 +2,9 @@
from __future__ import annotations
from collections.abc import Iterable
from datetime import datetime, timedelta
from ipaddress import IPv4Address, IPv6Address, ip_address
import re
from types import MappingProxyType
from typing import Any, cast
@ -43,6 +43,7 @@ from homeassistant.util.dt import utcnow
from .const import (
API_WS_URL,
BASIC_INPUTS_EVENTS_TYPES,
COMPONENT_ID_PATTERN,
CONF_COAP_PORT,
CONF_GEN,
DEVICES_WITHOUT_FIRMWARE_CHANGELOG,
@ -326,7 +327,7 @@ def get_rpc_channel_name(device: RpcDevice, key: str) -> str:
channel_id = key.split(":")[-1]
if key.startswith(("cover:", "input:", "light:", "switch:", "thermostat:")):
return f"{device_name} {channel.title()} {channel_id}"
if key.startswith(("rgb:", "rgbw:")):
if key.startswith(("cct", "rgb:", "rgbw:")):
return f"{device_name} {channel.upper()} light {channel_id}"
if key.startswith("em1"):
return f"{device_name} EM{channel_id}"
@ -544,15 +545,15 @@ def get_virtual_component_ids(config: dict[str, Any], platform: str) -> list[str
@callback
def async_remove_orphaned_virtual_entities(
def async_remove_orphaned_entities(
hass: HomeAssistant,
config_entry_id: str,
mac: str,
platform: str,
virt_comp_type: str,
virt_comp_ids: list[str],
keys: Iterable[str],
key_suffix: str | None = None,
) -> None:
"""Remove orphaned virtual entities."""
"""Remove orphaned entities."""
orphaned_entities = []
entity_reg = er.async_get(hass)
device_reg = dr.async_get(hass)
@ -567,14 +568,15 @@ def async_remove_orphaned_virtual_entities(
for entity in entities:
if not entity.entity_id.startswith(platform):
continue
if virt_comp_type not in entity.unique_id:
if key_suffix is not None and key_suffix not in entity.unique_id:
continue
# we are looking for the component ID, e.g. boolean:201
if not (match := re.search(r"[a-z]+:\d+", entity.unique_id)):
# we are looking for the component ID, e.g. boolean:201, em1data:1
if not (match := COMPONENT_ID_PATTERN.search(entity.unique_id)):
continue
virt_comp_id = match.group()
if virt_comp_id not in virt_comp_ids:
orphaned_entities.append(f"{virt_comp_id}-{virt_comp_type}")
key = match.group()
if key not in keys:
orphaned_entities.append(entity.unique_id.split("-", 1)[1])
if orphaned_entities:
async_remove_shelly_rpc_entities(hass, platform, mac, orphaned_entities)

View File

@ -1,5 +1,6 @@
"""Tests for Shelly light platform."""
from copy import deepcopy
from unittest.mock import AsyncMock, Mock
from aioshelly.const import (
@ -15,10 +16,13 @@ import pytest
from homeassistant.components.light import (
ATTR_BRIGHTNESS,
ATTR_BRIGHTNESS_PCT,
ATTR_COLOR_MODE,
ATTR_COLOR_TEMP_KELVIN,
ATTR_EFFECT,
ATTR_EFFECT_LIST,
ATTR_MAX_COLOR_TEMP_KELVIN,
ATTR_MIN_COLOR_TEMP_KELVIN,
ATTR_RGB_COLOR,
ATTR_RGBW_COLOR,
ATTR_SUPPORTED_COLOR_MODES,
@ -29,7 +33,6 @@ from homeassistant.components.light import (
ColorMode,
LightEntityFeature,
)
from homeassistant.components.shelly.const import SHELLY_PLUS_RGBW_CHANNELS
from homeassistant.const import (
ATTR_ENTITY_ID,
ATTR_SUPPORTED_FEATURES,
@ -37,13 +40,21 @@ from homeassistant.const import (
STATE_ON,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers.device_registry import DeviceRegistry
from homeassistant.helpers.entity_registry import EntityRegistry
from . import get_entity, init_integration, mutate_rpc_device_status, register_entity
from . import (
get_entity,
init_integration,
mutate_rpc_device_status,
register_device,
register_entity,
)
from .conftest import mock_white_light_set_state
RELAY_BLOCK_ID = 0
LIGHT_BLOCK_ID = 2
SHELLY_PLUS_RGBW_CHANNELS = 4
async def test_block_device_rgbw_bulb(
@ -682,21 +693,39 @@ async def test_rpc_rgbw_device_light_mode_remove_others(
hass: HomeAssistant,
mock_rpc_device: Mock,
entity_registry: EntityRegistry,
device_registry: DeviceRegistry,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Test Shelly RPC RGBW device in light mode removes RGB/RGBW entities."""
# register lights
monkeypatch.delitem(mock_rpc_device.status, "rgb:0")
monkeypatch.delitem(mock_rpc_device.status, "rgbw:0")
register_entity(hass, LIGHT_DOMAIN, "test_rgb_0", "rgb:0")
register_entity(hass, LIGHT_DOMAIN, "test_rgbw_0", "rgbw:0")
# register rgb and rgbw lights
config_entry = await init_integration(hass, 2, skip_setup=True)
device_entry = register_device(device_registry, config_entry)
register_entity(
hass,
LIGHT_DOMAIN,
"test_rgb_0",
"rgb:0",
config_entry,
device_id=device_entry.id,
)
register_entity(
hass,
LIGHT_DOMAIN,
"test_rgbw_0",
"rgbw:0",
config_entry,
device_id=device_entry.id,
)
# verify RGB & RGBW entities created
assert get_entity(hass, LIGHT_DOMAIN, "rgb:0") is not None
assert get_entity(hass, LIGHT_DOMAIN, "rgbw:0") is not None
# init to remove RGB & RGBW
await init_integration(hass, 2)
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
# verify we have 4 lights
for i in range(SHELLY_PLUS_RGBW_CHANNELS):
@ -722,27 +751,45 @@ async def test_rpc_rgbw_device_rgb_w_modes_remove_others(
hass: HomeAssistant,
mock_rpc_device: Mock,
entity_registry: EntityRegistry,
device_registry: DeviceRegistry,
monkeypatch: pytest.MonkeyPatch,
active_mode: str,
removed_mode: str,
) -> None:
"""Test Shelly RPC RGBW device in RGB/W modes other lights."""
removed_key = f"{removed_mode}:0"
config_entry = await init_integration(hass, 2, skip_setup=True)
device_entry = register_device(device_registry, config_entry)
# register lights
for i in range(SHELLY_PLUS_RGBW_CHANNELS):
monkeypatch.delitem(mock_rpc_device.status, f"light:{i}")
entity_id = f"light.test_light_{i}"
register_entity(hass, LIGHT_DOMAIN, entity_id, f"light:{i}")
register_entity(
hass,
LIGHT_DOMAIN,
entity_id,
f"light:{i}",
config_entry,
device_id=device_entry.id,
)
monkeypatch.delitem(mock_rpc_device.status, f"{removed_mode}:0")
register_entity(hass, LIGHT_DOMAIN, f"test_{removed_key}", removed_key)
register_entity(
hass,
LIGHT_DOMAIN,
f"test_{removed_key}",
removed_key,
config_entry,
device_id=device_entry.id,
)
# verify lights entities created
for i in range(SHELLY_PLUS_RGBW_CHANNELS):
assert get_entity(hass, LIGHT_DOMAIN, f"light:{i}") is not None
assert get_entity(hass, LIGHT_DOMAIN, removed_key) is not None
await init_integration(hass, 2)
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
# verify we have RGB/w light
entity_id = f"light.test_{active_mode}_0"
@ -755,3 +802,126 @@ async def test_rpc_rgbw_device_rgb_w_modes_remove_others(
for i in range(SHELLY_PLUS_RGBW_CHANNELS):
assert get_entity(hass, LIGHT_DOMAIN, f"light:{i}") is None
assert get_entity(hass, LIGHT_DOMAIN, removed_key) is None
async def test_rpc_cct_light(
hass: HomeAssistant,
mock_rpc_device: Mock,
entity_registry: EntityRegistry,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Test RPC CCT light."""
entity_id = f"{LIGHT_DOMAIN}.test_name_cct_light_0"
config = deepcopy(mock_rpc_device.config)
config["cct:0"] = {"id": 0, "name": None, "ct_range": [3333, 5555]}
monkeypatch.setattr(mock_rpc_device, "config", config)
status = deepcopy(mock_rpc_device.status)
status["cct:0"] = {"id": 0, "output": False, "brightness": 77, "ct": 3666}
monkeypatch.setattr(mock_rpc_device, "status", status)
await init_integration(hass, 2)
entry = entity_registry.async_get(entity_id)
assert entry
assert entry.unique_id == "123456789ABC-cct:0"
# Turn off
await hass.services.async_call(
LIGHT_DOMAIN,
SERVICE_TURN_OFF,
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
mock_rpc_device.call_rpc.assert_called_once_with("CCT.Set", {"id": 0, "on": False})
state = hass.states.get(entity_id)
assert state.state == STATE_OFF
# Turn on
mock_rpc_device.call_rpc.reset_mock()
mutate_rpc_device_status(monkeypatch, mock_rpc_device, "cct:0", "output", True)
await hass.services.async_call(
LIGHT_DOMAIN,
SERVICE_TURN_ON,
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
mock_rpc_device.mock_update()
mock_rpc_device.call_rpc.assert_called_once_with("CCT.Set", {"id": 0, "on": True})
state = hass.states.get(entity_id)
assert state.state == STATE_ON
assert state.attributes[ATTR_COLOR_MODE] == ColorMode.COLOR_TEMP
assert state.attributes[ATTR_BRIGHTNESS] == 196 # 77% of 255
assert state.attributes[ATTR_COLOR_TEMP_KELVIN] == 3666
assert state.attributes[ATTR_MIN_COLOR_TEMP_KELVIN] == 3333
assert state.attributes[ATTR_MAX_COLOR_TEMP_KELVIN] == 5555
# Turn on, brightness = 88
mock_rpc_device.call_rpc.reset_mock()
await hass.services.async_call(
LIGHT_DOMAIN,
SERVICE_TURN_ON,
{ATTR_ENTITY_ID: entity_id, ATTR_BRIGHTNESS_PCT: 88},
blocking=True,
)
mutate_rpc_device_status(monkeypatch, mock_rpc_device, "cct:0", "brightness", 88)
mock_rpc_device.mock_update()
mock_rpc_device.call_rpc.assert_called_once_with(
"CCT.Set", {"id": 0, "on": True, "brightness": 88}
)
state = hass.states.get(entity_id)
assert state.state == STATE_ON
assert state.attributes[ATTR_BRIGHTNESS] == 224 # 88% of 255
# Turn on, color temp = 4444 K
mock_rpc_device.call_rpc.reset_mock()
await hass.services.async_call(
LIGHT_DOMAIN,
SERVICE_TURN_ON,
{ATTR_ENTITY_ID: entity_id, ATTR_COLOR_TEMP_KELVIN: 4444},
blocking=True,
)
mutate_rpc_device_status(monkeypatch, mock_rpc_device, "cct:0", "ct", 4444)
mock_rpc_device.mock_update()
mock_rpc_device.call_rpc.assert_called_once_with(
"CCT.Set", {"id": 0, "on": True, "ct": 4444}
)
state = hass.states.get(entity_id)
assert state.state == STATE_ON
assert state.attributes[ATTR_COLOR_TEMP_KELVIN] == 4444
async def test_rpc_remove_cct_light(
hass: HomeAssistant,
mock_rpc_device: Mock,
device_registry: DeviceRegistry,
) -> None:
"""Test Shelly RPC remove orphaned CCT light entity."""
# register CCT light entity
config_entry = await init_integration(hass, 2, skip_setup=True)
device_entry = register_device(device_registry, config_entry)
register_entity(
hass,
LIGHT_DOMAIN,
"cct_light_0",
"cct:0",
config_entry,
device_id=device_entry.id,
)
# verify CCT light entity created
assert get_entity(hass, LIGHT_DOMAIN, "cct:0") is not None
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
# there is no cct:0 in the status, so the CCT light entity should be removed
assert get_entity(hass, LIGHT_DOMAIN, "cct:0") is None