Update flux_led for upstream strict typing (home-assistant#60800)

- Bump library to 0.25.10

- Changelog: lightinglibs/flux_led@0.25.2...0.25.10

- This is a squashed version of home-assistant#60554 since that one keeps failing to restore the python env on 3.9

homeassistant/components/flux_led/__init__.py

@@ -4,12 +4,13 @@
 import asyncio
 from datetime import timedelta
 import logging
-from typing import Any, Final, cast
+from typing import Any, Final
 
 from flux_led import DeviceType
 from flux_led.aio import AIOWifiLedBulb
 from flux_led.aioscanner import AIOBulbScanner
 from flux_led.const import ATTR_ID, ATTR_IPADDR, ATTR_MODEL, ATTR_MODEL_DESCRIPTION
+from flux_led.scanner import FluxLEDDiscovery
 
 from homeassistant import config_entries
 from homeassistant.config_entries import ConfigEntry
@@ -50,43 +51,44 @@ def async_wifi_bulb_for_host(host: str) -> AIOWifiLedBulb:
 
 
 @callback
-def async_name_from_discovery(device: dict[str, Any]) -> str:
+def async_name_from_discovery(device: FluxLEDDiscovery) -> str:
     """Convert a flux_led discovery to a human readable name."""
-    if (mac := device.get(ATTR_ID)) is None:
-        return cast(str, device[ATTR_IPADDR])
-    short_mac = mac[-6:]
-    if device.get(ATTR_MODEL_DESCRIPTION):
+    mac_address = device[ATTR_ID]
+    if mac_address is None:
+        return device[ATTR_IPADDR]
+    short_mac = mac_address[-6:]
+    if device[ATTR_MODEL_DESCRIPTION]:
         return f"{device[ATTR_MODEL_DESCRIPTION]} {short_mac}"
     return f"{device[ATTR_MODEL]} {short_mac}"
 
 
 @callback
 def async_update_entry_from_discovery(
-    hass: HomeAssistant, entry: config_entries.ConfigEntry, device: dict[str, Any]
+    hass: HomeAssistant, entry: config_entries.ConfigEntry, device: FluxLEDDiscovery
 ) -> None:
     """Update a config entry from a flux_led discovery."""
     name = async_name_from_discovery(device)
+    mac_address = device[ATTR_ID]
+    assert mac_address is not None
     hass.config_entries.async_update_entry(
         entry,
         data={**entry.data, CONF_NAME: name},
         title=name,
-        unique_id=dr.format_mac(device[ATTR_ID]),
+        unique_id=dr.format_mac(mac_address),
     )
 
 
 async def async_discover_devices(
     hass: HomeAssistant, timeout: int, address: str | None = None
-) -> list[dict[str, str]]:
+) -> list[FluxLEDDiscovery]:
     """Discover flux led devices."""
     domain_data = hass.data.setdefault(DOMAIN, {})
     if FLUX_LED_DISCOVERY_LOCK not in domain_data:
         domain_data[FLUX_LED_DISCOVERY_LOCK] = asyncio.Lock()
     async with domain_data[FLUX_LED_DISCOVERY_LOCK]:
         scanner = AIOBulbScanner()
         try:
-            discovered: list[dict[str, str]] = await scanner.async_scan(
-                timeout=timeout, address=address
-            )
+            discovered = await scanner.async_scan(timeout=timeout, address=address)
         except OSError as ex:
             _LOGGER.debug("Scanning failed with error: %s", ex)
             return []
@@ -96,7 +98,7 @@ async def async_discover_devices(
 
 async def async_discover_device(
     hass: HomeAssistant, host: str
-) -> dict[str, str] | None:
+) -> FluxLEDDiscovery | None:
     """Direct discovery at a single ip instead of broadcast."""
     # If we are missing the unique_id we should be able to fetch it
     # from the device by doing a directed discovery at the host only
@@ -109,15 +111,15 @@ async def async_discover_device(
 @callback
 def async_trigger_discovery(
     hass: HomeAssistant,
-    discovered_devices: list[dict[str, Any]],
+    discovered_devices: list[FluxLEDDiscovery],
 ) -> None:
     """Trigger config flows for discovered devices."""
     for device in discovered_devices:
         hass.async_create_task(
             hass.config_entries.flow.async_init(
                 DOMAIN,
                 context={"source": config_entries.SOURCE_DISCOVERY},
-                data=device,
+                data={**device},
             )
         )
 

homeassistant/components/flux_led/config_flow.py

@@ -2,9 +2,10 @@
 from __future__ import annotations
 
 import logging
-from typing import Any, Final
+from typing import Any, Final, cast
 
 from flux_led.const import ATTR_ID, ATTR_IPADDR, ATTR_MODEL, ATTR_MODEL_DESCRIPTION
+from flux_led.scanner import FluxLEDDiscovery
 import voluptuous as vol
 
 from homeassistant import config_entries
@@ -48,8 +49,8 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
 
     def __init__(self) -> None:
         """Initialize the config flow."""
-        self._discovered_devices: dict[str, dict[str, Any]] = {}
-        self._discovered_device: dict[str, Any] = {}
+        self._discovered_devices: dict[str, FluxLEDDiscovery] = {}
+        self._discovered_device: FluxLEDDiscovery | None = None
 
     @staticmethod
     @callback
@@ -84,24 +85,32 @@ async def async_step_import(self, user_input: dict[str, Any]) -> FlowResult:
 
     async def async_step_dhcp(self, discovery_info: dhcp.DhcpServiceInfo) -> FlowResult:
         """Handle discovery via dhcp."""
-        self._discovered_device = {
-            ATTR_IPADDR: discovery_info.ip,
-            ATTR_MODEL: discovery_info.hostname,
-            ATTR_ID: discovery_info.macaddress.replace(":", ""),
-        }
+        self._discovered_device = FluxLEDDiscovery(
+            ipaddr=discovery_info.ip,
+            model=discovery_info.hostname,
+            id=discovery_info.macaddress.replace(":", ""),
+            model_num=None,
+            version_num=None,
+            firmware_date=None,
+            model_info=None,
+            model_description=None,
+        )
         return await self._async_handle_discovery()
 
     async def async_step_discovery(
         self, discovery_info: DiscoveryInfoType
     ) -> FlowResult:
         """Handle discovery."""
-        self._discovered_device = discovery_info
+        self._discovered_device = cast(FluxLEDDiscovery, discovery_info)
         return await self._async_handle_discovery()
 
     async def _async_handle_discovery(self) -> FlowResult:
         """Handle any discovery."""
         device = self._discovered_device
-        mac = dr.format_mac(device[ATTR_ID])
+        assert device is not None
+        mac_address = device[ATTR_ID]
+        assert mac_address is not None
+        mac = dr.format_mac(mac_address)
         host = device[ATTR_IPADDR]
         await self.async_set_unique_id(mac)
         self._abort_if_unique_id_configured(updates={CONF_HOST: host})
@@ -113,28 +122,33 @@ async def _async_handle_discovery(self) -> FlowResult:
         for progress in self._async_in_progress():
             if progress.get("context", {}).get(CONF_HOST) == host:
                 return self.async_abort(reason="already_in_progress")
-        if not device.get(ATTR_MODEL_DESCRIPTION):
+        if not device[ATTR_MODEL_DESCRIPTION]:
             try:
-                device = await self._async_try_connect(host)
+                device = await self._async_try_connect(
+                    host, device[ATTR_ID], device[ATTR_MODEL]
+                )
             except FLUX_LED_EXCEPTIONS:
                 return self.async_abort(reason="cannot_connect")
             else:
-                if device.get(ATTR_MODEL_DESCRIPTION):
+                if device[ATTR_MODEL_DESCRIPTION]:
                     self._discovered_device = device
         return await self.async_step_discovery_confirm()
 
     async def async_step_discovery_confirm(
         self, user_input: dict[str, Any] | None = None
     ) -> FlowResult:
         """Confirm discovery."""
+        assert self._discovered_device is not None
+        device = self._discovered_device
+        mac_address = device[ATTR_ID]
+        assert mac_address is not None
         if user_input is not None:
             return self._async_create_entry_from_device(self._discovered_device)
 
         self._set_confirm_only()
-        device = self._discovered_device
         placeholders = {
-            "model": device.get(ATTR_MODEL_DESCRIPTION, device[ATTR_MODEL]),
-            "id": device[ATTR_ID][-6:],
+            "model": device[ATTR_MODEL_DESCRIPTION] or device[ATTR_MODEL],
+            "id": mac_address[-6:],
             "ipaddr": device[ATTR_IPADDR],
         }
         self.context["title_placeholders"] = placeholders
@@ -143,7 +157,7 @@ async def async_step_discovery_confirm(
         )
 
     @callback
-    def _async_create_entry_from_device(self, device: dict[str, Any]) -> FlowResult:
+    def _async_create_entry_from_device(self, device: FluxLEDDiscovery) -> FlowResult:
         """Create a config entry from a device."""
         self._async_abort_entries_match({CONF_HOST: device[ATTR_IPADDR]})
         name = async_name_from_discovery(device)
@@ -164,13 +178,14 @@ async def async_step_user(
             if not (host := user_input[CONF_HOST]):
                 return await self.async_step_pick_device()
             try:
-                device = await self._async_try_connect(host)
+                device = await self._async_try_connect(host, None, None)
             except FLUX_LED_EXCEPTIONS:
                 errors["base"] = "cannot_connect"
             else:
-                if device[ATTR_ID]:
+                mac_address = device[ATTR_ID]
+                if mac_address is not None:
                     await self.async_set_unique_id(
-                        dr.format_mac(device[ATTR_ID]), raise_on_progress=False
+                        dr.format_mac(mac_address), raise_on_progress=False
                     )
                     self._abort_if_unique_id_configured(updates={CONF_HOST: host})
                 return self._async_create_entry_from_device(device)
@@ -198,9 +213,11 @@ async def async_step_pick_device(
         discovered_devices = await async_discover_devices(
             self.hass, DISCOVER_SCAN_TIMEOUT
         )
-        self._discovered_devices = {
-            dr.format_mac(device[ATTR_ID]): device for device in discovered_devices
-        }
+        self._discovered_devices = {}
+        for device in discovered_devices:
+            mac_address = device[ATTR_ID]
+            assert mac_address is not None
+            self._discovered_devices[dr.format_mac(mac_address)] = device
         devices_name = {
             mac: f"{async_name_from_discovery(device)} ({device[ATTR_IPADDR]})"
             for mac, device in self._discovered_devices.items()
@@ -215,7 +232,9 @@ async def async_step_pick_device(
             data_schema=vol.Schema({vol.Required(CONF_DEVICE): vol.In(devices_name)}),
         )
 
-    async def _async_try_connect(self, host: str) -> dict[str, Any]:
+    async def _async_try_connect(
+        self, host: str, mac_address: str | None, model: str | None
+    ) -> FluxLEDDiscovery:
         """Try to connect."""
         self._async_abort_entries_match({CONF_HOST: host})
         if device := await async_discover_device(self.hass, host):
@@ -225,7 +244,16 @@ async def _async_try_connect(self, host: str) -> dict[str, Any]:
             await bulb.async_setup(lambda: None)
         finally:
             await bulb.async_stop()
-        return {ATTR_ID: None, ATTR_MODEL: None, ATTR_IPADDR: host}
+        return FluxLEDDiscovery(
+            ipaddr=host,
+            model=model,
+            id=mac_address,
+            model_num=bulb.model_num,
+            version_num=bulb.version_num,
+            firmware_date=None,
+            model_info=None,
+            model_description=bulb.model_data.description,
+        )
 
 
 class OptionsFlow(config_entries.OptionsFlow):

homeassistant/components/flux_led/entity.py

@@ -2,7 +2,7 @@
 from __future__ import annotations
 
 from abc import abstractmethod
-from typing import Any, cast
+from typing import Any
 
 from flux_led.aiodevice import AIOWifiLedBulb
 
@@ -72,7 +72,7 @@ class FluxOnOffEntity(FluxEntity):
     @property
     def is_on(self) -> bool:
         """Return true if device is on."""
-        return cast(bool, self._device.is_on)
+        return self._device.is_on
 
     async def async_turn_on(self, **kwargs: Any) -> None:
         """Turn the specified device on."""

homeassistant/components/flux_led/light.py

@@ -3,7 +3,7 @@
 
 import ast
 import logging
-from typing import Any, Final, cast
+from typing import Any, Final
 
 from flux_led.const import ATTR_ID, ATTR_IPADDR
 from flux_led.utils import (
@@ -244,7 +244,7 @@ def __init__(
     @property
     def brightness(self) -> int:
         """Return the brightness of this light between 0..255."""
-        return cast(int, self._device.brightness)
+        return self._device.brightness
 
     @property
     def color_temp(self) -> int:
@@ -254,20 +254,17 @@ def color_temp(self) -> int:
     @property
     def rgb_color(self) -> tuple[int, int, int]:
         """Return the rgb color value."""
-        rgb: tuple[int, int, int] = self._device.rgb_unscaled
-        return rgb
+        return self._device.rgb_unscaled
 
     @property
     def rgbw_color(self) -> tuple[int, int, int, int]:
         """Return the rgbw color value."""
-        rgbw: tuple[int, int, int, int] = self._device.rgbw
-        return rgbw
+        return self._device.rgbw
 
     @property
     def rgbww_color(self) -> tuple[int, int, int, int, int]:
         """Return the rgbww aka rgbcw color value."""
-        rgbcw: tuple[int, int, int, int, int] = self._device.rgbcw
-        return rgbcw
+        return self._device.rgbcw
 
     @property
     def color_mode(self) -> str:
@@ -279,10 +276,7 @@ def color_mode(self) -> str:
     @property
     def effect(self) -> str | None:
         """Return the current effect."""
-        effect = self._device.effect
-        if effect is None:
-            return None
-        return cast(str, effect)
+        return self._device.effect
 
     async def _async_turn_on(self, **kwargs: Any) -> None:
         """Turn the specified or all lights on."""
@@ -353,7 +347,8 @@ async def _async_set_mode(self, **kwargs: Any) -> None:
             return
         # Handle switch to RGB Color Mode
         if rgb := kwargs.get(ATTR_RGB_COLOR):
-            await self._device.async_set_levels(*rgb, brightness=brightness)
+            red, green, blue = rgb
+            await self._device.async_set_levels(red, green, blue, brightness=brightness)
             return
         # Handle switch to RGBW Color Mode
         if rgbw := kwargs.get(ATTR_RGBW_COLOR):

homeassistant/components/flux_led/manifest.json

@@ -3,7 +3,7 @@
   "name": "Flux LED/MagicHome",
   "config_flow": true,
   "documentation": "https://www.home-assistant.io/integrations/flux_led",
-  "requirements": ["flux_led==0.25.2"],
+  "requirements": ["flux_led==0.25.10"],
   "quality_scale": "platinum",
   "codeowners": ["@icemanch"],
   "iot_class": "local_push",

homeassistant/components/flux_led/util.py

@@ -18,8 +18,12 @@ def _hass_color_modes(device: AIOWifiLedBulb) -> set[str]:
     return {_flux_color_mode_to_hass(mode, color_modes) for mode in color_modes}
 
 
-def _flux_color_mode_to_hass(flux_color_mode: str, flux_color_modes: set[str]) -> str:
+def _flux_color_mode_to_hass(
+    flux_color_mode: str | None, flux_color_modes: set[str]
+) -> str:
     """Map the flux color mode to Home Assistant color mode."""
+    if flux_color_mode is None:
+        return COLOR_MODE_ONOFF
     if flux_color_mode == FLUX_COLOR_MODE_DIM:
         if len(flux_color_modes) > 1:
             return COLOR_MODE_WHITE

requirements_all.txt

@@ -658,7 +658,7 @@ fjaraskupan==1.0.2
 flipr-api==1.4.1
 
 # homeassistant.components.flux_led
-flux_led==0.25.2
+flux_led==0.25.10
 
 # homeassistant.components.homekit
 fnvhash==0.1.0

requirements_test_all.txt

@@ -399,7 +399,7 @@ fjaraskupan==1.0.2
 flipr-api==1.4.1
 
 # homeassistant.components.flux_led
-flux_led==0.25.2
+flux_led==0.25.10
 
 # homeassistant.components.homekit
 fnvhash==0.1.0