Add an RSSI sensor to the LIFX integration (home-assistant#80993)

homeassistant/components/lifx/__init__.py

@@ -57,7 +57,13 @@
 )
 
 
-PLATFORMS = [Platform.BINARY_SENSOR, Platform.BUTTON, Platform.LIGHT, Platform.SELECT]
+PLATFORMS = [
+    Platform.BINARY_SENSOR,
+    Platform.BUTTON,
+    Platform.LIGHT,
+    Platform.SELECT,
+    Platform.SENSOR,
+]
 DISCOVERY_INTERVAL = timedelta(minutes=15)
 MIGRATION_INTERVAL = timedelta(minutes=5)
 
@@ -199,6 +205,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
     coordinator.async_setup()
     try:
         await coordinator.async_config_entry_first_refresh()
+        await coordinator.sensor_coordinator.async_config_entry_first_refresh()
     except ConfigEntryNotReady:
         connection.async_stop()
         raise

homeassistant/components/lifx/binary_sensor.py

@@ -12,8 +12,8 @@
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
 
 from .const import DOMAIN, HEV_CYCLE_STATE
-from .coordinator import LIFXUpdateCoordinator
-from .entity import LIFXEntity
+from .coordinator import LIFXSensorUpdateCoordinator, LIFXUpdateCoordinator
+from .entity import LIFXSensorEntity
 from .util import lifx_features
 
 HEV_CYCLE_STATE_SENSOR = BinarySensorEntityDescription(
@@ -34,28 +34,28 @@ async def async_setup_entry(
         async_add_entities(
             [
                 LIFXHevCycleBinarySensorEntity(
-                    coordinator=coordinator, description=HEV_CYCLE_STATE_SENSOR
+                    coordinator=coordinator.sensor_coordinator,
+                    description=HEV_CYCLE_STATE_SENSOR,
                 )
             ]
         )
 
 
-class LIFXHevCycleBinarySensorEntity(LIFXEntity, BinarySensorEntity):
+class LIFXHevCycleBinarySensorEntity(LIFXSensorEntity, BinarySensorEntity):
     """LIFX HEV cycle state binary sensor."""
 
     _attr_has_entity_name = True
 
     def __init__(
         self,
-        coordinator: LIFXUpdateCoordinator,
+        coordinator: LIFXSensorUpdateCoordinator,
         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._attr_unique_id = f"{coordinator.parent.serial_number}_{description.key}"
         self._async_update_attrs()
 
     @callback

homeassistant/components/lifx/button.py

@@ -12,8 +12,8 @@
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
 
 from .const import DOMAIN, IDENTIFY, RESTART
-from .coordinator import LIFXUpdateCoordinator
-from .entity import LIFXEntity
+from .coordinator import LIFXSensorUpdateCoordinator, LIFXUpdateCoordinator
+from .entity import LIFXSensorEntity
 
 RESTART_BUTTON_DESCRIPTION = ButtonEntityDescription(
     key=RESTART,
@@ -38,20 +38,22 @@ async def async_setup_entry(
     domain_data = hass.data[DOMAIN]
     coordinator: LIFXUpdateCoordinator = domain_data[entry.entry_id]
     async_add_entities(
-        cls(coordinator) for cls in (LIFXRestartButton, LIFXIdentifyButton)
+        cls(coordinator.sensor_coordinator)
+        for cls in (LIFXRestartButton, LIFXIdentifyButton)
     )
 
 
-class LIFXButton(LIFXEntity, ButtonEntity):
+class LIFXButton(LIFXSensorEntity, ButtonEntity):
     """Base LIFX button."""
 
     _attr_has_entity_name: bool = True
+    _attr_should_poll: bool = False
 
-    def __init__(self, coordinator: LIFXUpdateCoordinator) -> None:
+    def __init__(self, coordinator: LIFXSensorUpdateCoordinator) -> None:
         """Initialise a LIFX button."""
         super().__init__(coordinator)
         self._attr_unique_id = (
-            f"{coordinator.serial_number}_{self.entity_description.key}"
+            f"{coordinator.parent.serial_number}_{self.entity_description.key}"
         )
 
 

homeassistant/components/lifx/const.py

@@ -35,6 +35,7 @@
 ATTR_INFRARED = "infrared"
 ATTR_POWER = "power"
 ATTR_REMAINING = "remaining"
+ATTR_RSSI = "rssi"
 ATTR_ZONES = "zones"
 
 ATTR_THEME = "theme"

homeassistant/components/lifx/coordinator.py

@@ -2,9 +2,11 @@
 from __future__ import annotations
 
 import asyncio
+from collections.abc import Callable
 from datetime import timedelta
 from enum import IntEnum
 from functools import partial
+from math import floor, log10
 from typing import Any, cast
 
 from aiolifx.aiolifx import (
@@ -15,8 +17,13 @@
 )
 from aiolifx.connection import LIFXConnection
 from aiolifx_themes.themes import ThemeLibrary, ThemePainter
+from awesomeversion import AwesomeVersion
 
-from homeassistant.const import Platform
+from homeassistant.const import (
+    SIGNAL_STRENGTH_DECIBELS,
+    SIGNAL_STRENGTH_DECIBELS_MILLIWATT,
+    Platform,
+)
 from homeassistant.core import HomeAssistant, callback
 from homeassistant.exceptions import HomeAssistantError
 from homeassistant.helpers import entity_registry as er
@@ -41,8 +48,11 @@
     lifx_features,
 )
 
+LIGHT_UPDATE_INTERVAL = 10
+SENSOR_UPDATE_INTERVAL = 30
 REQUEST_REFRESH_DELAY = 0.35
 LIFX_IDENTIFY_DELAY = 3.0
+RSSI_DBM_FW = AwesomeVersion("2.77")
 
 
 class FirmwareEffect(IntEnum):
@@ -69,14 +79,13 @@ def __init__(
         self.device: Light = connection.device
         self.lock = asyncio.Lock()
         self.active_effect = FirmwareEffect.OFF
-        update_interval = timedelta(seconds=10)
-        self.last_used_theme: str = ""
+        self.sensor_coordinator = LIFXSensorUpdateCoordinator(hass, self, title)
 
         super().__init__(
             hass,
             _LOGGER,
             name=f"{title} ({self.device.ip_addr})",
-            update_interval=update_interval,
+            update_interval=timedelta(seconds=LIGHT_UPDATE_INTERVAL),
             # We don't want an immediate refresh since the device
             # takes a moment to reflect the state change
             request_refresh_debouncer=Debouncer(
@@ -112,11 +121,6 @@ def label(self) -> str:
         """Return the label of the bulb."""
         return cast(str, self.device.label)
 
-    @property
-    def current_infrared_brightness(self) -> str | None:
-        """Return the current infrared brightness as a string."""
-        return infrared_brightness_value_to_option(self.device.infrared_brightness)
-
     async def diagnostics(self) -> dict[str, Any]:
         """Return diagnostic information about the device."""
         features = lifx_features(self.device)
@@ -162,19 +166,6 @@ def async_get_entity_id(self, platform: Platform, key: str) -> str | None:
             platform, DOMAIN, f"{self.serial_number}_{key}"
         )
 
-    async def async_identify_bulb(self) -> None:
-        """Identify the device by flashing it three times."""
-        bulb: Light = self.device
-        if bulb.power_level:
-            # just flash the bulb for three seconds
-            await self.async_set_waveform_optional(value=IDENTIFY_WAVEFORM)
-            return
-        # Turn the bulb on first, flash for 3 seconds, then turn off
-        await self.async_set_power(state=True, duration=1)
-        await self.async_set_waveform_optional(value=IDENTIFY_WAVEFORM)
-        await asyncio.sleep(LIFX_IDENTIFY_DELAY)
-        await self.async_set_power(state=False, duration=1)
-
     async def _async_update_data(self) -> None:
         """Fetch all device data from the api."""
         async with self.lock:
@@ -203,12 +194,6 @@ async def _async_update_data(self) -> None:
                 await self.async_get_color_zones()
                 await self.async_get_multizone_effect()
 
-            if lifx_features(self.device)["hev"]:
-                await self.async_get_hev_cycle()
-
-            if lifx_features(self.device)["infrared"]:
-                response = await async_execute_lifx(self.device.get_infrared)
-
     async def async_get_color_zones(self) -> None:
         """Get updated color information for each zone."""
         zone = 0
@@ -234,17 +219,6 @@ async def async_get_extended_color_zones(self) -> None:
                 f"Timeout getting color zones from {self.name}"
             ) from ex
 
-    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:
@@ -381,20 +355,118 @@ def async_get_active_effect(self) -> int:
         """Return the enum value of the currently active firmware effect."""
         return self.active_effect.value
 
-    async def async_set_hev_cycle_state(self, enable: bool, duration: int = 0) -> None:
-        """Start or stop an HEV cycle on a LIFX Clean bulb."""
+
+class LIFXSensorUpdateCoordinator(DataUpdateCoordinator):
+    """DataUpdateCoordinator to gather data for a specific lifx device."""
+
+    def __init__(
+        self,
+        hass: HomeAssistant,
+        parent: LIFXUpdateCoordinator,
+        title: str,
+    ) -> None:
+        """Initialize DataUpdateCoordinator."""
+        self.parent: LIFXUpdateCoordinator = parent
+        self.device: Light = parent.device
+        self._update_rssi: bool = False
+        self._rssi: int = 0
+        self.last_used_theme: str = ""
+
+        super().__init__(
+            hass,
+            _LOGGER,
+            name=f"{title} Sensors ({self.device.ip_addr})",
+            update_interval=timedelta(seconds=SENSOR_UPDATE_INTERVAL),
+            # Refresh immediately because the changes are not visible
+            request_refresh_debouncer=Debouncer(
+                hass, _LOGGER, cooldown=0, immediate=True
+            ),
+        )
+
+    @property
+    def rssi(self) -> int:
+        """Return stored RSSI value."""
+        return self._rssi
+
+    @property
+    def rssi_uom(self) -> str:
+        """Return the RSSI unit of measurement."""
+        if AwesomeVersion(self.device.host_firmware_version) <= RSSI_DBM_FW:
+            return SIGNAL_STRENGTH_DECIBELS
+
+        return SIGNAL_STRENGTH_DECIBELS_MILLIWATT
+
+    @property
+    def current_infrared_brightness(self) -> str | None:
+        """Return the current infrared brightness as a string."""
+        return infrared_brightness_value_to_option(self.device.infrared_brightness)
+
+    async def _async_update_data(self) -> None:
+        """Fetch all device data from the api."""
+
+        if self._update_rssi is True:
+            await self.async_update_rssi()
+
         if lifx_features(self.device)["hev"]:
-            await async_execute_lifx(
-                partial(self.device.set_hev_cycle, enable=enable, duration=duration)
-            )
+            await self.async_get_hev_cycle()
+
+        if lifx_features(self.device)["infrared"]:
+            await async_execute_lifx(self.device.get_infrared)
 
     async def async_set_infrared_brightness(self, option: str) -> None:
         """Set infrared brightness."""
         infrared_brightness = infrared_brightness_option_to_value(option)
         await async_execute_lifx(partial(self.device.set_infrared, infrared_brightness))
 
+    async def async_identify_bulb(self) -> None:
+        """Identify the device by flashing it three times."""
+        bulb: Light = self.device
+        if bulb.power_level:
+            # just flash the bulb for three seconds
+            await self.parent.async_set_waveform_optional(value=IDENTIFY_WAVEFORM)
+            return
+        # Turn the bulb on first, flash for 3 seconds, then turn off
+        await self.parent.async_set_power(state=True, duration=1)
+        await self.parent.async_set_waveform_optional(value=IDENTIFY_WAVEFORM)
+        await asyncio.sleep(LIFX_IDENTIFY_DELAY)
+        await self.parent.async_set_power(state=False, duration=1)
+
+    def async_enable_rssi_updates(self) -> Callable[[], None]:
+        """Enable RSSI signal strength updates."""
+
+        @callback
+        def _async_disable_rssi_updates() -> None:
+            """Disable RSSI updates when sensor removed."""
+            self._update_rssi = False
+
+        self._update_rssi = True
+        return _async_disable_rssi_updates
+
+    async def async_update_rssi(self) -> None:
+        """Update RSSI value."""
+        resp = await async_execute_lifx(self.device.get_wifiinfo)
+        self._rssi = int(floor(10 * log10(resp.signal) + 0.5))
+
+    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_hev_cycle_state(self, enable: bool, duration: int = 0) -> None:
+        """Start or stop an HEV cycle on a LIFX Clean bulb."""
+        if lifx_features(self.device)["hev"]:
+            await async_execute_lifx(
+                partial(self.device.set_hev_cycle, enable=enable, duration=duration)
+            )
+
     async def async_apply_theme(self, theme_name: str) -> None:
         """Apply the selected theme to the device."""
         self.last_used_theme = theme_name
         theme = ThemeLibrary().get_theme(theme_name)
-        await ThemePainter(self.hass.loop).paint(theme, [self.device])
+        await ThemePainter(self.hass.loop).paint(theme, [self.parent.device])

homeassistant/components/lifx/entity.py

@@ -8,7 +8,7 @@
 from homeassistant.helpers.update_coordinator import CoordinatorEntity
 
 from .const import DOMAIN
-from .coordinator import LIFXUpdateCoordinator
+from .coordinator import LIFXSensorUpdateCoordinator, LIFXUpdateCoordinator
 
 
 class LIFXEntity(CoordinatorEntity[LIFXUpdateCoordinator]):
@@ -26,3 +26,20 @@ def __init__(self, coordinator: LIFXUpdateCoordinator) -> None:
             model=products.product_map.get(self.bulb.product, "LIFX Bulb"),
             sw_version=self.bulb.host_firmware_version,
         )
+
+
+class LIFXSensorEntity(CoordinatorEntity[LIFXSensorUpdateCoordinator]):
+    """Representation of a LIFX sensor entity with a sensor coordinator."""
+
+    def __init__(self, coordinator: LIFXSensorUpdateCoordinator) -> None:
+        """Initialise the sensor."""
+        super().__init__(coordinator)
+        self.bulb = coordinator.parent.device
+        self._attr_device_info = DeviceInfo(
+            identifiers={(DOMAIN, coordinator.parent.serial_number)},
+            connections={(dr.CONNECTION_NETWORK_MAC, coordinator.parent.mac_address)},
+            manufacturer="LIFX",
+            name=coordinator.parent.label,
+            model=products.product_map.get(self.bulb.product, "LIFX Bulb"),
+            sw_version=self.bulb.host_firmware_version,
+        )

homeassistant/components/lifx/light.py

@@ -271,7 +271,9 @@ async def set_hev_cycle_state(
                 "This device does not support setting HEV cycle state"
             )
 
-        await self.coordinator.async_set_hev_cycle_state(power, duration or 0)
+        await self.coordinator.sensor_coordinator.async_set_hev_cycle_state(
+            power, duration or 0
+        )
         await self.update_during_transition(duration or 0)
 
     async def set_power(