Add new control4 helper function (#90234)
* Add new helper function to retrieve device variables and update light platform * seperate try catch from helper function and fix typing * Change helper function name * Remove unnecessary forced type changes * More type changes
This commit is contained in:
parent
3599515325
commit
0e7d7f32c1
2 changed files with 32 additions and 16 deletions
|
@ -1,5 +1,8 @@
|
|||
"""Provides data updates from the Control4 controller for platforms."""
|
||||
from collections import defaultdict
|
||||
from collections.abc import Set
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
from pyControl4.account import C4Account
|
||||
from pyControl4.director import C4Director
|
||||
|
@ -15,21 +18,28 @@ from .const import CONF_ACCOUNT, CONF_CONTROLLER_UNIQUE_ID, CONF_DIRECTOR, DOMAI
|
|||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
async def director_update_data(
|
||||
hass: HomeAssistant, entry: ConfigEntry, var: str
|
||||
) -> dict:
|
||||
"""Retrieve data from the Control4 director for update_coordinator."""
|
||||
# possibly implement usage of director_token_expiration to start
|
||||
# token refresh without waiting for error to occur
|
||||
async def _update_variables_for_config_entry(
|
||||
hass: HomeAssistant, entry: ConfigEntry, 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]
|
||||
data = await director.getAllItemVariableValue(variable_names)
|
||||
result_dict: defaultdict[int, dict[str, Any]] = defaultdict(dict)
|
||||
for item in data:
|
||||
result_dict[item["id"]][item["varName"]] = item["value"]
|
||||
return dict(result_dict)
|
||||
|
||||
|
||||
async def update_variables_for_config_entry(
|
||||
hass: HomeAssistant, entry: ConfigEntry, variable_names: Set[str]
|
||||
) -> dict[int, dict[str, Any]]:
|
||||
"""Try to Retrieve data from the Control4 director for update_coordinator."""
|
||||
try:
|
||||
director = hass.data[DOMAIN][entry.entry_id][CONF_DIRECTOR]
|
||||
data = await director.getAllItemVariableValue(var)
|
||||
return await _update_variables_for_config_entry(hass, entry, variable_names)
|
||||
except BadToken:
|
||||
_LOGGER.info("Updating Control4 director token")
|
||||
await refresh_tokens(hass, entry)
|
||||
director = hass.data[DOMAIN][entry.entry_id][CONF_DIRECTOR]
|
||||
data = await director.getAllItemVariableValue(var)
|
||||
return {key["id"]: key for key in data}
|
||||
return await _update_variables_for_config_entry(hass, entry, variable_names)
|
||||
|
||||
|
||||
async def refresh_tokens(hass: HomeAssistant, entry: ConfigEntry):
|
||||
|
|
|
@ -24,7 +24,7 @@ from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, Upda
|
|||
|
||||
from . import Control4Entity, get_items_of_category
|
||||
from .const import CONF_DIRECTOR, CONTROL4_ENTITY_TYPE, DOMAIN
|
||||
from .director_utils import director_update_data
|
||||
from .director_utils import update_variables_for_config_entry
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
@ -47,14 +47,18 @@ async def async_setup_entry(
|
|||
async def async_update_data_non_dimmer():
|
||||
"""Fetch data from Control4 director for non-dimmer lights."""
|
||||
try:
|
||||
return await director_update_data(hass, entry, CONTROL4_NON_DIMMER_VAR)
|
||||
return await update_variables_for_config_entry(
|
||||
hass, entry, {CONTROL4_NON_DIMMER_VAR}
|
||||
)
|
||||
except C4Exception as err:
|
||||
raise UpdateFailed(f"Error communicating with API: {err}") from err
|
||||
|
||||
async def async_update_data_dimmer():
|
||||
"""Fetch data from Control4 director for dimmer lights."""
|
||||
try:
|
||||
return await director_update_data(hass, entry, CONTROL4_DIMMER_VAR)
|
||||
return await update_variables_for_config_entry(
|
||||
hass, entry, {CONTROL4_DIMMER_VAR}
|
||||
)
|
||||
except C4Exception as err:
|
||||
raise UpdateFailed(f"Error communicating with API: {err}") from err
|
||||
|
||||
|
@ -185,13 +189,15 @@ class Control4Light(Control4Entity, LightEntity):
|
|||
@property
|
||||
def is_on(self):
|
||||
"""Return whether this light is on or off."""
|
||||
return self.coordinator.data[self._idx]["value"] > 0
|
||||
if self._is_dimmer:
|
||||
return self.coordinator.data[self._idx][CONTROL4_DIMMER_VAR] > 0
|
||||
return self.coordinator.data[self._idx][CONTROL4_NON_DIMMER_VAR] > 0
|
||||
|
||||
@property
|
||||
def brightness(self):
|
||||
"""Return the brightness of this light between 0..255."""
|
||||
if self._is_dimmer:
|
||||
return round(self.coordinator.data[self._idx]["value"] * 2.55)
|
||||
return round(self.coordinator.data[self._idx][CONTROL4_DIMMER_VAR] * 2.55)
|
||||
return None
|
||||
|
||||
@property
|
||||
|
|
Loading…
Add table
Reference in a new issue