Update flux_led for upstream strict typing (#60800)

- Bump library to 0.25.10

- Changelog: https://github.com/Danielhiversen/flux_led/compare/0.25.2...0.25.10

- This is a squashed version of #60554 since that one keeps failing to restore the python env on 3.9
This commit is contained in:
J. Nick Koston 2021-12-01 23:55:06 -10:00 committed by GitHub
parent 653fb5b637
commit da2fb17d94
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
14 changed files with 181 additions and 104 deletions

View File

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

View File

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

View File

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

View File

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

View File

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

View File

@ -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} 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.""" """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 flux_color_mode == FLUX_COLOR_MODE_DIM:
if len(flux_color_modes) > 1: if len(flux_color_modes) > 1:
return COLOR_MODE_WHITE return COLOR_MODE_WHITE

View File

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

View File

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

View File

@ -2,20 +2,19 @@
from __future__ import annotations from __future__ import annotations
import asyncio import asyncio
import datetime
from typing import Callable from typing import Callable
from unittest.mock import AsyncMock, MagicMock, patch from unittest.mock import AsyncMock, MagicMock, patch
from flux_led import DeviceType from flux_led import DeviceType
from flux_led.aio import AIOWifiLedBulb from flux_led.aio import AIOWifiLedBulb
from flux_led.const import ( from flux_led.const import (
ATTR_ID,
ATTR_IPADDR,
ATTR_MODEL,
ATTR_MODEL_DESCRIPTION,
COLOR_MODE_CCT as FLUX_COLOR_MODE_CCT, COLOR_MODE_CCT as FLUX_COLOR_MODE_CCT,
COLOR_MODE_RGB as FLUX_COLOR_MODE_RGB, COLOR_MODE_RGB as FLUX_COLOR_MODE_RGB,
) )
from flux_led.models_db import MODEL_MAP
from flux_led.protocol import LEDENETRawState from flux_led.protocol import LEDENETRawState
from flux_led.scanner import FluxLEDDiscovery
from homeassistant.components import dhcp from homeassistant.components import dhcp
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
@ -23,14 +22,14 @@ from homeassistant.core import HomeAssistant
MODULE = "homeassistant.components.flux_led" MODULE = "homeassistant.components.flux_led"
MODULE_CONFIG_FLOW = "homeassistant.components.flux_led.config_flow" MODULE_CONFIG_FLOW = "homeassistant.components.flux_led.config_flow"
IP_ADDRESS = "127.0.0.1" IP_ADDRESS = "127.0.0.1"
MODEL_NUM_HEX = "0x35"
MODEL = "AZ120444" MODEL = "AZ120444"
MODEL_DESCRIPTION = "RGBW Controller" MODEL_DESCRIPTION = "Bulb RGBCW"
MAC_ADDRESS = "aa:bb:cc:dd:ee:ff" MAC_ADDRESS = "aa:bb:cc:dd:ee:ff"
FLUX_MAC_ADDRESS = "aabbccddeeff" FLUX_MAC_ADDRESS = "aabbccddeeff"
SHORT_MAC_ADDRESS = "ddeeff" SHORT_MAC_ADDRESS = "ddeeff"
DEFAULT_ENTRY_TITLE = f"{MODEL_DESCRIPTION} {SHORT_MAC_ADDRESS}" DEFAULT_ENTRY_TITLE = f"{MODEL_DESCRIPTION} {SHORT_MAC_ADDRESS}"
DEFAULT_ENTRY_TITLE_PARTIAL = f"{MODEL} {SHORT_MAC_ADDRESS}"
DHCP_DISCOVERY = dhcp.DhcpServiceInfo( DHCP_DISCOVERY = dhcp.DhcpServiceInfo(
@ -38,17 +37,26 @@ DHCP_DISCOVERY = dhcp.DhcpServiceInfo(
ip=IP_ADDRESS, ip=IP_ADDRESS,
macaddress=MAC_ADDRESS, macaddress=MAC_ADDRESS,
) )
FLUX_DISCOVERY_PARTIAL = { FLUX_DISCOVERY_PARTIAL = FluxLEDDiscovery(
ATTR_IPADDR: IP_ADDRESS, ipaddr=IP_ADDRESS,
ATTR_MODEL: MODEL, model=MODEL,
ATTR_ID: FLUX_MAC_ADDRESS, id=FLUX_MAC_ADDRESS,
} model_num=None,
FLUX_DISCOVERY = { version_num=None,
ATTR_IPADDR: IP_ADDRESS, firmware_date=None,
ATTR_MODEL: MODEL, model_info=None,
ATTR_ID: FLUX_MAC_ADDRESS, model_description=None,
ATTR_MODEL_DESCRIPTION: MODEL_DESCRIPTION, )
} FLUX_DISCOVERY = FluxLEDDiscovery(
ipaddr=IP_ADDRESS,
model=MODEL,
id=FLUX_MAC_ADDRESS,
model_num=0x25,
version_num=0x04,
firmware_date=datetime.date(2021, 5, 5),
model_info=MODEL,
model_description=MODEL_DESCRIPTION,
)
def _mocked_bulb() -> AIOWifiLedBulb: def _mocked_bulb() -> AIOWifiLedBulb:
@ -85,9 +93,10 @@ def _mocked_bulb() -> AIOWifiLedBulb:
bulb.getWhiteTemperature = MagicMock(return_value=(2700, 128)) bulb.getWhiteTemperature = MagicMock(return_value=(2700, 128))
bulb.brightness = 128 bulb.brightness = 128
bulb.model_num = 0x35 bulb.model_num = 0x35
bulb.model_data = MODEL_MAP[0x35]
bulb.effect = None bulb.effect = None
bulb.speed = 50 bulb.speed = 50
bulb.model = "Smart Bulb (0x35)" bulb.model = "Bulb RGBCW (0x35)"
bulb.version_num = 8 bulb.version_num = 8
bulb.speed_adjust_off = True bulb.speed_adjust_off = True
bulb.rgbwcapable = True bulb.rgbwcapable = True
@ -112,7 +121,8 @@ def _mocked_switch() -> AIOWifiLedBulb:
switch.async_turn_off = AsyncMock() switch.async_turn_off = AsyncMock()
switch.async_turn_on = AsyncMock() switch.async_turn_on = AsyncMock()
switch.model_num = 0x97 switch.model_num = 0x97
switch.model = "Smart Switch (0x97)" switch.model_data = MODEL_MAP[0x97]
switch.model = "Switch (0x97)"
switch.version_num = 0x97 switch.version_num = 0x97
switch.raw_state = LEDENETRawState( switch.raw_state = LEDENETRawState(
0, 0x97, 0, 0x61, 0x97, 50, 255, 0, 0, 50, 8, 0, 0, 0 0, 0x97, 0, 0x61, 0x97, 50, 255, 0, 0, 50, 8, 0, 0, 0

View File

@ -29,7 +29,6 @@ from homeassistant.data_entry_flow import RESULT_TYPE_ABORT, RESULT_TYPE_FORM
from . import ( from . import (
DEFAULT_ENTRY_TITLE, DEFAULT_ENTRY_TITLE,
DEFAULT_ENTRY_TITLE_PARTIAL,
DHCP_DISCOVERY, DHCP_DISCOVERY,
FLUX_DISCOVERY, FLUX_DISCOVERY,
IP_ADDRESS, IP_ADDRESS,
@ -428,7 +427,7 @@ async def test_discovered_by_dhcp_no_udp_response(hass):
assert result2["type"] == "create_entry" assert result2["type"] == "create_entry"
assert result2["data"] == { assert result2["data"] == {
CONF_HOST: IP_ADDRESS, CONF_HOST: IP_ADDRESS,
CONF_NAME: DEFAULT_ENTRY_TITLE_PARTIAL, CONF_NAME: DEFAULT_ENTRY_TITLE,
} }
assert mock_async_setup.called assert mock_async_setup.called
assert mock_async_setup_entry.called assert mock_async_setup_entry.called
@ -509,4 +508,4 @@ async def test_options(hass: HomeAssistant):
assert result2["type"] == "create_entry" assert result2["type"] == "create_entry"
assert result2["data"] == user_input assert result2["data"] == user_input
assert result2["data"] == config_entry.options assert result2["data"] == config_entry.options
assert hass.states.get("light.rgbw_controller_ddeeff") is not None assert hass.states.get("light.bulb_rgbcw_ddeeff") is not None

View File

@ -15,7 +15,6 @@ from homeassistant.util.dt import utcnow
from . import ( from . import (
DEFAULT_ENTRY_TITLE, DEFAULT_ENTRY_TITLE,
DEFAULT_ENTRY_TITLE_PARTIAL,
FLUX_DISCOVERY, FLUX_DISCOVERY,
FLUX_DISCOVERY_PARTIAL, FLUX_DISCOVERY_PARTIAL,
IP_ADDRESS, IP_ADDRESS,
@ -75,7 +74,7 @@ async def test_config_entry_retry(hass: HomeAssistant) -> None:
"discovery,title", "discovery,title",
[ [
(FLUX_DISCOVERY, DEFAULT_ENTRY_TITLE), (FLUX_DISCOVERY, DEFAULT_ENTRY_TITLE),
(FLUX_DISCOVERY_PARTIAL, DEFAULT_ENTRY_TITLE_PARTIAL), (FLUX_DISCOVERY_PARTIAL, "AZ120444 ddeeff"),
], ],
) )
async def test_config_entry_fills_unique_id_with_directed_discovery( async def test_config_entry_fills_unique_id_with_directed_discovery(

View File

@ -84,7 +84,7 @@ async def test_light_unique_id(hass: HomeAssistant) -> None:
await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}}) await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}})
await hass.async_block_till_done() await hass.async_block_till_done()
entity_id = "light.rgbw_controller_ddeeff" entity_id = "light.bulb_rgbcw_ddeeff"
entity_registry = er.async_get(hass) entity_registry = er.async_get(hass)
assert entity_registry.async_get(entity_id).unique_id == MAC_ADDRESS assert entity_registry.async_get(entity_id).unique_id == MAC_ADDRESS
state = hass.states.get(entity_id) state = hass.states.get(entity_id)
@ -104,7 +104,7 @@ async def test_light_goes_unavailable_and_recovers(hass: HomeAssistant) -> None:
await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}}) await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}})
await hass.async_block_till_done() await hass.async_block_till_done()
entity_id = "light.rgbw_controller_ddeeff" entity_id = "light.bulb_rgbcw_ddeeff"
entity_registry = er.async_get(hass) entity_registry = er.async_get(hass)
assert entity_registry.async_get(entity_id).unique_id == MAC_ADDRESS assert entity_registry.async_get(entity_id).unique_id == MAC_ADDRESS
state = hass.states.get(entity_id) state = hass.states.get(entity_id)
@ -136,7 +136,7 @@ async def test_light_no_unique_id(hass: HomeAssistant) -> None:
await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}}) await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}})
await hass.async_block_till_done() await hass.async_block_till_done()
entity_id = "light.rgbw_controller_ddeeff" entity_id = "light.bulb_rgbcw_ddeeff"
entity_registry = er.async_get(hass) entity_registry = er.async_get(hass)
assert entity_registry.async_get(entity_id) is None assert entity_registry.async_get(entity_id) is None
state = hass.states.get(entity_id) state = hass.states.get(entity_id)
@ -194,7 +194,7 @@ async def test_rgb_light(hass: HomeAssistant) -> None:
await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}}) await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}})
await hass.async_block_till_done() await hass.async_block_till_done()
entity_id = "light.rgbw_controller_ddeeff" entity_id = "light.bulb_rgbcw_ddeeff"
state = hass.states.get(entity_id) state = hass.states.get(entity_id)
assert state.state == STATE_ON assert state.state == STATE_ON
@ -300,7 +300,7 @@ async def test_rgb_cct_light(hass: HomeAssistant) -> None:
await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}}) await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}})
await hass.async_block_till_done() await hass.async_block_till_done()
entity_id = "light.rgbw_controller_ddeeff" entity_id = "light.bulb_rgbcw_ddeeff"
state = hass.states.get(entity_id) state = hass.states.get(entity_id)
assert state.state == STATE_ON assert state.state == STATE_ON
@ -419,7 +419,7 @@ async def test_rgbw_light(hass: HomeAssistant) -> None:
await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}}) await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}})
await hass.async_block_till_done() await hass.async_block_till_done()
entity_id = "light.rgbw_controller_ddeeff" entity_id = "light.bulb_rgbcw_ddeeff"
state = hass.states.get(entity_id) state = hass.states.get(entity_id)
assert state.state == STATE_ON assert state.state == STATE_ON
@ -521,7 +521,7 @@ async def test_rgb_or_w_light(hass: HomeAssistant) -> None:
await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}}) await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}})
await hass.async_block_till_done() await hass.async_block_till_done()
entity_id = "light.rgbw_controller_ddeeff" entity_id = "light.bulb_rgbcw_ddeeff"
state = hass.states.get(entity_id) state = hass.states.get(entity_id)
assert state.state == STATE_ON assert state.state == STATE_ON
@ -632,7 +632,7 @@ async def test_rgbcw_light(hass: HomeAssistant) -> None:
await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}}) await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}})
await hass.async_block_till_done() await hass.async_block_till_done()
entity_id = "light.rgbw_controller_ddeeff" entity_id = "light.bulb_rgbcw_ddeeff"
state = hass.states.get(entity_id) state = hass.states.get(entity_id)
assert state.state == STATE_ON assert state.state == STATE_ON
@ -772,7 +772,7 @@ async def test_white_light(hass: HomeAssistant) -> None:
await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}}) await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}})
await hass.async_block_till_done() await hass.async_block_till_done()
entity_id = "light.rgbw_controller_ddeeff" entity_id = "light.bulb_rgbcw_ddeeff"
state = hass.states.get(entity_id) state = hass.states.get(entity_id)
assert state.state == STATE_ON assert state.state == STATE_ON
@ -806,6 +806,46 @@ async def test_white_light(hass: HomeAssistant) -> None:
bulb.async_set_brightness.reset_mock() bulb.async_set_brightness.reset_mock()
async def test_no_color_modes(hass: HomeAssistant) -> None:
"""Test a light that has no color modes defined in the database."""
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.mode = "ww"
bulb.protocol = None
bulb.color_modes = set()
bulb.color_mode = None
with _patch_discovery(device=bulb), _patch_wifibulb(device=bulb):
await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}})
await hass.async_block_till_done()
entity_id = "light.bulb_rgbcw_ddeeff"
state = hass.states.get(entity_id)
assert state.state == STATE_ON
attributes = state.attributes
assert attributes[ATTR_COLOR_MODE] == "onoff"
assert ATTR_EFFECT_LIST in attributes # single channel now supports effects
await hass.services.async_call(
LIGHT_DOMAIN, "turn_off", {ATTR_ENTITY_ID: entity_id}, blocking=True
)
bulb.async_turn_off.assert_called_once()
await async_mock_device_turn_off(hass, bulb)
assert hass.states.get(entity_id).state == STATE_OFF
await hass.services.async_call(
LIGHT_DOMAIN, "turn_on", {ATTR_ENTITY_ID: entity_id}, blocking=True
)
bulb.async_turn_on.assert_called_once()
bulb.async_turn_on.reset_mock()
async def test_rgb_light_custom_effects(hass: HomeAssistant) -> None: async def test_rgb_light_custom_effects(hass: HomeAssistant) -> None:
"""Test an rgb light with a custom effect.""" """Test an rgb light with a custom effect."""
config_entry = MockConfigEntry( config_entry = MockConfigEntry(
@ -827,7 +867,7 @@ async def test_rgb_light_custom_effects(hass: HomeAssistant) -> None:
await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}}) await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}})
await hass.async_block_till_done() await hass.async_block_till_done()
entity_id = "light.rgbw_controller_ddeeff" entity_id = "light.bulb_rgbcw_ddeeff"
state = hass.states.get(entity_id) state = hass.states.get(entity_id)
assert state.state == STATE_ON assert state.state == STATE_ON
@ -909,7 +949,7 @@ async def test_rgb_light_custom_effects_invalid_colors(
await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}}) await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}})
await hass.async_block_till_done() await hass.async_block_till_done()
entity_id = "light.rgbw_controller_ddeeff" entity_id = "light.bulb_rgbcw_ddeeff"
state = hass.states.get(entity_id) state = hass.states.get(entity_id)
assert state.state == STATE_ON assert state.state == STATE_ON
@ -938,7 +978,7 @@ async def test_rgb_light_custom_effect_via_service(
await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}}) await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}})
await hass.async_block_till_done() await hass.async_block_till_done()
entity_id = "light.rgbw_controller_ddeeff" entity_id = "light.bulb_rgbcw_ddeeff"
state = hass.states.get(entity_id) state = hass.states.get(entity_id)
assert state.state == STATE_ON assert state.state == STATE_ON
@ -1083,7 +1123,7 @@ async def test_addressable_light(hass: HomeAssistant) -> None:
await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}}) await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}})
await hass.async_block_till_done() await hass.async_block_till_done()
entity_id = "light.rgbw_controller_ddeeff" entity_id = "light.bulb_rgbcw_ddeeff"
state = hass.states.get(entity_id) state = hass.states.get(entity_id)
assert state.state == STATE_ON assert state.state == STATE_ON

View File

@ -45,7 +45,7 @@ async def test_number_unique_id(hass: HomeAssistant) -> None:
await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}}) await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}})
await hass.async_block_till_done() await hass.async_block_till_done()
entity_id = "number.rgbw_controller_ddeeff_effect_speed" entity_id = "number.bulb_rgbcw_ddeeff_effect_speed"
entity_registry = er.async_get(hass) entity_registry = er.async_get(hass)
assert entity_registry.async_get(entity_id).unique_id == MAC_ADDRESS assert entity_registry.async_get(entity_id).unique_id == MAC_ADDRESS
@ -70,8 +70,8 @@ async def test_rgb_light_effect_speed(hass: HomeAssistant) -> None:
await async_mock_device_turn_on(hass, bulb) await async_mock_device_turn_on(hass, bulb)
light_entity_id = "light.rgbw_controller_ddeeff" light_entity_id = "light.bulb_rgbcw_ddeeff"
number_entity_id = "number.rgbw_controller_ddeeff_effect_speed" number_entity_id = "number.bulb_rgbcw_ddeeff_effect_speed"
with pytest.raises(HomeAssistantError): with pytest.raises(HomeAssistantError):
await hass.services.async_call( await hass.services.async_call(
NUMBER_DOMAIN, NUMBER_DOMAIN,
@ -135,8 +135,8 @@ async def test_original_addressable_light_effect_speed(hass: HomeAssistant) -> N
await async_mock_device_turn_on(hass, bulb) await async_mock_device_turn_on(hass, bulb)
light_entity_id = "light.rgbw_controller_ddeeff" light_entity_id = "light.bulb_rgbcw_ddeeff"
number_entity_id = "number.rgbw_controller_ddeeff_effect_speed" number_entity_id = "number.bulb_rgbcw_ddeeff_effect_speed"
state = hass.states.get(light_entity_id) state = hass.states.get(light_entity_id)
assert state.state == STATE_ON assert state.state == STATE_ON
@ -192,8 +192,8 @@ async def test_addressable_light_effect_speed(hass: HomeAssistant) -> None:
await async_mock_device_turn_on(hass, bulb) await async_mock_device_turn_on(hass, bulb)
light_entity_id = "light.rgbw_controller_ddeeff" light_entity_id = "light.bulb_rgbcw_ddeeff"
number_entity_id = "number.rgbw_controller_ddeeff_effect_speed" number_entity_id = "number.bulb_rgbcw_ddeeff_effect_speed"
state = hass.states.get(light_entity_id) state = hass.states.get(light_entity_id)
assert state.state == STATE_ON assert state.state == STATE_ON

View File

@ -39,7 +39,7 @@ async def test_switch_on_off(hass: HomeAssistant) -> None:
await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}}) await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}})
await hass.async_block_till_done() await hass.async_block_till_done()
entity_id = "switch.rgbw_controller_ddeeff" entity_id = "switch.bulb_rgbcw_ddeeff"
state = hass.states.get(entity_id) state = hass.states.get(entity_id)
assert state.state == STATE_ON assert state.state == STATE_ON