Add DataUpdateCoordinator to Elgato (#64642)

This commit is contained in:
Franck Nijhof 2022-01-21 19:38:02 +01:00 committed by GitHub
parent 63f8e437ed
commit e982df5199
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 61 additions and 61 deletions

View File

@ -1,16 +1,15 @@
"""Support for Elgato Lights.""" """Support for Elgato Lights."""
import logging
from typing import NamedTuple from typing import NamedTuple
from elgato import Elgato, ElgatoConnectionError, Info from elgato import Elgato, Info, State
from homeassistant.config_entries import ConfigEntry from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_HOST, CONF_PORT, Platform from homeassistant.const import CONF_HOST, CONF_PORT, Platform
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers.aiohttp_client import async_get_clientsession from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
from .const import DOMAIN from .const import DOMAIN, LOGGER, SCAN_INTERVAL
PLATFORMS = [Platform.BUTTON, Platform.LIGHT] PLATFORMS = [Platform.BUTTON, Platform.LIGHT]
@ -18,6 +17,7 @@ PLATFORMS = [Platform.BUTTON, Platform.LIGHT]
class HomeAssistantElgatoData(NamedTuple): class HomeAssistantElgatoData(NamedTuple):
"""Elgato data stored in the Home Assistant data object.""" """Elgato data stored in the Home Assistant data object."""
coordinator: DataUpdateCoordinator[State]
client: Elgato client: Elgato
info: Info info: Info
@ -31,15 +31,19 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
session=session, session=session,
) )
# Ensure we can connect to it coordinator: DataUpdateCoordinator[State] = DataUpdateCoordinator(
try: hass,
info = await elgato.info() LOGGER,
except ElgatoConnectionError as exception: name=f"{DOMAIN}_{entry.data[CONF_HOST]}",
logging.getLogger(__name__).debug("Unable to connect: %s", exception) update_interval=SCAN_INTERVAL,
raise ConfigEntryNotReady from exception update_method=elgato.state,
)
await coordinator.async_config_entry_first_refresh()
info = await elgato.info()
hass.data.setdefault(DOMAIN, {})[entry.entry_id] = HomeAssistantElgatoData( hass.data.setdefault(DOMAIN, {})[entry.entry_id] = HomeAssistantElgatoData(
client=elgato, client=elgato,
coordinator=coordinator,
info=info, info=info,
) )

View File

@ -1,7 +1,15 @@
"""Constants for the Elgato Light integration.""" """Constants for the Elgato Light integration."""
from __future__ import annotations
from datetime import timedelta
import logging
from typing import Final
# Integration domain # Integration domain
DOMAIN = "elgato" DOMAIN: Final = "elgato"
LOGGER = logging.getLogger(__package__)
SCAN_INTERVAL = timedelta(seconds=10)
# Attributes # Attributes
ATTR_ON = "on" ATTR_ON = "on"

View File

@ -1,8 +1,6 @@
"""Support for Elgato lights.""" """Support for Elgato lights."""
from __future__ import annotations from __future__ import annotations
from datetime import timedelta
import logging
from typing import Any from typing import Any
from elgato import Elgato, ElgatoError, Info, Settings, State from elgato import Elgato, ElgatoError, Info, Settings, State
@ -22,15 +20,16 @@ from homeassistant.helpers.entity_platform import (
AddEntitiesCallback, AddEntitiesCallback,
async_get_current_platform, async_get_current_platform,
) )
from homeassistant.helpers.update_coordinator import (
CoordinatorEntity,
DataUpdateCoordinator,
)
from . import HomeAssistantElgatoData from . import HomeAssistantElgatoData
from .const import DOMAIN, SERVICE_IDENTIFY from .const import DOMAIN, LOGGER, SERVICE_IDENTIFY
from .entity import ElgatoEntity from .entity import ElgatoEntity
_LOGGER = logging.getLogger(__name__)
PARALLEL_UPDATES = 1 PARALLEL_UPDATES = 1
SCAN_INTERVAL = timedelta(seconds=10)
async def async_setup_entry( async def async_setup_entry(
@ -42,7 +41,16 @@ async def async_setup_entry(
data: HomeAssistantElgatoData = hass.data[DOMAIN][entry.entry_id] data: HomeAssistantElgatoData = hass.data[DOMAIN][entry.entry_id]
settings = await data.client.settings() settings = await data.client.settings()
async_add_entities( async_add_entities(
[ElgatoLight(data.client, data.info, entry.data.get(CONF_MAC), settings)], True [
ElgatoLight(
data.coordinator,
data.client,
data.info,
entry.data.get(CONF_MAC),
settings,
)
],
True,
) )
platform = async_get_current_platform() platform = async_get_current_platform()
@ -53,15 +61,22 @@ async def async_setup_entry(
) )
class ElgatoLight(ElgatoEntity, LightEntity): class ElgatoLight(ElgatoEntity, CoordinatorEntity, LightEntity):
"""Defines an Elgato Light.""" """Defines an Elgato Light."""
coordinator: DataUpdateCoordinator[State]
def __init__( def __init__(
self, client: Elgato, info: Info, mac: str | None, settings: Settings self,
coordinator: DataUpdateCoordinator,
client: Elgato,
info: Info,
mac: str | None,
settings: Settings,
) -> None: ) -> None:
"""Initialize Elgato Light.""" """Initialize Elgato Light."""
super().__init__(client, info, mac) super().__init__(client, info, mac)
self._state: State | None = None CoordinatorEntity.__init__(self, coordinator)
min_mired = 143 min_mired = 143
max_mired = 344 max_mired = 344
@ -79,27 +94,20 @@ class ElgatoLight(ElgatoEntity, LightEntity):
self._attr_supported_color_modes = supported_color_modes self._attr_supported_color_modes = supported_color_modes
self._attr_unique_id = info.serial_number self._attr_unique_id = info.serial_number
@property
def available(self) -> bool:
"""Return True if entity is available."""
return self._state is not None
@property @property
def brightness(self) -> int | None: def brightness(self) -> int | None:
"""Return the brightness of this light between 1..255.""" """Return the brightness of this light between 1..255."""
assert self._state is not None return round((self.coordinator.data.brightness * 255) / 100)
return round((self._state.brightness * 255) / 100)
@property @property
def color_temp(self) -> int | None: def color_temp(self) -> int | None:
"""Return the CT color value in mireds.""" """Return the CT color value in mireds."""
assert self._state is not None return self.coordinator.data.temperature
return self._state.temperature
@property @property
def color_mode(self) -> str | None: def color_mode(self) -> str | None:
"""Return the color mode of the light.""" """Return the color mode of the light."""
if self._state and self._state.hue is not None: if self.coordinator.data.hue is not None:
return COLOR_MODE_HS return COLOR_MODE_HS
return COLOR_MODE_COLOR_TEMP return COLOR_MODE_COLOR_TEMP
@ -107,28 +115,20 @@ class ElgatoLight(ElgatoEntity, LightEntity):
@property @property
def hs_color(self) -> tuple[float, float] | None: def hs_color(self) -> tuple[float, float] | None:
"""Return the hue and saturation color value [float, float].""" """Return the hue and saturation color value [float, float]."""
if ( return (self.coordinator.data.hue or 0, self.coordinator.data.saturation or 0)
self._state is None
or self._state.hue is None
or self._state.saturation is None
):
return None
return (self._state.hue, self._state.saturation)
@property @property
def is_on(self) -> bool: def is_on(self) -> bool:
"""Return the state of the light.""" """Return the state of the light."""
assert self._state is not None return self.coordinator.data.on
return self._state.on
async def async_turn_off(self, **kwargs: Any) -> None: async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn off the light.""" """Turn off the light."""
try: try:
await self.client.light(on=False) await self.client.light(on=False)
except ElgatoError: except ElgatoError:
_LOGGER.error("An error occurred while updating the Elgato Light") LOGGER.error("An error occurred while updating the Elgato Light")
self._state = None await self.coordinator.async_refresh()
async def async_turn_on(self, **kwargs: Any) -> None: async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn on the light.""" """Turn on the light."""
@ -165,25 +165,13 @@ class ElgatoLight(ElgatoEntity, LightEntity):
temperature=temperature, temperature=temperature,
) )
except ElgatoError: except ElgatoError:
_LOGGER.error("An error occurred while updating the Elgato Light") LOGGER.error("An error occurred while updating the Elgato Light")
self._state = None await self.coordinator.async_refresh()
async def async_update(self) -> None:
"""Update Elgato entity."""
restoring = self._state is None
try:
self._state = await self.client.state()
if restoring:
_LOGGER.info("Connection restored")
except ElgatoError as err:
meth = _LOGGER.error if self._state else _LOGGER.debug
meth("An error occurred while updating the Elgato Light: %s", err)
self._state = None
async def async_identify(self) -> None: async def async_identify(self) -> None:
"""Identify the light, will make it blink.""" """Identify the light, will make it blink."""
try: try:
await self.client.identify() await self.client.identify()
except ElgatoError: except ElgatoError:
_LOGGER.exception("An error occurred while identifying the Elgato Light") LOGGER.exception("An error occurred while identifying the Elgato Light")
self._state = None await self.coordinator.async_refresh()

View File

@ -36,11 +36,11 @@ async def test_config_entry_not_ready(
mock_elgato: MagicMock, mock_elgato: MagicMock,
) -> None: ) -> None:
"""Test the Elgato configuration entry not ready.""" """Test the Elgato configuration entry not ready."""
mock_elgato.info.side_effect = ElgatoConnectionError mock_elgato.state.side_effect = ElgatoConnectionError
mock_config_entry.add_to_hass(hass) mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id) await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(mock_elgato.info.mock_calls) == 1 assert len(mock_elgato.state.mock_calls) == 1
assert mock_config_entry.state is ConfigEntryState.SETUP_RETRY assert mock_config_entry.state is ConfigEntryState.SETUP_RETRY