Introduce wiz integration for the WiZ Platform (#44779)

Co-authored-by: Marvin Wichmann <marvin@fam-wichmann.de>
Co-authored-by: J. Nick Koston <nick@koston.org>
Co-authored-by: Jan Stienstra <65826735+j-stienstra@users.noreply.github.com>
Co-authored-by: Franck Nijhof <frenck@frenck.nl>
Co-authored-by: Paulus Schoutsen <paulus@home-assistant.io>
Co-authored-by: Franck Nijhof <git@frenck.dev>
Co-authored-by: Paulus Schoutsen <balloob@gmail.com>
This commit is contained in:
Stephan Traub 2022-02-05 01:20:21 +01:00 committed by GitHub
parent d8830aa4e0
commit 432d9a8f19
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
14 changed files with 722 additions and 0 deletions

View File

@ -1315,6 +1315,9 @@ omit =
homeassistant/components/waze_travel_time/sensor.py
homeassistant/components/wiffi/*
homeassistant/components/wirelesstag/*
homeassistant/components/wiz/__init__.py
homeassistant/components/wiz/const.py
homeassistant/components/wiz/light.py
homeassistant/components/wolflink/__init__.py
homeassistant/components/wolflink/sensor.py
homeassistant/components/wolflink/const.py

View File

@ -1065,6 +1065,8 @@ tests/components/wilight/* @leofig-rj
homeassistant/components/wirelesstag/* @sergeymaysak
homeassistant/components/withings/* @vangorra
tests/components/withings/* @vangorra
homeassistant/components/wiz/* @sbidy
tests/components/wiz/* @sbidy
homeassistant/components/wled/* @frenck
tests/components/wled/* @frenck
homeassistant/components/wolflink/* @adamkrol93

View File

@ -0,0 +1,52 @@
"""WiZ Platform integration."""
from dataclasses import dataclass
import logging
from pywizlight import wizlight
from pywizlight.exceptions import WizLightConnectionError, WizLightTimeOutError
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_HOST
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady
from .const import DOMAIN
_LOGGER = logging.getLogger(__name__)
PLATFORMS = ["light"]
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up the wiz integration from a config entry."""
ip_address = entry.data.get(CONF_HOST)
_LOGGER.debug("Get bulb with IP: %s", ip_address)
try:
bulb = wizlight(ip_address)
scenes = await bulb.getSupportedScenes()
await bulb.getMac()
except (
WizLightTimeOutError,
WizLightConnectionError,
ConnectionRefusedError,
) as err:
raise ConfigEntryNotReady from err
hass.data.setdefault(DOMAIN, {})[entry.entry_id] = WizData(bulb=bulb, scenes=scenes)
hass.config_entries.async_setup_platforms(entry, PLATFORMS)
return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry."""
if unload_ok := await hass.config_entries.async_unload_platforms(entry, PLATFORMS):
hass.data[DOMAIN].pop(entry.entry_id)
return unload_ok
@dataclass
class WizData:
"""Data for the wiz integration."""
bulb: wizlight
scenes: list

View File

@ -0,0 +1,52 @@
"""Config flow for WiZ Platform."""
import logging
from pywizlight import wizlight
from pywizlight.exceptions import WizLightConnectionError, WizLightTimeOutError
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.const import CONF_HOST, CONF_NAME
from .const import DEFAULT_NAME, DOMAIN
_LOGGER = logging.getLogger(__name__)
STEP_USER_DATA_SCHEMA = vol.Schema(
{
vol.Required(CONF_HOST): str,
vol.Optional(CONF_NAME, default=DEFAULT_NAME): str,
}
)
class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle a config flow for WiZ."""
VERSION = 1
async def async_step_user(self, user_input=None):
"""Handle a flow initialized by the user."""
errors = {}
if user_input is not None:
bulb = wizlight(user_input[CONF_HOST])
try:
mac = await bulb.getMac()
except WizLightTimeOutError:
errors["base"] = "bulb_time_out"
except ConnectionRefusedError:
errors["base"] = "cannot_connect"
except WizLightConnectionError:
errors["base"] = "no_wiz_light"
except Exception: # pylint: disable=broad-except
_LOGGER.exception("Unexpected exception")
errors["base"] = "unknown"
else:
await self.async_set_unique_id(mac)
self._abort_if_unique_id_configured()
return self.async_create_entry(
title=user_input[CONF_NAME], data=user_input
)
return self.async_show_form(
step_id="user", data_schema=STEP_USER_DATA_SCHEMA, errors=errors
)

View File

@ -0,0 +1,4 @@
"""Constants for the WiZ Platform integration."""
DOMAIN = "wiz"
DEFAULT_NAME = "WiZ"

View File

@ -0,0 +1,348 @@
"""WiZ integration."""
from __future__ import annotations
from datetime import timedelta
import logging
from pywizlight import PilotBuilder, wizlight
from pywizlight.bulblibrary import BulbClass, BulbType
from pywizlight.exceptions import WizLightNotKnownBulb, WizLightTimeOutError
from pywizlight.rgbcw import convertHSfromRGBCW
from pywizlight.scenes import get_id_from_scene_name
from homeassistant.components.light import (
ATTR_BRIGHTNESS,
ATTR_COLOR_TEMP,
ATTR_EFFECT,
ATTR_HS_COLOR,
ATTR_RGB_COLOR,
SUPPORT_BRIGHTNESS,
SUPPORT_COLOR,
SUPPORT_COLOR_TEMP,
SUPPORT_EFFECT,
LightEntity,
)
from homeassistant.const import CONF_NAME
from homeassistant.helpers.device_registry import CONNECTION_NETWORK_MAC
import homeassistant.util.color as color_utils
from .const import DOMAIN
_LOGGER = logging.getLogger(__name__)
SUPPORT_FEATURES_RGB = (
SUPPORT_BRIGHTNESS | SUPPORT_COLOR | SUPPORT_COLOR_TEMP | SUPPORT_EFFECT
)
# set poll interval to 15 sec because of changes from external to the bulb
SCAN_INTERVAL = timedelta(seconds=15)
async def async_setup_entry(hass, entry, async_add_entities):
"""Set up the WiZ Platform from config_flow."""
# Assign configuration variables.
wiz_data = hass.data[DOMAIN][entry.entry_id]
wizbulb = WizBulbEntity(wiz_data.bulb, entry.data.get(CONF_NAME), wiz_data.scenes)
# Add devices with defined name
async_add_entities([wizbulb], update_before_add=True)
return True
class WizBulbEntity(LightEntity):
"""Representation of WiZ Light bulb."""
def __init__(self, light: wizlight, name, scenes):
"""Initialize an WiZLight."""
self._light = light
self._state = None
self._brightness = None
self._attr_name = name
self._rgb_color = None
self._temperature = None
self._hscolor = None
self._available = None
self._effect = None
self._scenes: list[str] = scenes
self._bulbtype: BulbType = light.bulbtype
self._mac = light.mac
self._attr_unique_id = light.mac
# new init states
self._attr_min_mireds = self.get_min_mireds()
self._attr_max_mireds = self.get_max_mireds()
self._attr_supported_features = self.get_supported_features()
@property
def brightness(self):
"""Return the brightness of the light."""
return self._brightness
@property
def rgb_color(self):
"""Return the color property."""
return self._rgb_color
@property
def hs_color(self):
"""Return the hs color value."""
return self._hscolor
@property
def is_on(self):
"""Return true if light is on."""
return self._state
async def async_turn_on(self, **kwargs):
"""Instruct the light to turn on."""
brightness = None
if ATTR_BRIGHTNESS in kwargs:
brightness = kwargs.get(ATTR_BRIGHTNESS)
if ATTR_RGB_COLOR in kwargs:
pilot = PilotBuilder(rgb=kwargs.get(ATTR_RGB_COLOR), brightness=brightness)
if ATTR_HS_COLOR in kwargs:
pilot = PilotBuilder(
hucolor=(kwargs[ATTR_HS_COLOR][0], kwargs[ATTR_HS_COLOR][1]),
brightness=brightness,
)
else:
colortemp = None
if ATTR_COLOR_TEMP in kwargs:
kelvin = color_utils.color_temperature_mired_to_kelvin(
kwargs[ATTR_COLOR_TEMP]
)
colortemp = kelvin
_LOGGER.debug(
"[wizlight %s] kelvin changed and send to bulb: %s",
self._light.ip,
colortemp,
)
sceneid = None
if ATTR_EFFECT in kwargs:
sceneid = get_id_from_scene_name(kwargs[ATTR_EFFECT])
if sceneid == 1000: # rhythm
pilot = PilotBuilder()
else:
pilot = PilotBuilder(
brightness=brightness, colortemp=colortemp, scene=sceneid
)
_LOGGER.debug(
"[wizlight %s] Pilot will be send with brightness=%s, colortemp=%s, scene=%s",
self._light.ip,
brightness,
colortemp,
sceneid,
)
sceneid = None
if ATTR_EFFECT in kwargs:
sceneid = get_id_from_scene_name(kwargs[ATTR_EFFECT])
if sceneid == 1000: # rhythm
pilot = PilotBuilder()
else:
pilot = PilotBuilder(
brightness=brightness, colortemp=colortemp, scene=sceneid
)
await self._light.turn_on(pilot)
async def async_turn_off(self, **kwargs):
"""Instruct the light to turn off."""
await self._light.turn_off()
@property
def color_temp(self):
"""Return the CT color value in mireds."""
return self._temperature
def get_min_mireds(self) -> int:
"""Return the coldest color_temp that this light supports."""
if self._bulbtype is None:
return color_utils.color_temperature_kelvin_to_mired(6500)
# DW bulbs have no kelvin
if self._bulbtype.bulb_type == BulbClass.DW:
return 0
# If bulbtype is TW or RGB then return the kelvin value
try:
return color_utils.color_temperature_kelvin_to_mired(
self._bulbtype.kelvin_range.max
)
except WizLightNotKnownBulb:
_LOGGER.debug("Kelvin is not present in the library. Fallback to 6500")
return color_utils.color_temperature_kelvin_to_mired(6500)
def get_max_mireds(self) -> int:
"""Return the warmest color_temp that this light supports."""
if self._bulbtype is None:
return color_utils.color_temperature_kelvin_to_mired(2200)
# DW bulbs have no kelvin
if self._bulbtype.bulb_type == BulbClass.DW:
return 0
# If bulbtype is TW or RGB then return the kelvin value
try:
return color_utils.color_temperature_kelvin_to_mired(
self._bulbtype.kelvin_range.min
)
except WizLightNotKnownBulb:
_LOGGER.debug("Kelvin is not present in the library. Fallback to 2200")
return color_utils.color_temperature_kelvin_to_mired(2200)
def get_supported_features(self) -> int:
"""Flag supported features."""
if not self._bulbtype:
# fallback
return SUPPORT_FEATURES_RGB
features = 0
try:
# Map features for better reading
if self._bulbtype.features.brightness:
features = features | SUPPORT_BRIGHTNESS
if self._bulbtype.features.color:
features = features | SUPPORT_COLOR
if self._bulbtype.features.effect:
features = features | SUPPORT_EFFECT
if self._bulbtype.features.color_tmp:
features = features | SUPPORT_COLOR_TEMP
return features
except WizLightNotKnownBulb:
_LOGGER.warning(
"Bulb is not present in the library. Fallback to full feature"
)
return SUPPORT_FEATURES_RGB
@property
def effect(self):
"""Return the current effect."""
return self._effect
@property
def effect_list(self):
"""Return the list of supported effects.
URL: https://docs.pro.wizconnected.com/#light-modes
"""
return self._scenes
@property
def available(self):
"""Return if light is available."""
return self._available
async def async_update(self):
"""Fetch new state data for this light."""
await self.update_state()
if self._state is not None and self._state is not False:
self.update_brightness()
self.update_temperature()
self.update_color()
self.update_effect()
@property
def device_info(self):
"""Get device specific attributes."""
return {
"connections": {(CONNECTION_NETWORK_MAC, self._mac)},
"name": self._attr_name,
"manufacturer": "WiZ Light Platform",
"model": self._bulbtype.name,
}
def update_state_available(self):
"""Update the state if bulb is available."""
self._state = self._light.status
self._available = True
def update_state_unavailable(self):
"""Update the state if bulb is unavailable."""
self._state = False
self._available = False
async def update_state(self):
"""Update the state."""
try:
await self._light.updateState()
except (ConnectionRefusedError, TimeoutError, WizLightTimeOutError) as ex:
_LOGGER.debug(ex)
self.update_state_unavailable()
else:
if self._light.state is None:
self.update_state_unavailable()
else:
self.update_state_available()
_LOGGER.debug(
"[wizlight %s] updated state: %s and available: %s",
self._light.ip,
self._state,
self._available,
)
def update_brightness(self):
"""Update the brightness."""
if self._light.state.get_brightness() is None:
return
brightness = self._light.state.get_brightness()
if 0 <= int(brightness) <= 255:
self._brightness = int(brightness)
else:
_LOGGER.error(
"Received invalid brightness : %s. Expected: 0-255", brightness
)
self._brightness = None
def update_temperature(self):
"""Update the temperature."""
colortemp = self._light.state.get_colortemp()
if colortemp is None or colortemp == 0:
self._temperature = None
return
_LOGGER.debug(
"[wizlight %s] kelvin from the bulb: %s", self._light.ip, colortemp
)
temperature = color_utils.color_temperature_kelvin_to_mired(colortemp)
self._temperature = temperature
def update_color(self):
"""Update the hs color."""
colortemp = self._light.state.get_colortemp()
if colortemp is not None and colortemp != 0:
self._hscolor = None
return
if self._light.state.get_rgb() is None:
return
rgb = self._light.state.get_rgb()
if rgb[0] is None:
# this is the case if the temperature was changed
# do nothing until the RGB color was changed
return
warmwhite = self._light.state.get_warm_white()
if warmwhite is None:
return
self._hscolor = convertHSfromRGBCW(rgb, warmwhite)
def update_effect(self):
"""Update the bulb scene."""
self._effect = self._light.state.get_scene()
async def get_bulb_type(self):
"""Get the bulb type."""
if self._bulbtype is not None:
return self._bulbtype
try:
self._bulbtype = await self._light.get_bulbtype()
_LOGGER.info(
"[wizlight %s] Initiate the WiZ bulb as %s",
self._light.ip,
self._bulbtype.name,
)
except WizLightTimeOutError:
_LOGGER.debug(
"[wizlight %s] Bulbtype update failed - Timeout", self._light.ip
)

View File

@ -0,0 +1,13 @@
{
"domain": "wiz",
"name": "WiZ",
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/wiz",
"requirements": [
"pywizlight==0.4.15"
],
"iot_class": "local_polling",
"codeowners": [
"@sbidy"
]
}

View File

@ -0,0 +1,26 @@
{
"config": {
"step": {
"user": {
"data": {
"host": "[%key:common::config_flow::data::host%]",
"name": "[%key:common::config_flow::data::name%]"
},
"description": "Please enter a hostname or IP address and name to add a new bulb:"
},
"confirm": {
"description": "[%key:common::config_flow::description::confirm_setup%]"
}
},
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"unknown": "[%key:common::config_flow::error::unknown%]",
"bulb_time_out": "Can not connect to the bulb. Maybe the bulb is offline or a wrong IP/host was entered. Please turn on the light and try again!",
"no_wiz_light": "The bulb can not be connected via WiZ Platform integration."
},
"abort": {
"no_devices_found": "[%key:common::config_flow::abort::no_devices_found%]",
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]"
}
}
}

View File

@ -0,0 +1,26 @@
{
"config": {
"abort": {
"already_configured": "Device is already configured",
"no_devices_found": "No devices found on the network"
},
"error": {
"bulb_time_out": "Can not connect to the bulb. Maybe the bulb is offline or a wrong IP/host was entered. Please turn on the light and try again!",
"cannot_connect": "Failed to connect",
"no_wiz_light": "The bulb can not be connected via WiZ Platform integration.",
"unknown": "Unexpected error"
},
"step": {
"confirm": {
"description": "Do you want to add a new Bulb?"
},
"user": {
"data": {
"host": "Hostname or IP",
"name": "Name"
},
"description": "Please enter a hostname or IP address and name to add a new bulb:"
}
}
}
}

View File

@ -365,6 +365,7 @@ FLOWS = [
"wiffi",
"wilight",
"withings",
"wiz",
"wled",
"wolflink",
"xbox",

View File

@ -2053,6 +2053,9 @@ pywemo==0.7.0
# homeassistant.components.wilight
pywilight==0.0.70
# homeassistant.components.wiz
pywizlight==0.4.15
# homeassistant.components.xeoma
pyxeoma==1.4.1

View File

@ -1275,6 +1275,9 @@ pywemo==0.7.0
# homeassistant.components.wilight
pywilight==0.0.70
# homeassistant.components.wiz
pywizlight==0.4.15
# homeassistant.components.zerproc
pyzerproc==0.4.8

View File

@ -0,0 +1,61 @@
"""Tests for the WiZ Platform integration."""
import json
from homeassistant.components.wiz.const import DOMAIN
from homeassistant.const import CONF_IP_ADDRESS, CONF_NAME
from homeassistant.helpers.typing import HomeAssistantType
from tests.common import MockConfigEntry
FAKE_BULB_CONFIG = json.loads(
'{"method":"getSystemConfig","env":"pro","result":\
{"mac":"ABCABCABCABC",\
"homeId":653906,\
"roomId":989983,\
"moduleName":"ESP_0711_STR",\
"fwVersion":"1.21.0",\
"groupId":0,"drvConf":[20,2],\
"ewf":[255,0,255,255,0,0,0],\
"ewfHex":"ff00ffff000000",\
"ping":0}}'
)
REAL_BULB_CONFIG = json.loads(
'{"method":"getSystemConfig","env":"pro","result":\
{"mac":"ABCABCABCABC",\
"homeId":653906,\
"roomId":989983,\
"moduleName":"ESP01_SHRGB_03",\
"fwVersion":"1.21.0",\
"groupId":0,"drvConf":[20,2],\
"ewf":[255,0,255,255,0,0,0],\
"ewfHex":"ff00ffff000000",\
"ping":0}}'
)
TEST_SYSTEM_INFO = {"id": "ABCABCABCABC", "name": "Test Bulb"}
TEST_CONNECTION = {CONF_IP_ADDRESS: "1.1.1.1", CONF_NAME: "Test Bulb"}
async def setup_integration(
hass: HomeAssistantType,
) -> MockConfigEntry:
"""Mock ConfigEntry in Home Assistant."""
entry = MockConfigEntry(
domain=DOMAIN,
unique_id=TEST_SYSTEM_INFO["id"],
data={
CONF_IP_ADDRESS: "127.0.0.1",
CONF_NAME: TEST_SYSTEM_INFO["name"],
},
)
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
return entry

View File

@ -0,0 +1,128 @@
"""Test the WiZ Platform config flow."""
from unittest.mock import patch
import pytest
from homeassistant import config_entries
from homeassistant.components.wiz.config_flow import (
WizLightConnectionError,
WizLightTimeOutError,
)
from homeassistant.components.wiz.const import DOMAIN
from homeassistant.const import CONF_HOST, CONF_NAME
from tests.common import MockConfigEntry
FAKE_BULB_CONFIG = '{"method":"getSystemConfig","env":"pro","result":\
{"mac":"ABCABCABCABC",\
"homeId":653906,\
"roomId":989983,\
"moduleName":"ESP_0711_STR",\
"fwVersion":"1.21.0",\
"groupId":0,"drvConf":[20,2],\
"ewf":[255,0,255,255,0,0,0],\
"ewfHex":"ff00ffff000000",\
"ping":0}}'
TEST_SYSTEM_INFO = {"id": "ABCABCABCABC", "name": "Test Bulb"}
TEST_CONNECTION = {CONF_HOST: "1.1.1.1", CONF_NAME: "Test Bulb"}
TEST_NO_IP = {CONF_HOST: "this is no IP input", CONF_NAME: "Test Bulb"}
async def test_form(hass):
"""Test we get the form."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == "form"
assert result["errors"] == {}
# Patch functions
with patch(
"homeassistant.components.wiz.wizlight.getBulbConfig",
return_value=FAKE_BULB_CONFIG,
), patch(
"homeassistant.components.wiz.wizlight.getMac",
return_value="ABCABCABCABC",
) as mock_setup, patch(
"homeassistant.components.wiz.async_setup_entry",
return_value=True,
) as mock_setup_entry:
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
TEST_CONNECTION,
)
await hass.async_block_till_done()
assert result2["type"] == "create_entry"
assert result2["title"] == "Test Bulb"
assert result2["data"] == TEST_CONNECTION
assert len(mock_setup.mock_calls) == 1
assert len(mock_setup_entry.mock_calls) == 1
@pytest.mark.parametrize(
"side_effect, error_base",
[
(WizLightTimeOutError, "bulb_time_out"),
(WizLightConnectionError, "no_wiz_light"),
(Exception, "unknown"),
(ConnectionRefusedError, "cannot_connect"),
],
)
async def test_user_form_exceptions(hass, side_effect, error_base):
"""Test all user exceptions in the flow."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
with patch(
"homeassistant.components.wiz.wizlight.getBulbConfig",
side_effect=side_effect,
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
TEST_CONNECTION,
)
assert result2["type"] == "form"
assert result2["errors"] == {"base": error_base}
async def test_form_updates_unique_id(hass):
"""Test a duplicate id aborts and updates existing entry."""
entry = MockConfigEntry(
domain=DOMAIN,
unique_id=TEST_SYSTEM_INFO["id"],
data={
CONF_HOST: "dummy",
CONF_NAME: TEST_SYSTEM_INFO["name"],
"id": TEST_SYSTEM_INFO["id"],
},
)
entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
with patch(
"homeassistant.components.wiz.wizlight.getBulbConfig",
return_value=FAKE_BULB_CONFIG,
), patch(
"homeassistant.components.wiz.wizlight.getMac",
return_value="ABCABCABCABC",
), patch(
"homeassistant.components.wiz.async_setup_entry",
return_value=True,
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
TEST_CONNECTION,
)
await hass.async_block_till_done()
assert result2["type"] == "abort"
assert result2["reason"] == "already_configured"