Use runtime_data in flux_led (#138279)

pull/138162/head^2
epenet 2025-02-11 17:09:06 +01:00 committed by GitHub
parent 62b563eb60
commit 7f376ff004
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
11 changed files with 50 additions and 62 deletions

View File

@ -11,7 +11,6 @@ from flux_led.aio import AIOWifiLedBulb
from flux_led.const import ATTR_ID, WhiteChannelType
from flux_led.scanner import FluxLEDDiscovery
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_HOST, Platform
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import ConfigEntryNotReady
@ -39,7 +38,7 @@ from .const import (
FLUX_LED_EXCEPTIONS,
SIGNAL_STATE_UPDATED,
)
from .coordinator import FluxLedUpdateCoordinator
from .coordinator import FluxLedConfigEntry, FluxLedUpdateCoordinator
from .discovery import (
async_build_cached_discovery,
async_clear_discovery_cache,
@ -113,7 +112,9 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
return True
async def _async_migrate_unique_ids(hass: HomeAssistant, entry: ConfigEntry) -> None:
async def _async_migrate_unique_ids(
hass: HomeAssistant, entry: FluxLedConfigEntry
) -> None:
"""Migrate entities when the mac address gets discovered."""
@callback
@ -146,14 +147,16 @@ async def _async_migrate_unique_ids(hass: HomeAssistant, entry: ConfigEntry) ->
await er.async_migrate_entries(hass, entry.entry_id, _async_migrator)
async def _async_update_listener(hass: HomeAssistant, entry: ConfigEntry) -> None:
async def _async_update_listener(
hass: HomeAssistant, entry: FluxLedConfigEntry
) -> None:
"""Handle options update."""
coordinator: FluxLedUpdateCoordinator = hass.data[DOMAIN][entry.entry_id]
coordinator = entry.runtime_data
if entry.title != coordinator.title:
await hass.config_entries.async_reload(entry.entry_id)
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
async def async_setup_entry(hass: HomeAssistant, entry: FluxLedConfigEntry) -> bool:
"""Set up Flux LED/MagicLight from a config entry."""
host = entry.data[CONF_HOST]
discovery_cached = True
@ -206,7 +209,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
await _async_migrate_unique_ids(hass, entry)
coordinator = FluxLedUpdateCoordinator(hass, device, entry)
hass.data[DOMAIN][entry.entry_id] = coordinator
entry.runtime_data = coordinator
platforms = PLATFORMS_BY_TYPE[device.device_type]
await hass.config_entries.async_forward_entry_setups(entry, platforms)
@ -239,13 +242,12 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
async def async_unload_entry(hass: HomeAssistant, entry: FluxLedConfigEntry) -> bool:
"""Unload a config entry."""
device: AIOWifiLedBulb = hass.data[DOMAIN][entry.entry_id].device
device = entry.runtime_data.device
platforms = PLATFORMS_BY_TYPE[device.device_type]
if unload_ok := await hass.config_entries.async_unload_platforms(entry, platforms):
# Make sure we probe the device again in case something has changed externally
async_clear_discovery_cache(hass, entry.data[CONF_HOST])
del hass.data[DOMAIN][entry.entry_id]
await device.async_stop()
return unload_ok

View File

@ -5,7 +5,6 @@ from __future__ import annotations
from flux_led.aio import AIOWifiLedBulb
from flux_led.protocol import RemoteConfig
from homeassistant import config_entries
from homeassistant.components.button import (
ButtonDeviceClass,
ButtonEntity,
@ -15,8 +14,7 @@ from homeassistant.const import EntityCategory
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .const import DOMAIN
from .coordinator import FluxLedUpdateCoordinator
from .coordinator import FluxLedConfigEntry
from .entity import FluxBaseEntity
_RESTART_KEY = "restart"
@ -34,11 +32,11 @@ UNPAIR_REMOTES_DESCRIPTION = ButtonEntityDescription(
async def async_setup_entry(
hass: HomeAssistant,
entry: config_entries.ConfigEntry,
entry: FluxLedConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up Magic Home button based on a config entry."""
coordinator: FluxLedUpdateCoordinator = hass.data[DOMAIN][entry.entry_id]
coordinator = entry.runtime_data
device = coordinator.device
entities: list[FluxButton] = [
FluxButton(coordinator.device, entry, RESTART_BUTTON_DESCRIPTION)
@ -59,7 +57,7 @@ class FluxButton(FluxBaseEntity, ButtonEntity):
def __init__(
self,
device: AIOWifiLedBulb,
entry: config_entries.ConfigEntry,
entry: FluxLedConfigEntry,
description: ButtonEntityDescription,
) -> None:
"""Initialize the button."""

View File

@ -18,7 +18,6 @@ import voluptuous as vol
from homeassistant.config_entries import (
SOURCE_IGNORE,
ConfigEntry,
ConfigEntryState,
ConfigFlow,
ConfigFlowResult,
@ -46,6 +45,7 @@ from .const import (
TRANSITION_JUMP,
TRANSITION_STROBE,
)
from .coordinator import FluxLedConfigEntry
from .discovery import (
async_discover_device,
async_discover_devices,
@ -72,7 +72,7 @@ class FluxLedConfigFlow(ConfigFlow, domain=DOMAIN):
@staticmethod
@callback
def async_get_options_flow(
config_entry: ConfigEntry,
config_entry: FluxLedConfigEntry,
) -> FluxLedOptionsFlow:
"""Get the options flow for the Flux LED component."""
return FluxLedOptionsFlow()

View File

@ -20,14 +20,16 @@ _LOGGER = logging.getLogger(__name__)
REQUEST_REFRESH_DELAY: Final = 2.0
type FluxLedConfigEntry = ConfigEntry[FluxLedUpdateCoordinator]
class FluxLedUpdateCoordinator(DataUpdateCoordinator[None]):
"""DataUpdateCoordinator to gather data for a specific flux_led device."""
config_entry: ConfigEntry
config_entry: FluxLedConfigEntry
def __init__(
self, hass: HomeAssistant, device: AIOWifiLedBulb, entry: ConfigEntry
self, hass: HomeAssistant, device: AIOWifiLedBulb, entry: FluxLedConfigEntry
) -> None:
"""Initialize DataUpdateCoordinator to gather data for specific device."""
self.device = device

View File

@ -4,22 +4,19 @@ from __future__ import annotations
from typing import Any
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from .const import DOMAIN
from .coordinator import FluxLedUpdateCoordinator
from .coordinator import FluxLedConfigEntry
async def async_get_config_entry_diagnostics(
hass: HomeAssistant, entry: ConfigEntry
hass: HomeAssistant, entry: FluxLedConfigEntry
) -> dict[str, Any]:
"""Return diagnostics for a config entry."""
coordinator: FluxLedUpdateCoordinator = hass.data[DOMAIN][entry.entry_id]
return {
"entry": {
"title": entry.title,
"data": dict(entry.data),
},
"data": coordinator.device.diagnostics,
"data": entry.runtime_data.device.diagnostics,
}

View File

@ -23,9 +23,8 @@ from flux_led.const import (
from flux_led.models_db import get_model_description
from flux_led.scanner import FluxLEDDiscovery
from homeassistant import config_entries
from homeassistant.components import network
from homeassistant.config_entries import ConfigEntry, ConfigEntryState
from homeassistant.config_entries import SOURCE_INTEGRATION_DISCOVERY, ConfigEntryState
from homeassistant.const import CONF_HOST, CONF_MODEL, CONF_NAME
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import device_registry as dr, discovery_flow
@ -44,6 +43,7 @@ from .const import (
DOMAIN,
FLUX_LED_DISCOVERY,
)
from .coordinator import FluxLedConfigEntry
from .util import format_as_flux_mac, mac_matches_by_one
_LOGGER = logging.getLogger(__name__)
@ -63,7 +63,7 @@ CONF_TO_DISCOVERY: Final = {
@callback
def async_build_cached_discovery(entry: ConfigEntry) -> FluxLEDDiscovery:
def async_build_cached_discovery(entry: FluxLedConfigEntry) -> FluxLEDDiscovery:
"""When discovery is unavailable, load it from the config entry."""
data = entry.data
return FluxLEDDiscovery(
@ -116,7 +116,7 @@ def async_populate_data_from_discovery(
@callback
def async_update_entry_from_discovery(
hass: HomeAssistant,
entry: config_entries.ConfigEntry,
entry: FluxLedConfigEntry,
device: FluxLEDDiscovery,
model_num: int | None,
allow_update_mac: bool,
@ -230,6 +230,6 @@ def async_trigger_discovery(
discovery_flow.async_create_flow(
hass,
DOMAIN,
context={"source": config_entries.SOURCE_INTEGRATION_DISCOVERY},
context={"source": SOURCE_INTEGRATION_DISCOVERY},
data={**device},
)

View File

@ -11,7 +11,6 @@ from flux_led.protocol import MusicMode
from flux_led.utils import rgbcw_brightness, rgbcw_to_rgbwc, rgbw_brightness
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.components.light import (
ATTR_BRIGHTNESS,
ATTR_COLOR_TEMP_KELVIN,
@ -38,7 +37,6 @@ from .const import (
CONF_SPEED_PCT,
CONF_TRANSITION,
DEFAULT_EFFECT_SPEED,
DOMAIN,
MIN_CCT_BRIGHTNESS,
MIN_RGB_BRIGHTNESS,
MULTI_BRIGHTNESS_COLOR_MODES,
@ -46,7 +44,7 @@ from .const import (
TRANSITION_JUMP,
TRANSITION_STROBE,
)
from .coordinator import FluxLedUpdateCoordinator
from .coordinator import FluxLedConfigEntry, FluxLedUpdateCoordinator
from .entity import FluxOnOffEntity
from .util import (
_effect_brightness,
@ -134,11 +132,11 @@ SET_ZONES_DICT: VolDictType = {
async def async_setup_entry(
hass: HomeAssistant,
entry: config_entries.ConfigEntry,
entry: FluxLedConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up the Flux lights."""
coordinator: FluxLedUpdateCoordinator = hass.data[DOMAIN][entry.entry_id]
coordinator = entry.runtime_data
platform = entity_platform.async_get_current_platform()
platform.async_register_entity_service(

View File

@ -16,7 +16,6 @@ from flux_led.protocol import (
SEGMENTS_MAX,
)
from homeassistant import config_entries
from homeassistant.components.light import EFFECT_RANDOM
from homeassistant.components.number import NumberEntity, NumberMode
from homeassistant.const import EntityCategory
@ -26,8 +25,7 @@ from homeassistant.helpers.debounce import Debouncer
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import DOMAIN
from .coordinator import FluxLedUpdateCoordinator
from .coordinator import FluxLedConfigEntry, FluxLedUpdateCoordinator
from .entity import FluxEntity
from .util import _effect_brightness
@ -38,11 +36,11 @@ DEBOUNCE_TIME = 1
async def async_setup_entry(
hass: HomeAssistant,
entry: config_entries.ConfigEntry,
entry: FluxLedConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up the Flux lights."""
coordinator: FluxLedUpdateCoordinator = hass.data[DOMAIN][entry.entry_id]
coordinator = entry.runtime_data
device = coordinator.device
entities: list[
FluxSpeedNumber

View File

@ -13,14 +13,13 @@ from flux_led.const import (
)
from flux_led.protocol import PowerRestoreState, RemoteConfig
from homeassistant import config_entries
from homeassistant.components.select import SelectEntity
from homeassistant.const import CONF_NAME, EntityCategory
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .const import CONF_WHITE_CHANNEL_TYPE, DOMAIN, FLUX_COLOR_MODE_RGBW
from .coordinator import FluxLedUpdateCoordinator
from .const import CONF_WHITE_CHANNEL_TYPE, FLUX_COLOR_MODE_RGBW
from .coordinator import FluxLedConfigEntry, FluxLedUpdateCoordinator
from .entity import FluxBaseEntity, FluxEntity
from .util import _human_readable_option
@ -29,9 +28,7 @@ NAME_TO_POWER_RESTORE_STATE = {
}
async def _async_delayed_reload(
hass: HomeAssistant, entry: config_entries.ConfigEntry
) -> None:
async def _async_delayed_reload(hass: HomeAssistant, entry: FluxLedConfigEntry) -> None:
"""Reload after making a change that will effect the operation of the device."""
await asyncio.sleep(STATE_CHANGE_LATENCY)
hass.async_create_task(hass.config_entries.async_reload(entry.entry_id))
@ -39,11 +36,11 @@ async def _async_delayed_reload(
async def async_setup_entry(
hass: HomeAssistant,
entry: config_entries.ConfigEntry,
entry: FluxLedConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up the Flux selects."""
coordinator: FluxLedUpdateCoordinator = hass.data[DOMAIN][entry.entry_id]
coordinator = entry.runtime_data
device = coordinator.device
entities: list[
FluxPowerStateSelect
@ -97,7 +94,7 @@ class FluxPowerStateSelect(FluxConfigAtStartSelect, SelectEntity):
def __init__(
self,
device: AIOWifiLedBulb,
entry: config_entries.ConfigEntry,
entry: FluxLedConfigEntry,
) -> None:
"""Initialize the power state select."""
super().__init__(device, entry)
@ -228,7 +225,7 @@ class FluxWhiteChannelSelect(FluxConfigAtStartSelect):
def __init__(
self,
device: AIOWifiLedBulb,
entry: config_entries.ConfigEntry,
entry: FluxLedConfigEntry,
) -> None:
"""Initialize the white channel select."""
super().__init__(device, entry)

View File

@ -2,24 +2,22 @@
from __future__ import annotations
from homeassistant import config_entries
from homeassistant.components.sensor import SensorEntity
from homeassistant.const import EntityCategory
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .const import DOMAIN
from .coordinator import FluxLedUpdateCoordinator
from .coordinator import FluxLedConfigEntry
from .entity import FluxEntity
async def async_setup_entry(
hass: HomeAssistant,
entry: config_entries.ConfigEntry,
entry: FluxLedConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up the Magic Home sensors."""
coordinator: FluxLedUpdateCoordinator = hass.data[DOMAIN][entry.entry_id]
coordinator = entry.runtime_data
if coordinator.device.paired_remotes is not None:
async_add_entities(
[

View File

@ -8,7 +8,6 @@ from flux_led import DeviceType
from flux_led.aio import AIOWifiLedBulb
from flux_led.const import MODE_MUSIC
from homeassistant import config_entries
from homeassistant.components.switch import SwitchEntity
from homeassistant.const import EntityCategory
from homeassistant.core import HomeAssistant
@ -19,20 +18,19 @@ from .const import (
CONF_REMOTE_ACCESS_ENABLED,
CONF_REMOTE_ACCESS_HOST,
CONF_REMOTE_ACCESS_PORT,
DOMAIN,
)
from .coordinator import FluxLedUpdateCoordinator
from .coordinator import FluxLedConfigEntry, FluxLedUpdateCoordinator
from .discovery import async_clear_discovery_cache
from .entity import FluxBaseEntity, FluxEntity, FluxOnOffEntity
async def async_setup_entry(
hass: HomeAssistant,
entry: config_entries.ConfigEntry,
entry: FluxLedConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up the Flux lights."""
coordinator: FluxLedUpdateCoordinator = hass.data[DOMAIN][entry.entry_id]
coordinator = entry.runtime_data
entities: list[FluxSwitch | FluxRemoteAccessSwitch | FluxMusicSwitch] = []
base_unique_id = entry.unique_id or entry.entry_id
@ -70,7 +68,7 @@ class FluxRemoteAccessSwitch(FluxBaseEntity, SwitchEntity):
def __init__(
self,
device: AIOWifiLedBulb,
entry: config_entries.ConfigEntry,
entry: FluxLedConfigEntry,
) -> None:
"""Initialize the light."""
super().__init__(device, entry)