1
mirror of https://github.com/home-assistant/core synced 2024-08-02 23:40:32 +02:00

Add device configuration entities to flux_led (#62786)

Co-authored-by: Chris Talkington <chris@talkingtontech.com>
This commit is contained in:
J. Nick Koston 2022-01-06 21:02:19 -10:00 committed by GitHub
parent 250af90acb
commit e222e1b6f0
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
10 changed files with 626 additions and 48 deletions

View File

@ -48,6 +48,7 @@ PLATFORMS_BY_TYPE: Final = {
Platform.BUTTON,
Platform.LIGHT,
Platform.NUMBER,
Platform.SELECT,
Platform.SWITCH,
],
DeviceType.Switch: [Platform.BUTTON, Platform.SELECT, Platform.SWITCH],

View File

@ -28,7 +28,6 @@ async def async_setup_entry(
class FluxRestartButton(FluxBaseEntity, ButtonEntity):
"""Representation of a Flux restart button."""
_attr_should_poll = False
_attr_entity_category = EntityCategory.CONFIG
def __init__(

View File

@ -40,6 +40,8 @@ def _async_device_info(
class FluxBaseEntity(Entity):
"""Representation of a Flux entity without a coordinator."""
_attr_should_poll = False
def __init__(
self,
device: AIOWifiLedBulb,
@ -64,13 +66,17 @@ class FluxEntity(CoordinatorEntity):
coordinator: FluxLedUpdateCoordinator,
unique_id: str | None,
name: str,
key: str | None,
) -> None:
"""Initialize the light."""
super().__init__(coordinator)
self._device: AIOWifiLedBulb = coordinator.device
self._responding = True
self._attr_name = name
self._attr_unique_id = unique_id
if key:
self._attr_unique_id = f"{unique_id}_{key}"
else:
self._attr_unique_id = unique_id
if unique_id:
self._attr_device_info = _async_device_info(
unique_id, self._device, coordinator.entry

View File

@ -202,7 +202,7 @@ class FluxLight(FluxOnOffEntity, CoordinatorEntity, LightEntity):
custom_effect_transition: str,
) -> None:
"""Initialize the light."""
super().__init__(coordinator, unique_id, name)
super().__init__(coordinator, unique_id, name, None)
self._attr_min_mireds = (
color_temperature_kelvin_to_mired(self._device.max_temp) + 1
) # for rounding

View File

@ -1,20 +1,38 @@
"""Support for LED numbers."""
from __future__ import annotations
from abc import abstractmethod
import logging
from typing import cast
from flux_led.protocol import (
MUSIC_PIXELS_MAX,
MUSIC_PIXELS_PER_SEGMENT_MAX,
MUSIC_SEGMENTS_MAX,
PIXELS_MAX,
PIXELS_PER_SEGMENT_MAX,
SEGMENTS_MAX,
)
from homeassistant import config_entries
from homeassistant.components.light import EFFECT_RANDOM
from homeassistant.components.number import NumberEntity, NumberMode
from homeassistant.const import CONF_NAME
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.debounce import Debouncer
from homeassistant.helpers.entity import EntityCategory
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import DOMAIN, EFFECT_SPEED_SUPPORT_MODES
from .const import DOMAIN
from .coordinator import FluxLedUpdateCoordinator
from .entity import FluxEntity
from .util import _effect_brightness, _hass_color_modes
from .util import _effect_brightness
_LOGGER = logging.getLogger(__name__)
DEBOUNCE_TIME = 1
async def async_setup_entry(
@ -24,23 +42,55 @@ async def async_setup_entry(
) -> None:
"""Set up the Flux lights."""
coordinator: FluxLedUpdateCoordinator = hass.data[DOMAIN][entry.entry_id]
device = coordinator.device
entities: list[
FluxSpeedNumber
| FluxPixelsPerSegmentNumber
| FluxSegmentsNumber
| FluxMusicPixelsPerSegmentNumber
| FluxMusicSegmentsNumber
] = []
name = entry.data[CONF_NAME]
unique_id = entry.unique_id
color_modes = _hass_color_modes(coordinator.device)
if not color_modes.intersection(EFFECT_SPEED_SUPPORT_MODES):
return
async_add_entities(
[
FluxNumber(
if device.pixels_per_segment is not None:
entities.append(
FluxPixelsPerSegmentNumber(
coordinator,
entry.unique_id,
entry.data[CONF_NAME],
unique_id,
f"{name} Pixels Per Segment",
"pixels_per_segment",
)
]
)
)
if device.segments is not None:
entities.append(
FluxSegmentsNumber(coordinator, unique_id, f"{name} Segments", "segments")
)
if device.music_pixels_per_segment is not None:
entities.append(
FluxMusicPixelsPerSegmentNumber(
coordinator,
unique_id,
f"{name} Music Pixels Per Segment",
"music_pixels_per_segment",
)
)
if device.music_segments is not None:
entities.append(
FluxMusicSegmentsNumber(
coordinator, unique_id, f"{name} Music Segments", "music_segments"
)
)
if device.effect_list and device.effect_list != [EFFECT_RANDOM]:
entities.append(
FluxSpeedNumber(coordinator, unique_id, f"{name} Effect Speed", None)
)
if entities:
async_add_entities(entities)
class FluxNumber(FluxEntity, CoordinatorEntity, NumberEntity):
class FluxSpeedNumber(FluxEntity, CoordinatorEntity, NumberEntity):
"""Defines a flux_led speed number."""
_attr_min_value = 1
@ -49,16 +99,6 @@ class FluxNumber(FluxEntity, CoordinatorEntity, NumberEntity):
_attr_mode = NumberMode.SLIDER
_attr_icon = "mdi:speedometer"
def __init__(
self,
coordinator: FluxLedUpdateCoordinator,
unique_id: str | None,
name: str,
) -> None:
"""Initialize the flux number."""
super().__init__(coordinator, unique_id, name)
self._attr_name = f"{name} Effect Speed"
@property
def value(self) -> float:
"""Return the effect speed."""
@ -78,3 +118,174 @@ class FluxNumber(FluxEntity, CoordinatorEntity, NumberEntity):
current_effect, new_speed, _effect_brightness(self._device.brightness)
)
await self.coordinator.async_request_refresh()
class FluxConfigNumber(FluxEntity, CoordinatorEntity, NumberEntity):
"""Base class for flux config numbers."""
_attr_entity_category = EntityCategory.CONFIG
_attr_min_value = 1
_attr_step = 1
_attr_mode = NumberMode.BOX
def __init__(
self,
coordinator: FluxLedUpdateCoordinator,
unique_id: str | None,
name: str,
key: str | None,
) -> None:
"""Initialize the flux number."""
super().__init__(coordinator, unique_id, name, key)
self._debouncer: Debouncer | None = None
self._pending_value: int | None = None
async def async_added_to_hass(self) -> None:
"""Set up the debouncer when adding to hass."""
self._debouncer = Debouncer(
hass=self.hass,
logger=_LOGGER,
cooldown=DEBOUNCE_TIME,
immediate=False,
function=self._async_set_value,
)
await super().async_added_to_hass()
async def async_set_value(self, value: float) -> None:
"""Set the value."""
self._pending_value = int(value)
assert self._debouncer is not None
await self._debouncer.async_call()
@abstractmethod
async def _async_set_value(self) -> None:
"""Call on debounce to set the value."""
def _pixels_and_segments_fit_in_music_mode(self) -> bool:
"""Check if the base pixel and segment settings will fit for music mode.
If they fit, they do not need to be configured.
"""
pixels_per_segment = self._device.pixels_per_segment
segments = self._device.segments
assert pixels_per_segment is not None
assert segments is not None
return bool(
pixels_per_segment <= MUSIC_PIXELS_PER_SEGMENT_MAX
and segments <= MUSIC_SEGMENTS_MAX
and pixels_per_segment * segments <= MUSIC_PIXELS_MAX
)
class FluxPixelsPerSegmentNumber(FluxConfigNumber):
"""Defines a flux_led pixels per segment number."""
_attr_icon = "mdi:dots-grid"
@property
def max_value(self) -> int:
"""Return the max value."""
return min(
PIXELS_PER_SEGMENT_MAX, int(PIXELS_MAX / (self._device.segments or 1))
)
@property
def value(self) -> int:
"""Return the pixels per segment."""
assert self._device.pixels_per_segment is not None
return self._device.pixels_per_segment
async def _async_set_value(self) -> None:
"""Set the pixels per segment."""
assert self._pending_value is not None
await self._device.async_set_device_config(
pixels_per_segment=self._pending_value
)
class FluxSegmentsNumber(FluxConfigNumber):
"""Defines a flux_led segments number."""
_attr_icon = "mdi:segment"
@property
def max_value(self) -> int:
"""Return the max value."""
assert self._device.pixels_per_segment is not None
return min(
SEGMENTS_MAX, int(PIXELS_MAX / (self._device.pixels_per_segment or 1))
)
@property
def value(self) -> int:
"""Return the segments."""
assert self._device.segments is not None
return self._device.segments
async def _async_set_value(self) -> None:
"""Set the segments."""
assert self._pending_value is not None
await self._device.async_set_device_config(segments=self._pending_value)
class FluxMusicNumber(FluxConfigNumber):
"""A number that is only available if the base pixels do not fit in music mode."""
@property
def available(self) -> bool:
"""Return if music pixels per segment can be set."""
return super().available and not self._pixels_and_segments_fit_in_music_mode()
class FluxMusicPixelsPerSegmentNumber(FluxMusicNumber):
"""Defines a flux_led music pixels per segment number."""
_attr_icon = "mdi:dots-grid"
@property
def max_value(self) -> int:
"""Return the max value."""
assert self._device.music_segments is not None
return min(
MUSIC_PIXELS_PER_SEGMENT_MAX,
int(MUSIC_PIXELS_MAX / (self._device.music_segments or 1)),
)
@property
def value(self) -> int:
"""Return the music pixels per segment."""
assert self._device.music_pixels_per_segment is not None
return self._device.music_pixels_per_segment
async def _async_set_value(self) -> None:
"""Set the music pixels per segment."""
assert self._pending_value is not None
await self._device.async_set_device_config(
music_pixels_per_segment=self._pending_value
)
class FluxMusicSegmentsNumber(FluxMusicNumber):
"""Defines a flux_led music segments number."""
_attr_icon = "mdi:segment"
@property
def max_value(self) -> int:
"""Return the max value."""
assert self._device.pixels_per_segment is not None
return min(
MUSIC_SEGMENTS_MAX,
int(MUSIC_PIXELS_MAX / (self._device.music_pixels_per_segment or 1)),
)
@property
def value(self) -> int:
"""Return the music segments."""
assert self._device.music_segments is not None
return self._device.music_segments
async def _async_set_value(self) -> None:
"""Set the music segments."""
assert self._pending_value is not None
await self._device.async_set_device_config(music_segments=self._pending_value)

View File

@ -2,6 +2,7 @@
from __future__ import annotations
from flux_led.aio import AIOWifiLedBulb
from flux_led.base_device import DeviceType
from flux_led.protocol import PowerRestoreState
from homeassistant import config_entries
@ -13,7 +14,7 @@ from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .const import DOMAIN
from .coordinator import FluxLedUpdateCoordinator
from .entity import FluxBaseEntity
from .entity import FluxBaseEntity, FluxEntity
async def async_setup_entry(
@ -23,17 +24,46 @@ async def async_setup_entry(
) -> None:
"""Set up the Flux selects."""
coordinator: FluxLedUpdateCoordinator = hass.data[DOMAIN][entry.entry_id]
async_add_entities([FluxPowerState(coordinator.device, entry)])
device = coordinator.device
entities: list[
FluxPowerStateSelect
| FluxOperatingModesSelect
| FluxWiringsSelect
| FluxICTypeSelect
] = []
name = entry.data[CONF_NAME]
unique_id = entry.unique_id
if device.device_type == DeviceType.Switch:
entities.append(FluxPowerStateSelect(coordinator.device, entry))
if device.operating_modes:
entities.append(
FluxOperatingModesSelect(
coordinator, unique_id, f"{name} Operating Mode", "operating_mode"
)
)
if device.wirings:
entities.append(
FluxWiringsSelect(coordinator, unique_id, f"{name} Wiring", "wiring")
)
if device.ic_types:
entities.append(
FluxICTypeSelect(coordinator, unique_id, f"{name} IC Type", "ic_type")
)
if entities:
async_add_entities(entities)
def _human_readable_option(const_option: str) -> str:
return const_option.replace("_", " ").title()
class FluxPowerState(FluxBaseEntity, SelectEntity):
class FluxPowerStateSelect(FluxBaseEntity, SelectEntity):
"""Representation of a Flux power restore state option."""
_attr_should_poll = False
_attr_icon = "mdi:transmission-tower-off"
_attr_entity_category = EntityCategory.CONFIG
def __init__(
self,
@ -42,7 +72,6 @@ class FluxPowerState(FluxBaseEntity, SelectEntity):
) -> None:
"""Initialize the power state select."""
super().__init__(device, entry)
self._attr_entity_category = EntityCategory.CONFIG
self._attr_name = f"{entry.data[CONF_NAME]} Power Restored"
if entry.unique_id:
self._attr_unique_id = f"{entry.unique_id}_power_restored"
@ -65,3 +94,74 @@ class FluxPowerState(FluxBaseEntity, SelectEntity):
await self._device.async_set_power_restore(channel1=self._name_to_state[option])
self._async_set_current_option_from_device()
self.async_write_ha_state()
class FluxConfigSelect(FluxEntity, SelectEntity):
"""Representation of a flux config entity that updates."""
_attr_entity_category = EntityCategory.CONFIG
class FluxICTypeSelect(FluxConfigSelect):
"""Representation of Flux ic type."""
_attr_icon = "mdi:chip"
@property
def options(self) -> list[str]:
"""Return the available ic types."""
assert self._device.ic_types is not None
return self._device.ic_types
@property
def current_option(self) -> str | None:
"""Return the current ic type."""
return self._device.ic_type
async def async_select_option(self, option: str) -> None:
"""Change the ic type."""
await self._device.async_set_device_config(ic_type=option)
class FluxWiringsSelect(FluxConfigSelect):
"""Representation of Flux wirings."""
_attr_icon = "mdi:led-strip-variant"
@property
def options(self) -> list[str]:
"""Return the available wiring options based on the strip protocol."""
assert self._device.wirings is not None
return self._device.wirings
@property
def current_option(self) -> str | None:
"""Return the current wiring."""
return self._device.wiring
async def async_select_option(self, option: str) -> None:
"""Change the wiring."""
await self._device.async_set_device_config(wiring=option)
class FluxOperatingModesSelect(FluxConfigSelect):
"""Representation of Flux operating modes."""
@property
def options(self) -> list[str]:
"""Return the current operating mode."""
assert self._device.operating_modes is not None
return self._device.operating_modes
@property
def current_option(self) -> str | None:
"""Return the current operating mode."""
return self._device.operating_mode
async def async_select_option(self, option: str) -> None:
"""Change the ic type."""
await self._device.async_set_device_config(operating_mode=option)
# reload since we need to reinit the device
self.hass.async_create_task(
self.hass.config_entries.async_reload(self.coordinator.entry.entry_id)
)

View File

@ -38,13 +38,15 @@ async def async_setup_entry(
name = entry.data[CONF_NAME]
if coordinator.device.device_type == DeviceType.Switch:
entities.append(FluxSwitch(coordinator, unique_id, name))
entities.append(FluxSwitch(coordinator, unique_id, name, None))
if entry.data.get(CONF_REMOTE_ACCESS_HOST):
entities.append(FluxRemoteAccessSwitch(coordinator.device, entry))
if coordinator.device.microphone:
entities.append(FluxMusicSwitch(coordinator, unique_id, name))
entities.append(
FluxMusicSwitch(coordinator, unique_id, f"{name} Music", "music")
)
if entities:
async_add_entities(entities)
@ -62,7 +64,6 @@ class FluxSwitch(FluxOnOffEntity, CoordinatorEntity, SwitchEntity):
class FluxRemoteAccessSwitch(FluxBaseEntity, SwitchEntity):
"""Representation of a Flux remote access switch."""
_attr_should_poll = False
_attr_entity_category = EntityCategory.CONFIG
def __init__(
@ -112,18 +113,6 @@ class FluxRemoteAccessSwitch(FluxBaseEntity, SwitchEntity):
class FluxMusicSwitch(FluxEntity, SwitchEntity):
"""Representation of a Flux music switch."""
def __init__(
self,
coordinator: FluxLedUpdateCoordinator,
unique_id: str | None,
name: str,
) -> None:
"""Initialize the flux music switch."""
super().__init__(coordinator, unique_id, name)
self._attr_name = f"{name} Music"
if unique_id:
self._attr_unique_id = f"{unique_id}_music"
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn the microphone on."""
await self._async_ensure_device_on()

View File

@ -81,6 +81,7 @@ def _mocked_bulb() -> AIOWifiLedBulb:
bulb.async_set_effect = AsyncMock()
bulb.async_set_white_temp = AsyncMock()
bulb.async_set_brightness = AsyncMock()
bulb.async_set_device_config = AsyncMock()
bulb.pixels_per_segment = 300
bulb.segments = 2
bulb.music_pixels_per_segment = 150
@ -142,6 +143,16 @@ def _mocked_switch() -> AIOWifiLedBulb:
channel3=PowerRestoreState.LAST_STATE,
channel4=PowerRestoreState.LAST_STATE,
)
switch.pixels_per_segment = None
switch.segments = None
switch.music_pixels_per_segment = None
switch.music_segments = None
switch.operating_mode = None
switch.operating_modes = None
switch.wirings = None
switch.wiring = None
switch.ic_types = None
switch.ic_type = None
switch.requires_turn_on = True
switch.async_set_time = AsyncMock()
switch.async_reboot = AsyncMock()

View File

@ -1,17 +1,26 @@
"""Tests for the flux_led number platform."""
from unittest.mock import patch
from flux_led.const import COLOR_MODE_RGB as FLUX_COLOR_MODE_RGB
import pytest
from homeassistant.components import flux_led
from homeassistant.components.flux_led import number as flux_number
from homeassistant.components.flux_led.const import DOMAIN
from homeassistant.components.number import (
ATTR_VALUE,
DOMAIN as NUMBER_DOMAIN,
SERVICE_SET_VALUE,
)
from homeassistant.const import ATTR_ENTITY_ID, CONF_HOST, CONF_NAME, STATE_ON
from homeassistant.const import (
ATTR_ENTITY_ID,
CONF_HOST,
CONF_NAME,
STATE_ON,
STATE_UNAVAILABLE,
)
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import entity_registry as er
@ -225,3 +234,155 @@ async def test_addressable_light_effect_speed(hass: HomeAssistant) -> None:
state = hass.states.get(number_entity_id)
assert state.state == "100"
async def test_addressable_light_pixel_config(hass: HomeAssistant) -> None:
"""Test an addressable light pixel config."""
config_entry = MockConfigEntry(
domain=DOMAIN,
data={CONF_HOST: IP_ADDRESS, CONF_NAME: DEFAULT_ENTRY_TITLE},
unique_id=MAC_ADDRESS,
)
config_entry.add_to_hass(hass)
bulb = _mocked_bulb()
bulb.raw_state = bulb.raw_state._replace(
model_num=0xA2
) # Original addressable model
bulb.color_modes = {FLUX_COLOR_MODE_RGB}
bulb.color_mode = FLUX_COLOR_MODE_RGB
with patch.object(
flux_number, "DEBOUNCE_TIME", 0
), _patch_discovery(), _patch_wifibulb(device=bulb):
await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}})
await hass.async_block_till_done()
pixels_per_segment_entity_id = "number.bulb_rgbcw_ddeeff_pixels_per_segment"
state = hass.states.get(pixels_per_segment_entity_id)
assert state.state == "300"
segments_entity_id = "number.bulb_rgbcw_ddeeff_segments"
state = hass.states.get(segments_entity_id)
assert state.state == "2"
music_pixels_per_segment_entity_id = (
"number.bulb_rgbcw_ddeeff_music_pixels_per_segment"
)
state = hass.states.get(music_pixels_per_segment_entity_id)
assert state.state == "150"
music_segments_entity_id = "number.bulb_rgbcw_ddeeff_music_segments"
state = hass.states.get(music_segments_entity_id)
assert state.state == "4"
with pytest.raises(ValueError):
await hass.services.async_call(
NUMBER_DOMAIN,
SERVICE_SET_VALUE,
{ATTR_ENTITY_ID: pixels_per_segment_entity_id, ATTR_VALUE: 5000},
blocking=True,
)
await hass.services.async_call(
NUMBER_DOMAIN,
SERVICE_SET_VALUE,
{ATTR_ENTITY_ID: pixels_per_segment_entity_id, ATTR_VALUE: 100},
blocking=True,
)
bulb.async_set_device_config.assert_called_with(pixels_per_segment=100)
bulb.async_set_device_config.reset_mock()
with pytest.raises(ValueError):
await hass.services.async_call(
NUMBER_DOMAIN,
SERVICE_SET_VALUE,
{ATTR_ENTITY_ID: music_pixels_per_segment_entity_id, ATTR_VALUE: 5000},
blocking=True,
)
await hass.services.async_call(
NUMBER_DOMAIN,
SERVICE_SET_VALUE,
{ATTR_ENTITY_ID: music_pixels_per_segment_entity_id, ATTR_VALUE: 100},
blocking=True,
)
bulb.async_set_device_config.assert_called_with(music_pixels_per_segment=100)
bulb.async_set_device_config.reset_mock()
with pytest.raises(ValueError):
await hass.services.async_call(
NUMBER_DOMAIN,
SERVICE_SET_VALUE,
{ATTR_ENTITY_ID: segments_entity_id, ATTR_VALUE: 50},
blocking=True,
)
await hass.services.async_call(
NUMBER_DOMAIN,
SERVICE_SET_VALUE,
{ATTR_ENTITY_ID: segments_entity_id, ATTR_VALUE: 5},
blocking=True,
)
bulb.async_set_device_config.assert_called_with(segments=5)
bulb.async_set_device_config.reset_mock()
with pytest.raises(ValueError):
await hass.services.async_call(
NUMBER_DOMAIN,
SERVICE_SET_VALUE,
{ATTR_ENTITY_ID: music_segments_entity_id, ATTR_VALUE: 50},
blocking=True,
)
await hass.services.async_call(
NUMBER_DOMAIN,
SERVICE_SET_VALUE,
{ATTR_ENTITY_ID: music_segments_entity_id, ATTR_VALUE: 5},
blocking=True,
)
bulb.async_set_device_config.assert_called_with(music_segments=5)
bulb.async_set_device_config.reset_mock()
async def test_addressable_light_pixel_config_music_disabled(
hass: HomeAssistant,
) -> None:
"""Test an addressable light pixel config with music pixels disabled."""
config_entry = MockConfigEntry(
domain=DOMAIN,
data={CONF_HOST: IP_ADDRESS, CONF_NAME: DEFAULT_ENTRY_TITLE},
unique_id=MAC_ADDRESS,
)
config_entry.add_to_hass(hass)
bulb = _mocked_bulb()
bulb.pixels_per_segment = 150
bulb.segments = 1
bulb.music_pixels_per_segment = 150
bulb.music_segments = 1
bulb.raw_state = bulb.raw_state._replace(
model_num=0xA2
) # Original addressable model
bulb.color_modes = {FLUX_COLOR_MODE_RGB}
bulb.color_mode = FLUX_COLOR_MODE_RGB
with patch.object(
flux_number, "DEBOUNCE_TIME", 0
), _patch_discovery(), _patch_wifibulb(device=bulb):
await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}})
await hass.async_block_till_done()
pixels_per_segment_entity_id = "number.bulb_rgbcw_ddeeff_pixels_per_segment"
state = hass.states.get(pixels_per_segment_entity_id)
assert state.state == "150"
segments_entity_id = "number.bulb_rgbcw_ddeeff_segments"
state = hass.states.get(segments_entity_id)
assert state.state == "1"
music_pixels_per_segment_entity_id = (
"number.bulb_rgbcw_ddeeff_music_pixels_per_segment"
)
state = hass.states.get(music_pixels_per_segment_entity_id)
assert state.state == STATE_UNAVAILABLE
music_segments_entity_id = "number.bulb_rgbcw_ddeeff_music_segments"
state = hass.states.get(music_segments_entity_id)
assert state.state == STATE_UNAVAILABLE

View File

@ -1,5 +1,8 @@
"""Tests for select platform."""
from unittest.mock import patch
from flux_led.protocol import PowerRestoreState
import pytest
from homeassistant.components import flux_led
from homeassistant.components.flux_led.const import DOMAIN
@ -12,6 +15,7 @@ from . import (
DEFAULT_ENTRY_TITLE,
IP_ADDRESS,
MAC_ADDRESS,
_mocked_bulb,
_mocked_switch,
_patch_discovery,
_patch_wifibulb,
@ -47,3 +51,99 @@ async def test_switch_power_restore_state(hass: HomeAssistant) -> None:
switch.async_set_power_restore.assert_called_once_with(
channel1=PowerRestoreState.ALWAYS_ON
)
async def test_select_addressable_strip_config(hass: HomeAssistant) -> None:
"""Test selecting addressable strip configs."""
config_entry = MockConfigEntry(
domain=DOMAIN,
data={CONF_HOST: IP_ADDRESS, CONF_NAME: DEFAULT_ENTRY_TITLE},
unique_id=MAC_ADDRESS,
)
config_entry.add_to_hass(hass)
bulb = _mocked_bulb()
bulb.raw_state = bulb.raw_state._replace(model_num=0xA2) # addressable model
with _patch_discovery(), _patch_wifibulb(device=bulb):
await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}})
await hass.async_block_till_done()
wiring_entity_id = "select.bulb_rgbcw_ddeeff_wiring"
state = hass.states.get(wiring_entity_id)
assert state.state == "BGRW"
ic_type_entity_id = "select.bulb_rgbcw_ddeeff_ic_type"
state = hass.states.get(ic_type_entity_id)
assert state.state == "WS2812B"
with pytest.raises(ValueError):
await hass.services.async_call(
SELECT_DOMAIN,
"select_option",
{ATTR_ENTITY_ID: wiring_entity_id, ATTR_OPTION: "INVALID"},
blocking=True,
)
await hass.services.async_call(
SELECT_DOMAIN,
"select_option",
{ATTR_ENTITY_ID: wiring_entity_id, ATTR_OPTION: "GRBW"},
blocking=True,
)
bulb.async_set_device_config.assert_called_once_with(wiring="GRBW")
bulb.async_set_device_config.reset_mock()
with pytest.raises(ValueError):
await hass.services.async_call(
SELECT_DOMAIN,
"select_option",
{ATTR_ENTITY_ID: ic_type_entity_id, ATTR_OPTION: "INVALID"},
blocking=True,
)
await hass.services.async_call(
SELECT_DOMAIN,
"select_option",
{ATTR_ENTITY_ID: ic_type_entity_id, ATTR_OPTION: "UCS1618"},
blocking=True,
)
bulb.async_set_device_config.assert_called_once_with(ic_type="UCS1618")
async def test_select_mutable_0x25_strip_config(hass: HomeAssistant) -> None:
"""Test selecting mutable 0x25 strip configs."""
config_entry = MockConfigEntry(
domain=DOMAIN,
data={CONF_HOST: IP_ADDRESS, CONF_NAME: DEFAULT_ENTRY_TITLE},
unique_id=MAC_ADDRESS,
)
config_entry.add_to_hass(hass)
bulb = _mocked_bulb()
bulb.operating_mode = "RGBWW"
bulb.operating_modes = ["DIM", "CCT", "RGB", "RGBW", "RGBWW"]
bulb.raw_state = bulb.raw_state._replace(model_num=0x25) # addressable model
with _patch_discovery(), _patch_wifibulb(device=bulb):
await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}})
await hass.async_block_till_done()
operating_mode_entity_id = "select.bulb_rgbcw_ddeeff_operating_mode"
state = hass.states.get(operating_mode_entity_id)
assert state.state == "RGBWW"
with pytest.raises(ValueError):
await hass.services.async_call(
SELECT_DOMAIN,
"select_option",
{ATTR_ENTITY_ID: operating_mode_entity_id, ATTR_OPTION: "INVALID"},
blocking=True,
)
with patch(
"homeassistant.components.flux_led.async_setup_entry"
) as mock_setup_entry:
await hass.services.async_call(
SELECT_DOMAIN,
"select_option",
{ATTR_ENTITY_ID: operating_mode_entity_id, ATTR_OPTION: "CCT"},
blocking=True,
)
await hass.async_block_till_done()
bulb.async_set_device_config.assert_called_once_with(operating_mode="CCT")
assert len(mock_setup_entry.mock_calls) == 1