mirror of
https://github.com/Electric-Special/ha-core.git
synced 2026-03-21 06:05:26 +01:00
110 lines
4.1 KiB
Python
110 lines
4.1 KiB
Python
"""Select platform for OpenRGB integration."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from homeassistant.components.select import SelectEntity
|
|
from homeassistant.core import HomeAssistant, callback
|
|
from homeassistant.exceptions import HomeAssistantError
|
|
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
|
|
from homeassistant.helpers.update_coordinator import CoordinatorEntity
|
|
|
|
from .const import CONNECTION_ERRORS, DOMAIN, UID_SEPARATOR
|
|
from .coordinator import OpenRGBConfigEntry, OpenRGBCoordinator
|
|
|
|
PARALLEL_UPDATES = 0
|
|
|
|
|
|
async def async_setup_entry(
|
|
hass: HomeAssistant,
|
|
config_entry: OpenRGBConfigEntry,
|
|
async_add_entities: AddConfigEntryEntitiesCallback,
|
|
) -> None:
|
|
"""Set up the OpenRGB select platform."""
|
|
coordinator = config_entry.runtime_data
|
|
async_add_entities([OpenRGBProfileSelect(coordinator, config_entry)])
|
|
|
|
|
|
class OpenRGBProfileSelect(CoordinatorEntity[OpenRGBCoordinator], SelectEntity):
|
|
"""Representation of an OpenRGB profile select entity."""
|
|
|
|
_attr_translation_key = "profile"
|
|
_attr_has_entity_name = True
|
|
|
|
_state_hash: int | None = None
|
|
_pending_profile: str | None = None
|
|
|
|
def __init__(
|
|
self, coordinator: OpenRGBCoordinator, entry: OpenRGBConfigEntry
|
|
) -> None:
|
|
"""Initialize the select entity."""
|
|
super().__init__(coordinator)
|
|
self._attr_unique_id = UID_SEPARATOR.join([entry.entry_id, "profile"])
|
|
self._attr_device_info = {
|
|
"identifiers": {(DOMAIN, entry.entry_id)},
|
|
}
|
|
self._update_attrs()
|
|
|
|
def _compute_state_hash(self) -> int:
|
|
"""Compute a hash of device states."""
|
|
return hash(
|
|
"\n".join(str(device.data) for device in self.coordinator.client.devices)
|
|
)
|
|
|
|
@callback
|
|
def _update_attrs(self) -> None:
|
|
"""Update the attributes based on the current profile list."""
|
|
profiles = self.coordinator.client.profiles
|
|
self._attr_options = [profile.name for profile in profiles]
|
|
|
|
# If a profile was just applied, set it as current
|
|
if self._pending_profile is not None:
|
|
self._attr_current_option = self._pending_profile
|
|
self._pending_profile = None
|
|
self._state_hash = self._compute_state_hash()
|
|
# Only check for state changes if we have a current option to potentially clear
|
|
elif self._attr_current_option is not None:
|
|
current_hash = self._compute_state_hash()
|
|
# If state changed, we can no longer assume current profile
|
|
if current_hash != self._state_hash:
|
|
self._attr_current_option = None
|
|
self._state_hash = None
|
|
|
|
@callback
|
|
def _handle_coordinator_update(self) -> None:
|
|
"""Handle updated data from the coordinator."""
|
|
self._update_attrs()
|
|
super()._handle_coordinator_update()
|
|
|
|
@property
|
|
def available(self) -> bool:
|
|
"""Return if the select is available."""
|
|
return super().available and bool(self._attr_options)
|
|
|
|
async def async_select_option(self, option: str) -> None:
|
|
"""Load the selected profile."""
|
|
async with self.coordinator.client_lock:
|
|
try:
|
|
await self.hass.async_add_executor_job(
|
|
self.coordinator.client.load_profile, option
|
|
)
|
|
except CONNECTION_ERRORS as err:
|
|
raise HomeAssistantError(
|
|
translation_domain=DOMAIN,
|
|
translation_key="communication_error",
|
|
translation_placeholders={
|
|
"server_address": self.coordinator.server_address,
|
|
"error": str(err),
|
|
},
|
|
) from err
|
|
except ValueError as err:
|
|
raise HomeAssistantError(
|
|
translation_domain=DOMAIN,
|
|
translation_key="openrgb_error",
|
|
translation_placeholders={
|
|
"error": str(err),
|
|
},
|
|
) from err
|
|
|
|
self._pending_profile = option
|
|
await self.coordinator.async_refresh()
|