jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-19 11:17:21 +00:00
Code Issues Packages Projects Releases Wiki Activity

Use runtime_data in control4 (#136403)

Browse Source
This commit is contained in:
epenet 2025-01-29 17:50:36 +01:00 committed by GitHub
parent 660653e226
commit 89e6791fee
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
7 changed files with 101 additions and 111 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

102
homeassistant/components/control4/__init__.py
View File

@ -2,8 +2,10 @@
from __future__ import annotations
from dataclasses import dataclass
import json
import logging
from typing import Any
from aiohttp import client_exceptions
from pyControl4.account import C4Account
@ -25,14 +27,7 @@ from homeassistant.helpers import aiohttp_client, device_registry as dr
from .const import (
API_RETRY_TIMES,
CONF_ACCOUNT,
CONF_CONFIG_LISTENER,
CONF_CONTROLLER_UNIQUE_ID,
CONF_DIRECTOR,
CONF_DIRECTOR_ALL_ITEMS,
CONF_DIRECTOR_MODEL,
CONF_DIRECTOR_SW_VERSION,
CONF_UI_CONFIGURATION,
DEFAULT_SCAN_INTERVAL,
DOMAIN,
)
@ -42,6 +37,23 @@ _LOGGER = logging.getLogger(__name__)
PLATFORMS = [Platform.LIGHT, Platform.MEDIA_PLAYER]
@dataclass
class Control4RuntimeData:
"""Control4 runtime data."""
account: C4Account
controller_unique_id: str
director: C4Director
director_all_items: list[dict[str, Any]]
director_model: str
director_sw_version: str
scan_interval: int
ui_configuration: dict[str, Any] | None
type Control4ConfigEntry = ConfigEntry[Control4RuntimeData]
async def call_c4_api_retry(func, *func_args):
"""Call C4 API function and retry on failure."""
# Ruff doesn't understand this loop - the exception is always raised after the retries
@ -54,10 +66,8 @@ async def call_c4_api_retry(func, *func_args):
raise ConfigEntryNotReady(exception) from exception
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
async def async_setup_entry(hass: HomeAssistant, entry: Control4ConfigEntry) -> bool:
"""Set up Control4 from a config entry."""
hass.data.setdefault(DOMAIN, {})
entry_data = hass.data[DOMAIN].setdefault(entry.entry_id, {})
account_session = aiohttp_client.async_get_clientsession(hass)
config = entry.data
@ -76,10 +86,8 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
exception,
)
return False
entry_data[CONF_ACCOUNT] = account
controller_unique_id = config[CONF_CONTROLLER_UNIQUE_ID]
entry_data[CONF_CONTROLLER_UNIQUE_ID] = controller_unique_id
controller_unique_id: str = config[CONF_CONTROLLER_UNIQUE_ID]
director_token_dict = await call_c4_api_retry(
account.getDirectorBearerToken, controller_unique_id
@ -89,15 +97,14 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
director = C4Director(
config[CONF_HOST], director_token_dict[CONF_TOKEN], director_session
)
entry_data[CONF_DIRECTOR] = director
controller_href = (await call_c4_api_retry(account.getAccountControllers))["href"]
entry_data[CONF_DIRECTOR_SW_VERSION] = await call_c4_api_retry(
director_sw_version = await call_c4_api_retry(
account.getControllerOSVersion, controller_href
)
_, model, mac_address = controller_unique_id.split("_", 3)
entry_data[CONF_DIRECTOR_MODEL] = model.upper()
director_model = model.upper()
device_registry = dr.async_get(hass)
device_registry.async_get_or_create(
@ -106,57 +113,60 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
connections={(dr.CONNECTION_NETWORK_MAC, mac_address)},
manufacturer="Control4",
name=controller_unique_id,
model=entry_data[CONF_DIRECTOR_MODEL],
sw_version=entry_data[CONF_DIRECTOR_SW_VERSION],
model=director_model,
sw_version=director_sw_version,
)
# Store all items found on controller for platforms to use
director_all_items = await director.getAllItemInfo()
director_all_items = json.loads(director_all_items)
entry_data[CONF_DIRECTOR_ALL_ITEMS] = director_all_items
# Check if OS version is 3 or higher to get UI configuration
entry_data[CONF_UI_CONFIGURATION] = None
if int(entry_data[CONF_DIRECTOR_SW_VERSION].split(".")[0]) >= 3:
entry_data[CONF_UI_CONFIGURATION] = json.loads(
await director.getUiConfiguration()
)
# Load options from config entry
entry_data[CONF_SCAN_INTERVAL] = entry.options.get(
CONF_SCAN_INTERVAL, DEFAULT_SCAN_INTERVAL
director_all_items: list[dict[str, Any]] = json.loads(
await director.getAllItemInfo()
)
entry_data[CONF_CONFIG_LISTENER] = entry.add_update_listener(update_listener)
# Check if OS version is 3 or higher to get UI configuration
ui_configuration: dict[str, Any] | None = None
if int(director_sw_version.split(".")[0]) >= 3:
ui_configuration = json.loads(await director.getUiConfiguration())
# Load options from config entry
scan_interval: int = entry.options.get(CONF_SCAN_INTERVAL, DEFAULT_SCAN_INTERVAL)
entry.runtime_data = Control4RuntimeData(
account=account,
controller_unique_id=controller_unique_id,
director=director,
director_all_items=director_all_items,
director_model=director_model,
director_sw_version=director_sw_version,
scan_interval=scan_interval,
ui_configuration=ui_configuration,
)
entry.async_on_unload(entry.add_update_listener(update_listener))
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
return True
async def update_listener(hass: HomeAssistant, config_entry: ConfigEntry) -> None:
async def update_listener(
hass: HomeAssistant, config_entry: Control4ConfigEntry
) -> None:
"""Update when config_entry options update."""
_LOGGER.debug("Config entry was updated, rerunning setup")
await hass.config_entries.async_reload(config_entry.entry_id)
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
async def async_unload_entry(hass: HomeAssistant, entry: Control4ConfigEntry) -> bool:
"""Unload a config entry."""
unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
hass.data[DOMAIN][entry.entry_id][CONF_CONFIG_LISTENER]()
if unload_ok:
hass.data[DOMAIN].pop(entry.entry_id)
_LOGGER.debug("Unloaded entry for %s", entry.entry_id)
return unload_ok
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
async def get_items_of_category(hass: HomeAssistant, entry: ConfigEntry, category: str):
async def get_items_of_category(
hass: HomeAssistant, entry: Control4ConfigEntry, category: str
):
"""Return a list of all Control4 items with the specified category."""
director_all_items = hass.data[DOMAIN][entry.entry_id][CONF_DIRECTOR_ALL_ITEMS]
return [
item
for item in director_all_items
for item in entry.runtime_data.director_all_items
if "categories" in item and category in item["categories"]
]

10
homeassistant/components/control4/config_flow.py
View File

@ -11,12 +11,7 @@ from pyControl4.director import C4Director
from pyControl4.error_handling import NotFound, Unauthorized
import voluptuous as vol
from homeassistant.config_entries import (
ConfigEntry,
ConfigFlow,
ConfigFlowResult,
OptionsFlow,
)
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult, OptionsFlow
from homeassistant.const import (
CONF_HOST,
CONF_PASSWORD,
@ -28,6 +23,7 @@ from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import aiohttp_client, config_validation as cv
from homeassistant.helpers.device_registry import format_mac
from . import Control4ConfigEntry
from .const import (
CONF_CONTROLLER_UNIQUE_ID,
DEFAULT_SCAN_INTERVAL,
@ -151,7 +147,7 @@ class Control4ConfigFlow(ConfigFlow, domain=DOMAIN):
@staticmethod
@callback
def async_get_options_flow(
config_entry: ConfigEntry,
config_entry: Control4ConfigEntry,
) -> OptionsFlowHandler:
"""Get the options flow for this handler."""
return OptionsFlowHandler()

8
homeassistant/components/control4/const.py
View File

@ -7,14 +7,6 @@ MIN_SCAN_INTERVAL = 1
API_RETRY_TIMES = 5
CONF_ACCOUNT = "account"
CONF_DIRECTOR = "director"
CONF_DIRECTOR_SW_VERSION = "director_sw_version"
CONF_DIRECTOR_MODEL = "director_model"
CONF_DIRECTOR_ALL_ITEMS = "director_all_items"
CONF_UI_CONFIGURATION = "ui_configuration"
CONF_CONTROLLER_UNIQUE_ID = "controller_unique_id"
CONF_CONFIG_LISTENER = "config_listener"
CONTROL4_ENTITY_TYPE = 7

19
homeassistant/components/control4/director_utils.py
View File

@ -8,21 +8,21 @@ from pyControl4.account import C4Account
from pyControl4.director import C4Director
from pyControl4.error_handling import BadToken
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_TOKEN, CONF_USERNAME
from homeassistant.core import HomeAssistant
from homeassistant.helpers import aiohttp_client
from .const import CONF_ACCOUNT, CONF_CONTROLLER_UNIQUE_ID, CONF_DIRECTOR, DOMAIN
from . import Control4ConfigEntry
from .const import CONF_CONTROLLER_UNIQUE_ID
_LOGGER = logging.getLogger(__name__)
async def _update_variables_for_config_entry(
hass: HomeAssistant, entry: ConfigEntry, variable_names: set[str]
hass: HomeAssistant, entry: Control4ConfigEntry, variable_names: set[str]
) -> dict[int, dict[str, Any]]:
"""Retrieve data from the Control4 director."""
director: C4Director = hass.data[DOMAIN][entry.entry_id][CONF_DIRECTOR]
director = entry.runtime_data.director
data = await director.getAllItemVariableValue(variable_names)
result_dict: defaultdict[int, dict[str, Any]] = defaultdict(dict)
for item in data:
@ -31,7 +31,7 @@ async def _update_variables_for_config_entry(
async def update_variables_for_config_entry(
hass: HomeAssistant, entry: ConfigEntry, variable_names: set[str]
hass: HomeAssistant, entry: Control4ConfigEntry, variable_names: set[str]
) -> dict[int, dict[str, Any]]:
"""Try to Retrieve data from the Control4 director for update_coordinator."""
try:
@ -42,8 +42,8 @@ async def update_variables_for_config_entry(
return await _update_variables_for_config_entry(hass, entry, variable_names)
async def refresh_tokens(hass: HomeAssistant, entry: ConfigEntry):
"""Store updated authentication and director tokens in hass.data."""
async def refresh_tokens(hass: HomeAssistant, entry: Control4ConfigEntry):
"""Store updated authentication and director tokens in runtime_data."""
config = entry.data
account_session = aiohttp_client.async_get_clientsession(hass)
@ -59,6 +59,5 @@ async def refresh_tokens(hass: HomeAssistant, entry: ConfigEntry):
)
_LOGGER.debug("Saving new tokens in hass data")
entry_data = hass.data[DOMAIN][entry.entry_id]
entry_data[CONF_ACCOUNT] = account
entry_data[CONF_DIRECTOR] = director
entry.runtime_data.account = account
entry.runtime_data.director = director

9
homeassistant/components/control4/entity.py
View File

@ -10,7 +10,8 @@ from homeassistant.helpers.update_coordinator import (
DataUpdateCoordinator,
)
from .const import CONF_CONTROLLER_UNIQUE_ID, DOMAIN
from . import Control4RuntimeData
from .const import DOMAIN
class Control4Entity(CoordinatorEntity[Any]):
@ -18,7 +19,7 @@ class Control4Entity(CoordinatorEntity[Any]):
def __init__(
self,
entry_data: dict,
runtime_data: Control4RuntimeData,
coordinator: DataUpdateCoordinator[Any],
name: str | None,
idx: int,
@ -29,11 +30,11 @@ class Control4Entity(CoordinatorEntity[Any]):
) -> None:
"""Initialize a Control4 entity."""
super().__init__(coordinator)
self.entry_data = entry_data
self.runtime_data = runtime_data
self._attr_name = name
self._attr_unique_id = str(idx)
self._idx = idx
self._controller_unique_id = entry_data[CONF_CONTROLLER_UNIQUE_ID]
self._controller_unique_id = runtime_data.controller_unique_id
self._device_name = device_name
self._device_manufacturer = device_manufacturer
self._device_model = device_model

32
homeassistant/components/control4/light.py
View File

@ -17,14 +17,12 @@ from homeassistant.components.light import (
LightEntity,
LightEntityFeature,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_SCAN_INTERVAL
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from . import get_items_of_category
from .const import CONF_DIRECTOR, CONTROL4_ENTITY_TYPE, DOMAIN
from . import Control4ConfigEntry, Control4RuntimeData, get_items_of_category
from .const import CONTROL4_ENTITY_TYPE
from .director_utils import update_variables_for_config_entry
from .entity import Control4Entity
@ -36,15 +34,13 @@ CONTROL4_DIMMER_VARS = ["LIGHT_LEVEL", "Brightness Percent"]
async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
hass: HomeAssistant,
entry: Control4ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up Control4 lights from a config entry."""
entry_data = hass.data[DOMAIN][entry.entry_id]
scan_interval = entry_data[CONF_SCAN_INTERVAL]
_LOGGER.debug(
"Scan interval = %s",
scan_interval,
)
runtime_data = entry.runtime_data
_LOGGER.debug("Scan interval = %s", runtime_data.scan_interval)
async def async_update_data_non_dimmer() -> dict[int, dict[str, Any]]:
"""Fetch data from Control4 director for non-dimmer lights."""
@ -69,14 +65,14 @@ async def async_setup_entry(
_LOGGER,
name="light",
update_method=async_update_data_non_dimmer,
update_interval=timedelta(seconds=scan_interval),
update_interval=timedelta(seconds=runtime_data.scan_interval),
)
dimmer_coordinator = DataUpdateCoordinator[dict[int, dict[str, Any]]](
hass,
_LOGGER,
name="light",
update_method=async_update_data_dimmer,
update_interval=timedelta(seconds=scan_interval),
update_interval=timedelta(seconds=runtime_data.scan_interval),
)
# Fetch initial data so we have data when entities subscribe
@ -118,7 +114,7 @@ async def async_setup_entry(
item_is_dimmer = False
item_coordinator = non_dimmer_coordinator
else:
director = entry_data[CONF_DIRECTOR]
director = runtime_data.director
item_variables = await director.getItemVariables(item_id)
_LOGGER.warning(
(
@ -132,7 +128,7 @@ async def async_setup_entry(
entity_list.append(
Control4Light(
entry_data,
runtime_data,
item_coordinator,
item_name,
item_id,
@ -154,7 +150,7 @@ class Control4Light(Control4Entity, LightEntity):
def __init__(
self,
entry_data: dict,
runtime_data: Control4RuntimeData,
coordinator: DataUpdateCoordinator[dict[int, dict[str, Any]]],
name: str,
idx: int,
@ -166,7 +162,7 @@ class Control4Light(Control4Entity, LightEntity):
) -> None:
"""Initialize Control4 light entity."""
super().__init__(
entry_data,
runtime_data,
coordinator,
name,
idx,
@ -188,7 +184,7 @@ class Control4Light(Control4Entity, LightEntity):
This exists so the director token used is always the latest one, without needing to re-init the entire entity.
"""
return C4Light(self.entry_data[CONF_DIRECTOR], self._idx)
return C4Light(self.runtime_data.director, self._idx)
@property
def is_on(self):

32
homeassistant/components/control4/media_player.py
View File

@ -18,13 +18,11 @@ from homeassistant.components.media_player import (
MediaPlayerState,
MediaType,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_SCAN_INTERVAL
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import CONF_DIRECTOR, CONF_DIRECTOR_ALL_ITEMS, CONF_UI_CONFIGURATION, DOMAIN
from . import Control4ConfigEntry, Control4RuntimeData
from .director_utils import update_variables_for_config_entry
from .entity import Control4Entity
@ -67,22 +65,23 @@ class _RoomSource:
name: str
async def get_rooms(hass: HomeAssistant, entry: ConfigEntry):
async def get_rooms(hass: HomeAssistant, entry: Control4ConfigEntry):
"""Return a list of all Control4 rooms."""
director_all_items = hass.data[DOMAIN][entry.entry_id][CONF_DIRECTOR_ALL_ITEMS]
return [
item
for item in director_all_items
for item in entry.runtime_data.director_all_items
if "typeName" in item and item["typeName"] == "room"
]
async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
hass: HomeAssistant,
entry: Control4ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up Control4 rooms from a config entry."""
entry_data = hass.data[DOMAIN][entry.entry_id]
ui_config = entry_data[CONF_UI_CONFIGURATION]
runtime_data = entry.runtime_data
ui_config = runtime_data.ui_configuration
# OS 2 will not have a ui_configuration
if not ui_config:
@ -93,7 +92,7 @@ async def async_setup_entry(
if not all_rooms:
return
scan_interval = entry_data[CONF_SCAN_INTERVAL]
scan_interval = runtime_data.scan_interval
_LOGGER.debug("Scan interval = %s", scan_interval)
async def async_update_data() -> dict[int, dict[str, Any]]:
@ -116,10 +115,7 @@ async def async_setup_entry(
# Fetch initial data so we have data when entities subscribe
await coordinator.async_refresh()
items_by_id = {
item["id"]: item
for item in hass.data[DOMAIN][entry.entry_id][CONF_DIRECTOR_ALL_ITEMS]
}
items_by_id = {item["id"]: item for item in runtime_data.director_all_items}
item_to_parent_map = {
k: item["parentId"]
for k, item in items_by_id.items()
@ -156,7 +152,7 @@ async def async_setup_entry(
hidden = room["roomHidden"]
entity_list.append(
Control4Room(
entry_data,
runtime_data,
coordinator,
room["name"],
room_id,
@ -182,7 +178,7 @@ class Control4Room(Control4Entity, MediaPlayerEntity):
def __init__(
self,
entry_data: dict,
runtime_data: Control4RuntimeData,
coordinator: DataUpdateCoordinator[dict[int, dict[str, Any]]],
name: str,
room_id: int,
@ -192,7 +188,7 @@ class Control4Room(Control4Entity, MediaPlayerEntity):
) -> None:
"""Initialize Control4 room entity."""
super().__init__(
entry_data,
runtime_data,
coordinator,
None,
room_id,
@ -220,7 +216,7 @@ class Control4Room(Control4Entity, MediaPlayerEntity):
This exists so the director token used is always the latest one, without needing to re-init the entire entity.
"""
return C4Room(self.entry_data[CONF_DIRECTOR], self._idx)
return C4Room(self.runtime_data.director, self._idx)
def _get_device_from_variable(self, var: str) -> int | None:
current_device = self.coordinator.data[self._idx][var]