Add twinkly integration (#42103)

* Add twinkly integration

* Add tests for the Twinkly integration

* Update Twinkly client package to fix typo

* Remove support of configuration.yaml from Twinkly integration

* Add ability to unload Twinkly component from the UI

* Remove dead code from Twinkly

* Fix invalid error namespace in Twinkly for python 3.7

* Fix tests failing on CI

* Workaround code analysis issue

* Move twinkly client init out of entry setup so it can be re-used between entries

* Test the twinkly component initialization

* React to PR review and add few more tests
pull/43403/head
David 2020-11-19 12:22:12 -05:00 committed by GitHub
parent 5dcbb634f6
commit f693c8a9fd
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
15 changed files with 821 additions and 0 deletions

View File

@ -471,6 +471,7 @@ homeassistant/components/transmission/* @engrbm87 @JPHutchins
homeassistant/components/tts/* @pvizeli
homeassistant/components/tuya/* @ollo69
homeassistant/components/twentemilieu/* @frenck
homeassistant/components/twinkly/* @dr1rrb
homeassistant/components/ubee/* @mzdrale
homeassistant/components/unifi/* @Kane610
homeassistant/components/unifiled/* @florisvdk

View File

@ -0,0 +1,44 @@
"""The twinkly component."""
import twinkly_client
from homeassistant.config_entries import ConfigEntry
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.typing import HomeAssistantType
from .const import CONF_ENTRY_HOST, CONF_ENTRY_ID, DOMAIN
async def async_setup(hass: HomeAssistantType, config: dict):
"""Set up the twinkly integration."""
return True
async def async_setup_entry(hass: HomeAssistantType, config_entry: ConfigEntry):
"""Set up entries from config flow."""
# We setup the client here so if at some point we add any other entity for this device,
# we will be able to properly share the connection.
uuid = config_entry.data[CONF_ENTRY_ID]
host = config_entry.data[CONF_ENTRY_HOST]
hass.data.setdefault(DOMAIN, {})[uuid] = twinkly_client.TwinklyClient(
host, async_get_clientsession(hass)
)
hass.async_create_task(
hass.config_entries.async_forward_entry_setup(config_entry, "light")
)
return True
async def async_unload_entry(hass: HomeAssistantType, config_entry: ConfigEntry):
"""Remove a twinkly entry."""
# For now light entries don't have unload method, so we don't have to async_forward_entry_unload
# However we still have to cleanup the shared client!
uuid = config_entry.data[CONF_ENTRY_ID]
hass.data[DOMAIN].pop(uuid)
return True

View File

@ -0,0 +1,63 @@
"""Config flow to configure the Twinkly integration."""
import asyncio
import logging
from aiohttp import ClientError
import twinkly_client
from voluptuous import Required, Schema
from homeassistant import config_entries
from homeassistant.const import CONF_HOST
from .const import (
CONF_ENTRY_HOST,
CONF_ENTRY_ID,
CONF_ENTRY_MODEL,
CONF_ENTRY_NAME,
DEV_ID,
DEV_MODEL,
DEV_NAME,
)
# https://github.com/PyCQA/pylint/issues/3202
from .const import DOMAIN # pylint: disable=unused-import
_LOGGER = logging.getLogger(__name__)
class TwinklyConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle twinkly config flow."""
VERSION = 1
async def async_step_user(self, user_input=None):
"""Handle config steps."""
host = user_input[CONF_HOST] if user_input else None
schema = {Required(CONF_HOST, default=host): str}
errors = {}
if host is not None:
try:
device_info = await twinkly_client.TwinklyClient(host).get_device_info()
await self.async_set_unique_id(device_info[DEV_ID])
self._abort_if_unique_id_configured()
return self.async_create_entry(
title=device_info[DEV_NAME],
data={
CONF_ENTRY_HOST: host,
CONF_ENTRY_ID: device_info[DEV_ID],
CONF_ENTRY_NAME: device_info[DEV_NAME],
CONF_ENTRY_MODEL: device_info[DEV_MODEL],
},
)
except (asyncio.TimeoutError, ClientError) as err:
_LOGGER.info("Cannot reach Twinkly '%s' (client)", host, exc_info=err)
errors[CONF_HOST] = "cannot_connect"
return self.async_show_form(
step_id="user", data_schema=Schema(schema), errors=errors
)

View File

@ -0,0 +1,23 @@
"""Const for Twinkly."""
DOMAIN = "twinkly"
# Keys of the config entry
CONF_ENTRY_ID = "id"
CONF_ENTRY_HOST = "host"
CONF_ENTRY_NAME = "name"
CONF_ENTRY_MODEL = "model"
# Strongly named HA attributes keys
ATTR_HOST = "host"
# Keys of attributes read from the get_device_info
DEV_ID = "uuid"
DEV_NAME = "device_name"
DEV_MODEL = "product_code"
HIDDEN_DEV_VALUES = (
"code", # This is the internal status code of the API response
"copyright", # We should not display a copyright "LEDWORKS 2018" in the Home-Assistant UI
"mac", # Does not report the actual device mac address
)

View File

@ -0,0 +1,216 @@
"""The Twinkly light component."""
import asyncio
import logging
from typing import Any, Dict, Optional
from aiohttp import ClientError
from homeassistant.components.light import (
ATTR_BRIGHTNESS,
SUPPORT_BRIGHTNESS,
LightEntity,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.helpers.typing import HomeAssistantType
from .const import (
ATTR_HOST,
CONF_ENTRY_HOST,
CONF_ENTRY_ID,
CONF_ENTRY_MODEL,
CONF_ENTRY_NAME,
DEV_MODEL,
DEV_NAME,
DOMAIN,
HIDDEN_DEV_VALUES,
)
_LOGGER = logging.getLogger(__name__)
async def async_setup_entry(
hass: HomeAssistantType, config_entry: ConfigEntry, async_add_entities
) -> None:
"""Setups an entity from a config entry (UI config flow)."""
entity = TwinklyLight(config_entry, hass)
async_add_entities([entity], update_before_add=True)
class TwinklyLight(LightEntity):
"""Implementation of the light for the Twinkly service."""
def __init__(
self,
conf: ConfigEntry,
hass: HomeAssistantType,
):
"""Initialize a TwinklyLight entity."""
self._id = conf.data[CONF_ENTRY_ID]
self._hass = hass
self._conf = conf
# Those are saved in the config entry in order to have meaningful values even
# if the device is currently offline.
# They are expected to be updated using the device_info.
self.__name = conf.data[CONF_ENTRY_NAME]
self.__model = conf.data[CONF_ENTRY_MODEL]
self._client = hass.data.get(DOMAIN, {}).get(self._id)
if self._client is None:
raise ValueError(f"Client for {self._id} has not been configured.")
# Set default state before any update
self._is_on = False
self._brightness = 0
self._is_available = False
self._attributes = {ATTR_HOST: self._client.host}
@property
def supported_features(self):
"""Get the features supported by this entity."""
return SUPPORT_BRIGHTNESS
@property
def should_poll(self) -> bool:
"""Get a boolean which indicates if this entity should be polled."""
return True
@property
def available(self) -> bool:
"""Get a boolean which indicates if this entity is currently available."""
return self._is_available
@property
def unique_id(self) -> Optional[str]:
"""Id of the device."""
return self._id
@property
def name(self) -> str:
"""Name of the device."""
return self.__name if self.__name else "Twinkly light"
@property
def model(self) -> str:
"""Name of the device."""
return self.__model
@property
def icon(self) -> str:
"""Icon of the device."""
return "mdi:string-lights"
@property
def device_info(self) -> Optional[Dict[str, Any]]:
"""Get device specific attributes."""
return (
{
"identifiers": {(DOMAIN, self._id)},
"name": self.name,
"manufacturer": "LEDWORKS",
"model": self.model,
}
if self._id
else None # device_info is available only for entities configured from the UI
)
@property
def is_on(self) -> bool:
"""Return true if light is on."""
return self._is_on
@property
def brightness(self) -> Optional[int]:
"""Return the brightness of the light."""
return self._brightness
@property
def state_attributes(self) -> dict:
"""Return device specific state attributes."""
attributes = self._attributes
# Make sure to update any normalized property
attributes[ATTR_HOST] = self._client.host
attributes[ATTR_BRIGHTNESS] = self._brightness
return attributes
async def async_turn_on(self, **kwargs) -> None:
"""Turn device on."""
if ATTR_BRIGHTNESS in kwargs:
brightness = int(int(kwargs[ATTR_BRIGHTNESS]) / 2.55)
# If brightness is 0, the twinkly will only "disable" the brightness,
# which means that it will be 100%.
if brightness == 0:
await self._client.set_is_on(False)
return
await self._client.set_brightness(brightness)
await self._client.set_is_on(True)
async def async_turn_off(self, **kwargs) -> None:
"""Turn device off."""
await self._client.set_is_on(False)
async def async_update(self) -> None:
"""Asynchronously updates the device properties."""
_LOGGER.info("Updating '%s'", self._client.host)
try:
self._is_on = await self._client.get_is_on()
self._brightness = (
int(round((await self._client.get_brightness()) * 2.55))
if self._is_on
else 0
)
device_info = await self._client.get_device_info()
if (
DEV_NAME in device_info
and DEV_MODEL in device_info
and (
device_info[DEV_NAME] != self.__name
or device_info[DEV_MODEL] != self.__model
)
):
self.__name = device_info[DEV_NAME]
self.__model = device_info[DEV_MODEL]
if self._conf is not None:
# If the name has changed, persist it in conf entry,
# so we will be able to restore this new name if hass is started while the LED string is offline.
self._hass.config_entries.async_update_entry(
self._conf,
data={
CONF_ENTRY_HOST: self._client.host, # this cannot change
CONF_ENTRY_ID: self._id, # this cannot change
CONF_ENTRY_NAME: self.__name,
CONF_ENTRY_MODEL: self.__model,
},
)
for key, value in device_info.items():
if key not in HIDDEN_DEV_VALUES:
self._attributes[key] = value
if not self._is_available:
_LOGGER.info("Twinkly '%s' is now available", self._client.host)
# We don't use the echo API to track the availability since we already have to pull
# the device to get its state.
self._is_available = True
except (asyncio.TimeoutError, ClientError):
# We log this as "info" as it's pretty common that the christmas light are not reachable in july
if self._is_available:
_LOGGER.info(
"Twinkly '%s' is not reachable (client error)", self._client.host
)
self._is_available = False

View File

@ -0,0 +1,9 @@
{
"domain": "twinkly",
"name": "Twinkly",
"documentation": "https://www.home-assistant.io/integrations/twinkly",
"requirements": ["twinkly-client==0.0.2"],
"dependencies": [],
"codeowners": ["@dr1rrb"],
"config_flow": true
}

View File

@ -0,0 +1,19 @@
{
"config": {
"step": {
"user": {
"title": "Twinkly",
"description": "Set up your Twinkly led string",
"data": {
"host": "Host (or IP address) of your twinkly device"
}
}
},
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]"
},
"abort": {
"device_exists": "[%key:common::config_flow::abort::already_configured_device%]"
}
}
}

View File

@ -0,0 +1,19 @@
{
"config": {
"abort": {
"device_exists": "Device is already configured"
},
"error": {
"cannot_connect": "Failed to connect"
},
"step": {
"user": {
"data": {
"host": "Host (or IP address) of your twinkly device"
},
"description": "Set up your Twinkly led string",
"title": "Twinkly"
}
}
}
}

View File

@ -208,6 +208,7 @@ FLOWS = [
"tuya",
"twentemilieu",
"twilio",
"twinkly",
"unifi",
"upb",
"upcloud",

View File

@ -2209,6 +2209,9 @@ twentemilieu==0.3.0
# homeassistant.components.twilio
twilio==6.32.0
# homeassistant.components.twinkly
twinkly-client==0.0.2
# homeassistant.components.rainforest_eagle
uEagle==0.0.2

View File

@ -1062,6 +1062,9 @@ twentemilieu==0.3.0
# homeassistant.components.twilio
twilio==6.32.0
# homeassistant.components.twinkly
twinkly-client==0.0.2
# homeassistant.components.upb
upb_lib==0.4.11

View File

@ -0,0 +1,69 @@
"""Constants and mock for the twkinly component tests."""
from uuid import uuid4
from aiohttp.client_exceptions import ClientConnectionError
from homeassistant.components.twinkly.const import DEV_NAME
TEST_HOST = "test.twinkly.com"
TEST_ID = "twinkly_test_device_id"
TEST_NAME = "twinkly_test_device_name"
TEST_NAME_ORIGINAL = "twinkly_test_original_device_name" # the original (deprecated) name stored in the conf
TEST_MODEL = "twinkly_test_device_model"
class ClientMock:
"""A mock of the twinkly_client.TwinklyClient."""
def __init__(self) -> None:
"""Create a mocked client."""
self.is_offline = False
self.is_on = True
self.brightness = 10
self.id = str(uuid4())
self.device_info = {
"uuid": self.id,
"device_name": self.id, # we make sure that entity id is different for each test
"product_code": TEST_MODEL,
}
@property
def host(self) -> str:
"""Get the mocked host."""
return TEST_HOST
async def get_device_info(self):
"""Get the mocked device info."""
if self.is_offline:
raise ClientConnectionError()
return self.device_info
async def get_is_on(self) -> bool:
"""Get the mocked on/off state."""
if self.is_offline:
raise ClientConnectionError()
return self.is_on
async def set_is_on(self, is_on: bool) -> None:
"""Set the mocked on/off state."""
if self.is_offline:
raise ClientConnectionError()
self.is_on = is_on
async def get_brightness(self) -> int:
"""Get the mocked brightness."""
if self.is_offline:
raise ClientConnectionError()
return self.brightness
async def set_brightness(self, brightness: int) -> None:
"""Set the mocked brightness."""
if self.is_offline:
raise ClientConnectionError()
self.brightness = brightness
def change_name(self, new_name: str) -> None:
"""Change the name of this virtual device."""
self.device_info[DEV_NAME] = new_name

View File

@ -0,0 +1,60 @@
"""Tests for the config_flow of the twinly component."""
from homeassistant import config_entries
from homeassistant.components.twinkly.const import (
CONF_ENTRY_HOST,
CONF_ENTRY_ID,
CONF_ENTRY_MODEL,
CONF_ENTRY_NAME,
DOMAIN as TWINKLY_DOMAIN,
)
from tests.async_mock import patch
from tests.components.twinkly import TEST_MODEL, ClientMock
async def test_invalid_host(hass):
"""Test the failure when invalid host provided."""
result = await hass.config_entries.flow.async_init(
TWINKLY_DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == "form"
assert result["step_id"] == "user"
assert result["errors"] == {}
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_ENTRY_HOST: "dummy"},
)
assert result["type"] == "form"
assert result["step_id"] == "user"
assert result["errors"] == {CONF_ENTRY_HOST: "cannot_connect"}
async def test_success_flow(hass):
"""Test that an entity is created when the flow completes."""
client = ClientMock()
with patch("twinkly_client.TwinklyClient", return_value=client):
result = await hass.config_entries.flow.async_init(
TWINKLY_DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == "form"
assert result["step_id"] == "user"
assert result["errors"] == {}
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_ENTRY_HOST: "dummy"},
)
assert result["type"] == "create_entry"
assert result["title"] == client.id
assert result["data"] == {
CONF_ENTRY_HOST: "dummy",
CONF_ENTRY_ID: client.id,
CONF_ENTRY_NAME: client.id,
CONF_ENTRY_MODEL: TEST_MODEL,
}

View File

@ -0,0 +1,67 @@
"""Tests of the initialization of the twinly integration."""
from uuid import uuid4
from homeassistant.components.twinkly import async_setup_entry, async_unload_entry
from homeassistant.components.twinkly.const import (
CONF_ENTRY_HOST,
CONF_ENTRY_ID,
CONF_ENTRY_MODEL,
CONF_ENTRY_NAME,
DOMAIN as TWINKLY_DOMAIN,
)
from homeassistant.core import HomeAssistant
from tests.async_mock import patch
from tests.common import MockConfigEntry
from tests.components.twinkly import TEST_HOST, TEST_MODEL, TEST_NAME_ORIGINAL
async def test_setup_entry(hass: HomeAssistant):
"""Validate that setup entry also configure the client."""
id = str(uuid4())
config_entry = MockConfigEntry(
domain=TWINKLY_DOMAIN,
data={
CONF_ENTRY_HOST: TEST_HOST,
CONF_ENTRY_ID: id,
CONF_ENTRY_NAME: TEST_NAME_ORIGINAL,
CONF_ENTRY_MODEL: TEST_MODEL,
},
entry_id=id,
)
def setup_mock(_, __):
return True
with patch(
"homeassistant.config_entries.ConfigEntries.async_forward_entry_setup",
side_effect=setup_mock,
):
await async_setup_entry(hass, config_entry)
assert hass.data[TWINKLY_DOMAIN][id] is not None
async def test_unload_entry(hass: HomeAssistant):
"""Validate that unload entry also clear the client."""
id = str(uuid4())
config_entry = MockConfigEntry(
domain=TWINKLY_DOMAIN,
data={
CONF_ENTRY_HOST: TEST_HOST,
CONF_ENTRY_ID: id,
CONF_ENTRY_NAME: TEST_NAME_ORIGINAL,
CONF_ENTRY_MODEL: TEST_MODEL,
},
entry_id=id,
)
# Put random content at the location where the client should have been placed by setup
hass.data.setdefault(TWINKLY_DOMAIN, {})[id] = config_entry
await async_unload_entry(hass, config_entry)
assert hass.data[TWINKLY_DOMAIN].get(id) is None

View File

@ -0,0 +1,224 @@
"""Tests for the integration of a twinly device."""
from typing import Tuple
from homeassistant.components.twinkly.const import (
CONF_ENTRY_HOST,
CONF_ENTRY_ID,
CONF_ENTRY_MODEL,
CONF_ENTRY_NAME,
DOMAIN as TWINKLY_DOMAIN,
)
from homeassistant.components.twinkly.light import TwinklyLight
from homeassistant.core import HomeAssistant
from homeassistant.helpers.device_registry import DeviceEntry
from homeassistant.helpers.entity_registry import RegistryEntry
from tests.async_mock import patch
from tests.common import MockConfigEntry
from tests.components.twinkly import (
TEST_HOST,
TEST_ID,
TEST_MODEL,
TEST_NAME_ORIGINAL,
ClientMock,
)
async def test_missing_client(hass: HomeAssistant):
"""Validate that if client has not been setup, it fails immediately in setup."""
try:
config_entry = MockConfigEntry(
data={
CONF_ENTRY_HOST: TEST_HOST,
CONF_ENTRY_ID: TEST_ID,
CONF_ENTRY_NAME: TEST_NAME_ORIGINAL,
CONF_ENTRY_MODEL: TEST_MODEL,
}
)
TwinklyLight(config_entry, hass)
except ValueError:
return
assert False
async def test_initial_state(hass: HomeAssistant):
"""Validate that entity and device states are updated on startup."""
entity, device, _ = await _create_entries(hass)
state = hass.states.get(entity.entity_id)
# Basic state properties
assert state.name == entity.unique_id
assert state.state == "on"
assert state.attributes["host"] == TEST_HOST
assert state.attributes["brightness"] == 26
assert state.attributes["friendly_name"] == entity.unique_id
assert state.attributes["icon"] == "mdi:string-lights"
# Validates that custom properties of the API device_info are propagated through attributes
assert state.attributes["uuid"] == entity.unique_id
assert entity.original_name == entity.unique_id
assert entity.original_icon == "mdi:string-lights"
assert device.name == entity.unique_id
assert device.model == TEST_MODEL
assert device.manufacturer == "LEDWORKS"
async def test_initial_state_offline(hass: HomeAssistant):
"""Validate that entity and device are restored from config is offline on startup."""
client = ClientMock()
client.is_offline = True
entity, device, _ = await _create_entries(hass, client)
state = hass.states.get(entity.entity_id)
assert state.name == TEST_NAME_ORIGINAL
assert state.state == "unavailable"
assert state.attributes["friendly_name"] == TEST_NAME_ORIGINAL
assert state.attributes["icon"] == "mdi:string-lights"
assert entity.original_name == TEST_NAME_ORIGINAL
assert entity.original_icon == "mdi:string-lights"
assert device.name == TEST_NAME_ORIGINAL
assert device.model == TEST_MODEL
assert device.manufacturer == "LEDWORKS"
async def test_turn_on(hass: HomeAssistant):
"""Test support of the light.turn_on service."""
client = ClientMock()
client.is_on = False
client.brightness = 20
entity, _, _ = await _create_entries(hass, client)
assert hass.states.get(entity.entity_id).state == "off"
await hass.services.async_call(
"light", "turn_on", service_data={"entity_id": entity.entity_id}
)
await hass.async_block_till_done()
state = hass.states.get(entity.entity_id)
assert state.state == "on"
assert state.attributes["brightness"] == 51
async def test_turn_on_with_brightness(hass: HomeAssistant):
"""Test support of the light.turn_on service with a brightness parameter."""
client = ClientMock()
client.is_on = False
client.brightness = 20
entity, _, _ = await _create_entries(hass, client)
assert hass.states.get(entity.entity_id).state == "off"
await hass.services.async_call(
"light",
"turn_on",
service_data={"entity_id": entity.entity_id, "brightness": 255},
)
await hass.async_block_till_done()
state = hass.states.get(entity.entity_id)
assert state.state == "on"
assert state.attributes["brightness"] == 255
async def test_turn_off(hass: HomeAssistant):
"""Test support of the light.turn_off service."""
entity, _, _ = await _create_entries(hass)
assert hass.states.get(entity.entity_id).state == "on"
await hass.services.async_call(
"light", "turn_off", service_data={"entity_id": entity.entity_id}
)
await hass.async_block_till_done()
state = hass.states.get(entity.entity_id)
assert state.state == "off"
assert state.attributes["brightness"] == 0
async def test_update_name(hass: HomeAssistant):
"""
Validate device's name update behavior.
Validate that if device name is changed from the Twinkly app,
then the name of the entity is updated and it's also persisted,
so it can be restored when starting HA while Twinkly is offline.
"""
entity, _, client = await _create_entries(hass)
updated_config_entry = None
async def on_update(ha, co):
nonlocal updated_config_entry
updated_config_entry = co
hass.config_entries.async_get_entry(entity.unique_id).add_update_listener(on_update)
client.change_name("new_device_name")
await hass.services.async_call(
"light", "turn_off", service_data={"entity_id": entity.entity_id}
) # We call turn_off which will automatically cause an async_update
await hass.async_block_till_done()
state = hass.states.get(entity.entity_id)
assert updated_config_entry is not None
assert updated_config_entry.data[CONF_ENTRY_NAME] == "new_device_name"
assert state.attributes["friendly_name"] == "new_device_name"
async def test_unload(hass: HomeAssistant):
"""Validate that entities can be unloaded from the UI."""
_, _, client = await _create_entries(hass)
entry_id = client.id
assert await hass.config_entries.async_unload(entry_id)
async def _create_entries(
hass: HomeAssistant, client=None
) -> Tuple[RegistryEntry, DeviceEntry, ClientMock]:
client = ClientMock() if client is None else client
def get_client_mock(client, _):
return client
with patch("twinkly_client.TwinklyClient", side_effect=get_client_mock):
config_entry = MockConfigEntry(
domain=TWINKLY_DOMAIN,
data={
CONF_ENTRY_HOST: client,
CONF_ENTRY_ID: client.id,
CONF_ENTRY_NAME: TEST_NAME_ORIGINAL,
CONF_ENTRY_MODEL: TEST_MODEL,
},
entry_id=client.id,
)
config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(client.id)
await hass.async_block_till_done()
device_registry = await hass.helpers.device_registry.async_get_registry()
entity_registry = await hass.helpers.entity_registry.async_get_registry()
entity_id = entity_registry.async_get_entity_id("light", TWINKLY_DOMAIN, client.id)
entity = entity_registry.async_get(entity_id)
device = device_registry.async_get_device({(TWINKLY_DOMAIN, client.id)}, set())
assert entity is not None
assert device is not None
return entity, device, client