Add binary sensor platform to LIFX integration (#77535)

Co-authored-by: J. Nick Koston <nick@koston.org>

homeassistant/components/lifx/__init__.py

@@ -57,7 +57,7 @@ CONFIG_SCHEMA = vol.All(
 )
 
 
-PLATFORMS = [Platform.BUTTON, Platform.LIGHT]
+PLATFORMS = [Platform.BINARY_SENSOR, Platform.BUTTON, Platform.LIGHT]
 DISCOVERY_INTERVAL = timedelta(minutes=15)
 MIGRATION_INTERVAL = timedelta(minutes=5)
 

homeassistant/components/lifx/binary_sensor.py

@@ -0,0 +1,70 @@
+"""Binary sensor entities for LIFX integration."""
+from __future__ import annotations
+
+from homeassistant.components.binary_sensor import (
+    BinarySensorDeviceClass,
+    BinarySensorEntity,
+    BinarySensorEntityDescription,
+)
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant, callback
+from homeassistant.helpers.entity import EntityCategory
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+
+from .const import DOMAIN, HEV_CYCLE_STATE
+from .coordinator import LIFXUpdateCoordinator
+from .entity import LIFXEntity
+from .util import lifx_features
+
+HEV_CYCLE_STATE_SENSOR = BinarySensorEntityDescription(
+    key=HEV_CYCLE_STATE,
+    name="Clean Cycle",
+    entity_category=EntityCategory.DIAGNOSTIC,
+    device_class=BinarySensorDeviceClass.RUNNING,
+)
+
+
+async def async_setup_entry(
+    hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
+) -> None:
+    """Set up LIFX from a config entry."""
+    coordinator: LIFXUpdateCoordinator = hass.data[DOMAIN][entry.entry_id]
+
+    if lifx_features(coordinator.device)["hev"]:
+        async_add_entities(
+            [
+                LIFXBinarySensorEntity(
+                    coordinator=coordinator, description=HEV_CYCLE_STATE_SENSOR
+                )
+            ]
+        )
+
+
+class LIFXBinarySensorEntity(LIFXEntity, BinarySensorEntity):
+    """LIFX sensor entity base class."""
+
+    _attr_has_entity_name = True
+
+    def __init__(
+        self,
+        coordinator: LIFXUpdateCoordinator,
+        description: BinarySensorEntityDescription,
+    ) -> None:
+        """Initialise the sensor."""
+        super().__init__(coordinator)
+
+        self.entity_description = description
+        self._attr_name = description.name
+        self._attr_unique_id = f"{coordinator.serial_number}_{description.key}"
+        self._async_update_attrs()
+
+    @callback
+    def _handle_coordinator_update(self) -> None:
+        """Handle updated data from the coordinator."""
+        self._async_update_attrs()
+        super()._handle_coordinator_update()
+
+    @callback
+    def _async_update_attrs(self) -> None:
+        """Handle coordinator updates."""
+        self._attr_is_on = self.coordinator.async_get_hev_cycle_state()

homeassistant/components/lifx/const.py

@@ -29,6 +29,15 @@ IDENTIFY_WAVEFORM = {
 IDENTIFY = "identify"
 RESTART = "restart"
 
+ATTR_DURATION = "duration"
+ATTR_INDICATION = "indication"
+ATTR_INFRARED = "infrared"
+ATTR_POWER = "power"
+ATTR_REMAINING = "remaining"
+ATTR_ZONES = "zones"
+
+HEV_CYCLE_STATE = "hev_cycle_state"
+
 DATA_LIFX_MANAGER = "lifx_manager"
 
-_LOGGER = logging.getLogger(__name__)
+_LOGGER = logging.getLogger(__package__)

homeassistant/components/lifx/coordinator.py

@@ -15,6 +15,7 @@ from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, Upda
 
 from .const import (
     _LOGGER,
+    ATTR_REMAINING,
     IDENTIFY_WAVEFORM,
     MESSAGE_RETRIES,
     MESSAGE_TIMEOUT,
@@ -101,26 +102,25 @@ class LIFXUpdateCoordinator(DataUpdateCoordinator):
                 self.device.get_hostfirmware()
             if self.device.product is None:
                 self.device.get_version()
-            try:
             response = await async_execute_lifx(self.device.get_color)
-            except asyncio.TimeoutError as ex:
-                raise UpdateFailed(
-                    f"Failed to fetch state from device: {self.device.ip_addr}"
-                ) from ex
+
             if self.device.product is None:
                 raise UpdateFailed(
                     f"Failed to fetch get version from device: {self.device.ip_addr}"
                 )
+
             # device.mac_addr is not the mac_address, its the serial number
             if self.device.mac_addr == TARGET_ANY:
                 self.device.mac_addr = response.target_addr
+
             if lifx_features(self.device)["multizone"]:
-                try:
                 await self.async_update_color_zones()
-                except asyncio.TimeoutError as ex:
-                    raise UpdateFailed(
-                        f"Failed to fetch zones from device: {self.device.ip_addr}"
-                    ) from ex
+
+            if lifx_features(self.device)["hev"]:
+                if self.device.hev_cycle_configuration is None:
+                    self.device.get_hev_configuration()
+
+                await self.async_get_hev_cycle()
 
     async def async_update_color_zones(self) -> None:
         """Get updated color information for each zone."""
@@ -138,6 +138,17 @@ class LIFXUpdateCoordinator(DataUpdateCoordinator):
             if zone == top - 1:
                 zone -= 1
 
+    def async_get_hev_cycle_state(self) -> bool | None:
+        """Return the current HEV cycle state."""
+        if self.device.hev_cycle is None:
+            return None
+        return bool(self.device.hev_cycle.get(ATTR_REMAINING, 0) > 0)
+
+    async def async_get_hev_cycle(self) -> None:
+        """Update the HEV cycle status from a LIFX Clean bulb."""
+        if lifx_features(self.device)["hev"]:
+            await async_execute_lifx(self.device.get_hev_cycle)
+
     async def async_set_waveform_optional(
         self, value: dict[str, Any], rapid: bool = False
     ) -> None:

homeassistant/components/lifx/light.py

@@ -28,7 +28,7 @@ from homeassistant.helpers.entity_platform import AddEntitiesCallback
 from homeassistant.helpers.event import async_track_point_in_utc_time
 import homeassistant.util.color as color_util
 
-from .const import DATA_LIFX_MANAGER, DOMAIN
+from .const import ATTR_INFRARED, ATTR_POWER, ATTR_ZONES, DATA_LIFX_MANAGER, DOMAIN
 from .coordinator import LIFXUpdateCoordinator
 from .entity import LIFXEntity
 from .manager import (
@@ -39,14 +39,8 @@ from .manager import (
 )
 from .util import convert_8_to_16, convert_16_to_8, find_hsbk, lifx_features, merge_hsbk
 
-SERVICE_LIFX_SET_STATE = "set_state"
-
 COLOR_ZONE_POPULATE_DELAY = 0.3
 
-ATTR_INFRARED = "infrared"
-ATTR_ZONES = "zones"
-ATTR_POWER = "power"
-
 SERVICE_LIFX_SET_STATE = "set_state"
 
 LIFX_SET_STATE_SCHEMA = cv.make_entity_service_schema(

tests/components/lifx/__init__.py

@@ -22,10 +22,13 @@ DEFAULT_ENTRY_TITLE = LABEL
 class MockMessage:
     """Mock a lifx message."""
 
-    def __init__(self):
+    def __init__(self, **kwargs):
         """Init message."""
         self.target_addr = SERIAL
         self.count = 9
+        for k, v in kwargs.items():
+            if k != "callb":
+                setattr(self, k, v)
 
 
 class MockFailingLifxCommand:
@@ -50,15 +53,20 @@ class MockFailingLifxCommand:
 class MockLifxCommand:
     """Mock a lifx command."""
 
+    def __name__(self):
+        """Return name."""
+        return "mock_lifx_command"
+
     def __init__(self, bulb, **kwargs):
         """Init command."""
         self.bulb = bulb
         self.calls = []
+        self.msg_kwargs = kwargs
 
     def __call__(self, *args, **kwargs):
         """Call command."""
         if callb := kwargs.get("callb"):
-            callb(self.bulb, MockMessage())
+            callb(self.bulb, MockMessage(**self.msg_kwargs))
         self.calls.append([args, kwargs])
 
     def reset_mock(self):
@@ -108,6 +116,20 @@ def _mocked_brightness_bulb() -> Light:
     return bulb
 
 
+def _mocked_clean_bulb() -> Light:
+    bulb = _mocked_bulb()
+    bulb.get_hev_cycle = MockLifxCommand(
+        bulb, duration=7200, remaining=0, last_power=False
+    )
+    bulb.hev_cycle = {
+        "duration": 7200,
+        "remaining": 30,
+        "last_power": False,
+    }
+    bulb.product = 90
+    return bulb
+
+
 def _mocked_light_strip() -> Light:
     bulb = _mocked_bulb()
     bulb.product = 31  # LIFX Z

tests/components/lifx/test_binary_sensor.py

@@ -0,0 +1,74 @@
+"""Test the lifx binary sensor platwform."""
+from __future__ import annotations
+
+from datetime import timedelta
+
+from homeassistant.components import lifx
+from homeassistant.components.binary_sensor import BinarySensorDeviceClass
+from homeassistant.const import (
+    ATTR_DEVICE_CLASS,
+    CONF_HOST,
+    STATE_OFF,
+    STATE_ON,
+    STATE_UNKNOWN,
+)
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers import entity_registry as er
+from homeassistant.helpers.entity import EntityCategory
+from homeassistant.setup import async_setup_component
+from homeassistant.util import dt as dt_util
+
+from . import (
+    DEFAULT_ENTRY_TITLE,
+    IP_ADDRESS,
+    MAC_ADDRESS,
+    SERIAL,
+    _mocked_clean_bulb,
+    _patch_config_flow_try_connect,
+    _patch_device,
+    _patch_discovery,
+)
+
+from tests.common import MockConfigEntry, async_fire_time_changed
+
+
+async def test_hev_cycle_state(hass: HomeAssistant) -> None:
+    """Test HEV cycle state binary sensor."""
+    config_entry = MockConfigEntry(
+        domain=lifx.DOMAIN,
+        title=DEFAULT_ENTRY_TITLE,
+        data={CONF_HOST: IP_ADDRESS},
+        unique_id=MAC_ADDRESS,
+    )
+    config_entry.add_to_hass(hass)
+    bulb = _mocked_clean_bulb()
+    with _patch_discovery(device=bulb), _patch_config_flow_try_connect(
+        device=bulb
+    ), _patch_device(device=bulb):
+        await async_setup_component(hass, lifx.DOMAIN, {lifx.DOMAIN: {}})
+        await hass.async_block_till_done()
+
+    entity_id = "binary_sensor.my_bulb_clean_cycle"
+    entity_registry = er.async_get(hass)
+
+    state = hass.states.get(entity_id)
+    assert state
+    assert state.state == STATE_ON
+    assert state.attributes.get(ATTR_DEVICE_CLASS) == BinarySensorDeviceClass.RUNNING
+
+    entry = entity_registry.async_get(entity_id)
+    assert state
+    assert entry.unique_id == f"{SERIAL}_hev_cycle_state"
+    assert entry.entity_category == EntityCategory.DIAGNOSTIC
+
+    bulb.hev_cycle = {"duration": 7200, "remaining": 0, "last_power": False}
+
+    async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=30))
+    await hass.async_block_till_done()
+    assert hass.states.get(entity_id).state == STATE_OFF
+
+    bulb.hev_cycle = None
+
+    async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=30))
+    await hass.async_block_till_done()
+    assert hass.states.get(entity_id).state == STATE_UNKNOWN