Add the switch platform to flux_led (#57444)

pull/57498/head
J. Nick Koston 2021-10-11 07:20:11 -10:00 committed by GitHub
parent d0b37229dd
commit 381301d978
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
7 changed files with 249 additions and 86 deletions

View File

@ -5,6 +5,7 @@ from datetime import timedelta
import logging
from typing import Any, Final
from flux_led import DeviceType
from flux_led.aio import AIOWifiLedBulb
from flux_led.aioscanner import AIOBulbScanner
@ -34,7 +35,7 @@ from .const import (
_LOGGER = logging.getLogger(__name__)
PLATFORMS: Final = ["light"]
PLATFORMS_BY_TYPE: Final = {DeviceType.Bulb: ["light"], DeviceType.Switch: ["switch"]}
DISCOVERY_INTERVAL: Final = timedelta(minutes=15)
REQUEST_REFRESH_DELAY: Final = 1.5
@ -149,7 +150,9 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
) from ex
coordinator = FluxLedUpdateCoordinator(hass, device)
hass.data[DOMAIN][entry.entry_id] = coordinator
hass.config_entries.async_setup_platforms(entry, PLATFORMS)
hass.config_entries.async_setup_platforms(
entry, PLATFORMS_BY_TYPE[device.device_type]
)
entry.async_on_unload(entry.add_update_listener(async_update_listener))
return True
@ -157,7 +160,9 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry."""
if unload_ok := await hass.config_entries.async_unload_platforms(entry, PLATFORMS):
device: AIOWifiLedBulb = hass.data[DOMAIN][entry.entry_id].device
platforms = PLATFORMS_BY_TYPE[device.device_type]
if unload_ok := await hass.config_entries.async_unload_platforms(entry, platforms):
coordinator = hass.data[DOMAIN].pop(entry.entry_id)
await coordinator.device.async_stop()
return unload_ok

View File

@ -0,0 +1,92 @@
"""Support for FluxLED/MagicHome lights."""
from __future__ import annotations
from abc import abstractmethod
from typing import Any, cast
from flux_led.aiodevice import AIOWifiLedBulb
from homeassistant.const import (
ATTR_MANUFACTURER,
ATTR_MODEL,
ATTR_NAME,
ATTR_SW_VERSION,
)
from homeassistant.core import callback
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from . import FluxLedUpdateCoordinator
from .const import SIGNAL_STATE_UPDATED
class FluxEntity(CoordinatorEntity):
"""Representation of a Flux entity."""
coordinator: FluxLedUpdateCoordinator
def __init__(
self,
coordinator: FluxLedUpdateCoordinator,
unique_id: str | None,
name: str,
) -> 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 self.unique_id:
self._attr_device_info = {
"connections": {(dr.CONNECTION_NETWORK_MAC, self.unique_id)},
ATTR_MODEL: f"0x{self._device.model_num:02X}",
ATTR_NAME: self.name,
ATTR_SW_VERSION: str(self._device.version_num),
ATTR_MANUFACTURER: "FluxLED/Magic Home",
}
@property
def is_on(self) -> bool:
"""Return true if device is on."""
return cast(bool, self._device.is_on)
@property
def extra_state_attributes(self) -> dict[str, str]:
"""Return the attributes."""
return {"ip_address": self._device.ipaddr}
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn the specified device on."""
await self._async_turn_on(**kwargs)
self.async_write_ha_state()
await self.coordinator.async_request_refresh()
@abstractmethod
async def _async_turn_on(self, **kwargs: Any) -> None:
"""Turn the specified device on."""
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn the specified device off."""
await self._device.async_turn_off()
self.async_write_ha_state()
await self.coordinator.async_request_refresh()
@callback
def _handle_coordinator_update(self) -> None:
"""Handle updated data from the coordinator."""
if self.coordinator.last_update_success != self._responding:
self.async_write_ha_state()
self._responding = self.coordinator.last_update_success
async def async_added_to_hass(self) -> None:
"""Handle entity which will be added."""
self.async_on_remove(
async_dispatcher_connect(
self.hass,
SIGNAL_STATE_UPDATED.format(self._device.ipaddr),
self.async_write_ha_state,
)
)
await super().async_added_to_hass()

View File

@ -6,7 +6,6 @@ import logging
import random
from typing import Any, Final, cast
from flux_led.aiodevice import AIOWifiLedBulb
from flux_led.const import (
COLOR_MODE_CCT as FLUX_COLOR_MODE_CCT,
COLOR_MODE_DIM as FLUX_COLOR_MODE_DIM,
@ -47,11 +46,7 @@ from homeassistant.components.light import (
LightEntity,
)
from homeassistant.const import (
ATTR_MANUFACTURER,
ATTR_MODE,
ATTR_MODEL,
ATTR_NAME,
ATTR_SW_VERSION,
CONF_DEVICES,
CONF_HOST,
CONF_MAC,
@ -59,10 +54,9 @@ from homeassistant.const import (
CONF_NAME,
CONF_PROTOCOL,
)
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import device_registry as dr, entity_platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_platform
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from homeassistant.helpers.update_coordinator import CoordinatorEntity
@ -92,11 +86,11 @@ from .const import (
MODE_RGB,
MODE_RGBW,
MODE_WHITE,
SIGNAL_STATE_UPDATED,
TRANSITION_GRADUAL,
TRANSITION_JUMP,
TRANSITION_STROBE,
)
from .entity import FluxEntity
_LOGGER = logging.getLogger(__name__)
@ -284,11 +278,9 @@ async def async_setup_entry(
)
class FluxLight(CoordinatorEntity, LightEntity):
class FluxLight(FluxEntity, CoordinatorEntity, LightEntity):
"""Representation of a Flux light."""
coordinator: FluxLedUpdateCoordinator
def __init__(
self,
coordinator: FluxLedUpdateCoordinator,
@ -299,11 +291,7 @@ class FluxLight(CoordinatorEntity, LightEntity):
custom_effect_transition: str,
) -> 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
super().__init__(coordinator, unique_id, name)
self._attr_supported_features = SUPPORT_FLUX_LED
self._attr_min_mireds = (
color_temperature_kelvin_to_mired(self._device.max_temp) + 1
@ -319,19 +307,6 @@ class FluxLight(CoordinatorEntity, LightEntity):
self._custom_effect_colors = custom_effect_colors
self._custom_effect_speed_pct = custom_effect_speed_pct
self._custom_effect_transition = custom_effect_transition
if self.unique_id:
self._attr_device_info = {
"connections": {(dr.CONNECTION_NETWORK_MAC, self.unique_id)},
ATTR_MODEL: f"0x{self._device.model_num:02X}",
ATTR_NAME: self.name,
ATTR_SW_VERSION: str(self._device.version_num),
ATTR_MANUFACTURER: "FluxLED/Magic Home",
}
@property
def is_on(self) -> bool:
"""Return true if device is on."""
return cast(bool, self._device.is_on)
@property
def brightness(self) -> int:
@ -382,17 +357,6 @@ class FluxLight(CoordinatorEntity, LightEntity):
return EFFECT_CUSTOM
return EFFECT_ID_NAME.get(current_mode)
@property
def extra_state_attributes(self) -> dict[str, str]:
"""Return the attributes."""
return {"ip_address": self._device.ipaddr}
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn the specified or all lights on."""
await self._async_turn_on(**kwargs)
self.async_write_ha_state()
await self.coordinator.async_request_refresh()
async def _async_turn_on(self, **kwargs: Any) -> None:
"""Turn the specified or all lights on."""
if not self.is_on:
@ -506,27 +470,3 @@ class FluxLight(CoordinatorEntity, LightEntity):
speed_pct,
transition,
)
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn the specified or all lights off."""
await self._device.async_turn_off()
self.async_write_ha_state()
await self.coordinator.async_request_refresh()
@callback
def _handle_coordinator_update(self) -> None:
"""Handle updated data from the coordinator."""
if self.coordinator.last_update_success != self._responding:
self.async_write_ha_state()
self._responding = self.coordinator.last_update_success
async def async_added_to_hass(self) -> None:
"""Handle entity which will be added."""
self.async_on_remove(
async_dispatcher_connect(
self.hass,
SIGNAL_STATE_UPDATED.format(self._device.ipaddr),
self.async_write_ha_state,
)
)
await super().async_added_to_hass()

View File

@ -0,0 +1,42 @@
"""Support for FluxLED/MagicHome switches."""
from __future__ import annotations
from typing import Any
from homeassistant import config_entries
from homeassistant.components.switch import SwitchEntity
from homeassistant.const import CONF_NAME
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from . import FluxLedUpdateCoordinator
from .const import DOMAIN
from .entity import FluxEntity
async def async_setup_entry(
hass: HomeAssistant,
entry: config_entries.ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up the Flux lights."""
coordinator: FluxLedUpdateCoordinator = hass.data[DOMAIN][entry.entry_id]
async_add_entities(
[
FluxSwitch(
coordinator,
entry.unique_id,
entry.data[CONF_NAME],
)
]
)
class FluxSwitch(FluxEntity, CoordinatorEntity, SwitchEntity):
"""Representation of a Flux switch."""
async def _async_turn_on(self, **kwargs: Any) -> None:
"""Turn the device on."""
if not self.is_on:
await self._device.async_turn_on()

View File

@ -5,6 +5,7 @@ import asyncio
from typing import Callable
from unittest.mock import AsyncMock, MagicMock, patch
from flux_led import DeviceType
from flux_led.aio import AIOWifiLedBulb
from flux_led.const import (
COLOR_MODE_CCT as FLUX_COLOR_MODE_CCT,
@ -43,6 +44,7 @@ def _mocked_bulb() -> AIOWifiLedBulb:
async def _save_setup_callback(callback: Callable) -> None:
bulb.data_receive_callback = callback
bulb.device_type = DeviceType.Bulb
bulb.async_setup = AsyncMock(side_effect=_save_setup_callback)
bulb.async_set_custom_pattern = AsyncMock()
bulb.async_set_preset_pattern = AsyncMock()
@ -76,16 +78,36 @@ def _mocked_bulb() -> AIOWifiLedBulb:
return bulb
async def async_mock_bulb_turn_off(hass: HomeAssistant, bulb: AIOWifiLedBulb) -> None:
"""Mock the bulb being off."""
def _mocked_switch() -> AIOWifiLedBulb:
switch = MagicMock(auto_spec=AIOWifiLedBulb)
async def _save_setup_callback(callback: Callable) -> None:
switch.data_receive_callback = callback
switch.device_type = DeviceType.Switch
switch.async_setup = AsyncMock(side_effect=_save_setup_callback)
switch.async_stop = AsyncMock()
switch.async_update = AsyncMock()
switch.async_turn_off = AsyncMock()
switch.async_turn_on = AsyncMock()
switch.model_num = 0x97
switch.version_num = 0x97
switch.raw_state = LEDENETRawState(
0, 0x97, 0, 0x61, 0x97, 50, 255, 0, 0, 50, 8, 0, 0, 0
)
return switch
async def async_mock_device_turn_off(hass: HomeAssistant, bulb: AIOWifiLedBulb) -> None:
"""Mock the device being off."""
bulb.is_on = False
bulb.raw_state._replace(power_state=0x24)
bulb.data_receive_callback()
await hass.async_block_till_done()
async def async_mock_bulb_turn_on(hass: HomeAssistant, bulb: AIOWifiLedBulb) -> None:
"""Mock the bulb being on."""
async def async_mock_device_turn_on(hass: HomeAssistant, bulb: AIOWifiLedBulb) -> None:
"""Mock the device being on."""
bulb.is_on = True
bulb.raw_state._replace(power_state=0x23)
bulb.data_receive_callback()

View File

@ -64,8 +64,8 @@ from . import (
_mocked_bulb,
_patch_discovery,
_patch_wifibulb,
async_mock_bulb_turn_off,
async_mock_bulb_turn_on,
async_mock_device_turn_off,
async_mock_device_turn_on,
)
from tests.common import MockConfigEntry, async_fire_time_changed
@ -206,7 +206,7 @@ async def test_rgb_light(hass: HomeAssistant) -> None:
)
bulb.async_turn_off.assert_called_once()
await async_mock_bulb_turn_off(hass, bulb)
await async_mock_device_turn_off(hass, bulb)
assert hass.states.get(entity_id).state == STATE_OFF
await hass.services.async_call(
@ -291,7 +291,7 @@ async def test_rgb_cct_light(hass: HomeAssistant) -> None:
LIGHT_DOMAIN, "turn_off", {ATTR_ENTITY_ID: entity_id}, blocking=True
)
bulb.async_turn_off.assert_called_once()
await async_mock_bulb_turn_off(hass, bulb)
await async_mock_device_turn_off(hass, bulb)
assert hass.states.get(entity_id).state == STATE_OFF
@ -343,7 +343,7 @@ async def test_rgb_cct_light(hass: HomeAssistant) -> None:
bulb.raw_state = bulb.raw_state._replace(
red=0, green=0, blue=0, warm_white=1, cool_white=2
)
await async_mock_bulb_turn_on(hass, bulb)
await async_mock_device_turn_on(hass, bulb)
state = hass.states.get(entity_id)
assert state.state == STATE_ON
attributes = state.attributes
@ -410,7 +410,7 @@ async def test_rgbw_light(hass: HomeAssistant) -> None:
LIGHT_DOMAIN, "turn_off", {ATTR_ENTITY_ID: entity_id}, blocking=True
)
bulb.async_turn_off.assert_called_once()
await async_mock_bulb_turn_off(hass, bulb)
await async_mock_device_turn_off(hass, bulb)
assert hass.states.get(entity_id).state == STATE_OFF
@ -513,7 +513,7 @@ async def test_rgbcw_light(hass: HomeAssistant) -> None:
LIGHT_DOMAIN, "turn_off", {ATTR_ENTITY_ID: entity_id}, blocking=True
)
bulb.async_turn_off.assert_called_once()
await async_mock_bulb_turn_off(hass, bulb)
await async_mock_device_turn_off(hass, bulb)
assert hass.states.get(entity_id).state == STATE_OFF
@ -640,7 +640,7 @@ async def test_white_light(hass: HomeAssistant) -> None:
LIGHT_DOMAIN, "turn_off", {ATTR_ENTITY_ID: entity_id}, blocking=True
)
bulb.async_turn_off.assert_called_once()
await async_mock_bulb_turn_off(hass, bulb)
await async_mock_device_turn_off(hass, bulb)
assert hass.states.get(entity_id).state == STATE_OFF
@ -705,7 +705,7 @@ async def test_rgb_light_custom_effects(hass: HomeAssistant) -> None:
LIGHT_DOMAIN, "turn_off", {ATTR_ENTITY_ID: entity_id}, blocking=True
)
bulb.async_turn_off.assert_called_once()
await async_mock_bulb_turn_off(hass, bulb)
await async_mock_device_turn_off(hass, bulb)
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_OFF
@ -720,7 +720,7 @@ async def test_rgb_light_custom_effects(hass: HomeAssistant) -> None:
)
bulb.async_set_custom_pattern.reset_mock()
bulb.preset_pattern_num = EFFECT_CUSTOM_CODE
await async_mock_bulb_turn_on(hass, bulb)
await async_mock_device_turn_on(hass, bulb)
state = hass.states.get(entity_id)
assert state.state == STATE_ON
@ -738,7 +738,7 @@ async def test_rgb_light_custom_effects(hass: HomeAssistant) -> None:
)
bulb.async_set_custom_pattern.reset_mock()
bulb.preset_pattern_num = EFFECT_CUSTOM_CODE
await async_mock_bulb_turn_on(hass, bulb)
await async_mock_device_turn_on(hass, bulb)
state = hass.states.get(entity_id)
assert state.state == STATE_ON
@ -812,7 +812,7 @@ async def test_rgb_light_custom_effect_via_service(
)
bulb.async_turn_off.assert_called_once()
await async_mock_bulb_turn_off(hass, bulb)
await async_mock_device_turn_off(hass, bulb)
assert hass.states.get(entity_id).state == STATE_OFF
await hass.services.async_call(
@ -911,7 +911,7 @@ async def test_addressable_light(hass: HomeAssistant) -> None:
)
bulb.async_turn_off.assert_called_once()
await async_mock_bulb_turn_off(hass, bulb)
await async_mock_device_turn_off(hass, bulb)
assert hass.states.get(entity_id).state == STATE_OFF
await hass.services.async_call(
@ -919,7 +919,7 @@ async def test_addressable_light(hass: HomeAssistant) -> None:
)
bulb.async_turn_on.assert_called_once()
bulb.async_turn_on.reset_mock()
await async_mock_bulb_turn_on(hass, bulb)
await async_mock_device_turn_on(hass, bulb)
with pytest.raises(ValueError):
await hass.services.async_call(

View File

@ -0,0 +1,62 @@
"""Tests for switch platform."""
from homeassistant.components import flux_led
from homeassistant.components.flux_led.const import DOMAIN
from homeassistant.components.switch import DOMAIN as SWITCH_DOMAIN
from homeassistant.const import (
ATTR_ENTITY_ID,
CONF_HOST,
CONF_NAME,
STATE_OFF,
STATE_ON,
)
from homeassistant.core import HomeAssistant
from homeassistant.setup import async_setup_component
from . import (
DEFAULT_ENTRY_TITLE,
IP_ADDRESS,
MAC_ADDRESS,
_mocked_switch,
_patch_discovery,
_patch_wifibulb,
async_mock_device_turn_off,
async_mock_device_turn_on,
)
from tests.common import MockConfigEntry
async def test_switch_on_off(hass: HomeAssistant) -> None:
"""Test a switch light."""
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)
switch = _mocked_switch()
with _patch_discovery(device=switch), _patch_wifibulb(device=switch):
await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}})
await hass.async_block_till_done()
entity_id = "switch.az120444_aabbccddeeff"
state = hass.states.get(entity_id)
assert state.state == STATE_ON
await hass.services.async_call(
SWITCH_DOMAIN, "turn_off", {ATTR_ENTITY_ID: entity_id}, blocking=True
)
switch.async_turn_off.assert_called_once()
await async_mock_device_turn_off(hass, switch)
assert hass.states.get(entity_id).state == STATE_OFF
await hass.services.async_call(
SWITCH_DOMAIN, "turn_on", {ATTR_ENTITY_ID: entity_id}, blocking=True
)
switch.async_turn_on.assert_called_once()
switch.async_turn_on.reset_mock()
await async_mock_device_turn_on(hass, switch)
assert hass.states.get(entity_id).state == STATE_ON