Add Zerproc integration (#35477)

This commit is contained in:
Emily Mills 2020-05-12 18:26:44 -04:00 committed by GitHub
parent 3936cbebbb
commit 306f15723d
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
15 changed files with 724 additions and 0 deletions

View File

@ -886,6 +886,8 @@ omit =
homeassistant/components/zamg/weather.py
homeassistant/components/zengge/light.py
homeassistant/components/zeroconf/*
homeassistant/components/zerproc/__init__.py
homeassistant/components/zerproc/const.py
homeassistant/components/zestimate/sensor.py
homeassistant/components/zha/api.py
homeassistant/components/zha/core/channels/*

View File

@ -468,6 +468,7 @@ homeassistant/components/yessssms/* @flowolf
homeassistant/components/yi/* @bachya
homeassistant/components/yr/* @danielhiversen
homeassistant/components/zeroconf/* @robbiet480 @Kane610
homeassistant/components/zerproc/* @emlove
homeassistant/components/zha/* @dmulcahey @adminiuga
homeassistant/components/zone/* @home-assistant/core
homeassistant/components/zoneminder/* @rohankapoorcom

View File

@ -0,0 +1,40 @@
"""Zerproc lights integration."""
import asyncio
from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry
from homeassistant.core import HomeAssistant
from .const import DOMAIN
PLATFORMS = ["light"]
async def async_setup(hass, config):
"""Set up the Zerproc platform."""
hass.async_create_task(
hass.config_entries.flow.async_init(DOMAIN, context={"source": SOURCE_IMPORT})
)
return True
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry):
"""Set up Zerproc from a config entry."""
for component in PLATFORMS:
hass.async_create_task(
hass.config_entries.async_forward_entry_setup(entry, component)
)
return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry):
"""Unload a config entry."""
return all(
await asyncio.gather(
*[
hass.config_entries.async_forward_entry_unload(entry, component)
for component in PLATFORMS
]
)
)

View File

@ -0,0 +1,26 @@
"""Config flow for Zerproc."""
import logging
import pyzerproc
from homeassistant import config_entries
from homeassistant.helpers import config_entry_flow
from .const import DOMAIN
_LOGGER = logging.getLogger(__name__)
async def _async_has_devices(hass) -> bool:
"""Return if there are devices that can be discovered."""
try:
devices = await hass.async_add_executor_job(pyzerproc.discover)
return len(devices) > 0
except pyzerproc.ZerprocException:
_LOGGER.error("Unable to discover nearby Zerproc devices", exc_info=True)
return False
config_entry_flow.register_discovery_flow(
DOMAIN, "Zerproc", _async_has_devices, config_entries.CONN_CLASS_LOCAL_POLL
)

View File

@ -0,0 +1,2 @@
"""Constants for the Zerproc integration."""
DOMAIN = "zerproc"

View File

@ -0,0 +1,203 @@
"""Zerproc light platform."""
from datetime import timedelta
import logging
from typing import Callable, List
import pyzerproc
from homeassistant.components.light import (
ATTR_BRIGHTNESS,
ATTR_HS_COLOR,
SUPPORT_BRIGHTNESS,
SUPPORT_COLOR,
Light,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import EVENT_HOMEASSISTANT_STOP
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity import Entity
from homeassistant.helpers.event import async_track_time_interval
from homeassistant.helpers.typing import HomeAssistantType
import homeassistant.util.color as color_util
from .const import DOMAIN
_LOGGER = logging.getLogger(__name__)
SUPPORT_ZERPROC = SUPPORT_BRIGHTNESS | SUPPORT_COLOR
DISCOVERY_INTERVAL = timedelta(seconds=60)
PARALLEL_UPDATES = 0
def connect_lights(lights: List[pyzerproc.Light]) -> List[pyzerproc.Light]:
"""Attempt to connect to lights, and return the connected lights."""
connected = []
for light in lights:
try:
light.connect(auto_reconnect=True)
connected.append(light)
except pyzerproc.ZerprocException:
_LOGGER.debug("Unable to connect to '%s'", light.address, exc_info=True)
return connected
def discover_entities(hass: HomeAssistant) -> List[Entity]:
"""Attempt to discover new lights."""
lights = pyzerproc.discover()
# Filter out already discovered lights
new_lights = [
light for light in lights if light.address not in hass.data[DOMAIN]["addresses"]
]
entities = []
for light in connect_lights(new_lights):
# Double-check the light hasn't been added in another thread
if light.address not in hass.data[DOMAIN]["addresses"]:
hass.data[DOMAIN]["addresses"].add(light.address)
entities.append(ZerprocLight(light))
return entities
async def async_setup_entry(
hass: HomeAssistantType,
config_entry: ConfigEntry,
async_add_entities: Callable[[List[Entity], bool], None],
) -> None:
"""Set up Abode light devices."""
if DOMAIN not in hass.data:
hass.data[DOMAIN] = {}
if "addresses" not in hass.data[DOMAIN]:
hass.data[DOMAIN]["addresses"] = set()
warned = False
async def discover(*args):
"""Wrap discovery to include params."""
nonlocal warned
try:
entities = await hass.async_add_executor_job(discover_entities, hass)
async_add_entities(entities, update_before_add=True)
warned = False
except pyzerproc.ZerprocException:
if warned is False:
_LOGGER.warning("Error discovering Zerproc lights", exc_info=True)
warned = True
# Initial discovery
hass.async_create_task(discover())
# Perform recurring discovery of new devices
async_track_time_interval(hass, discover, DISCOVERY_INTERVAL)
class ZerprocLight(Light):
"""Representation of an Zerproc Light."""
def __init__(self, light):
"""Initialize a Zerproc light."""
self._light = light
self._name = None
self._is_on = None
self._hs_color = None
self._brightness = None
self._available = True
async def async_added_to_hass(self) -> None:
"""Run when entity about to be added to hass."""
self.async_on_remove(
self.hass.bus.async_listen_once(
EVENT_HOMEASSISTANT_STOP, self.on_hass_shutdown
)
)
async def async_will_remove_from_hass(self) -> None:
"""Run when entity will be removed from hass."""
await self.hass.async_add_executor_job(self._light.disconnect)
def on_hass_shutdown(self, event):
"""Execute when Home Assistant is shutting down."""
self._light.disconnect()
@property
def name(self):
"""Return the display name of this light."""
return self._light.name
@property
def unique_id(self):
"""Return the ID of this light."""
return self._light.address
@property
def device_info(self):
"""Device info for this light."""
return {
"identifiers": {(DOMAIN, self.unique_id)},
"name": self.name,
"manufacturer": "Zerproc",
}
@property
def supported_features(self):
"""Flag supported features."""
return SUPPORT_ZERPROC
@property
def brightness(self):
"""Return the brightness of the light."""
return self._brightness
@property
def hs_color(self):
"""Return the hs color."""
return self._hs_color
@property
def is_on(self):
"""Return true if light is on."""
return self._is_on
@property
def available(self) -> bool:
"""Return True if entity is available."""
return self._available
def turn_on(self, **kwargs):
"""Instruct the light to turn on."""
if ATTR_BRIGHTNESS in kwargs or ATTR_HS_COLOR in kwargs:
default_hs = (0, 0) if self._hs_color is None else self._hs_color
hue_sat = kwargs.get(ATTR_HS_COLOR, default_hs)
default_brightness = 255 if self._brightness is None else self._brightness
brightness = kwargs.get(ATTR_BRIGHTNESS, default_brightness)
rgb = color_util.color_hsv_to_RGB(*hue_sat, brightness / 255 * 100)
self._light.set_color(*rgb)
else:
self._light.turn_on()
def turn_off(self, **kwargs):
"""Instruct the light to turn off."""
self._light.turn_off()
def update(self):
"""Fetch new state data for this light."""
try:
state = self._light.get_state()
except pyzerproc.ZerprocException:
if self._available:
_LOGGER.warning("Unable to connect to %s", self.entity_id)
self._available = False
return
if self._available is False:
_LOGGER.info("Reconnected to %s", self.entity_id)
self._available = True
self._is_on = state.is_on
hsv = color_util.color_RGB_to_hsv(*state.color)
self._hs_color = hsv[:2]
self._brightness = int(round((hsv[2] / 100) * 255))

View File

@ -0,0 +1,12 @@
{
"domain": "zerproc",
"name": "Zerproc",
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/zerproc",
"requirements": [
"pyzerproc==0.2.4"
],
"codeowners": [
"@emlove"
]
}

View File

@ -0,0 +1,14 @@
{
"title": "Zerproc",
"config": {
"step": {
"confirm": {
"description": "[%key:common::config_flow::description::confirm_setup%]"
}
},
"abort": {
"single_instance_allowed": "[%key:common::config_flow::abort::single_instance_allowed%]",
"no_devices_found": "[%key:common::config_flow::abort::no_devices_found%]"
}
}
}

View File

@ -0,0 +1,14 @@
{
"config": {
"abort": {
"no_devices_found": "No Zerproc lights found nearby.",
"single_instance_allowed": "Only a single configuration of Zerproc lights is necessary."
},
"step": {
"confirm": {
"description": "Do you want to set up Zerproc lights?"
}
}
},
"title": "Zerproc"
}

View File

@ -158,6 +158,7 @@ FLOWS = [
"wled",
"wwlln",
"xiaomi_miio",
"zerproc",
"zha",
"zwave",
"zwave_mqtt"

View File

@ -1813,6 +1813,9 @@ pyzabbix==0.7.4
# homeassistant.components.qrcode
pyzbar==0.1.7
# homeassistant.components.zerproc
pyzerproc==0.2.4
# homeassistant.components.qnap
qnapstats==0.3.0

View File

@ -737,6 +737,9 @@ pyvizio==0.1.47
# homeassistant.components.html5
pywebpush==1.9.2
# homeassistant.components.zerproc
pyzerproc==0.2.4
# homeassistant.components.rachio
rachiopy==0.1.3

View File

@ -0,0 +1 @@
"""Tests for the zerproc integration."""

View File

@ -0,0 +1,86 @@
"""Test the zerproc config flow."""
from asynctest import patch
import pyzerproc
from homeassistant import config_entries, setup
from homeassistant.components.zerproc.config_flow import DOMAIN
async def test_flow_success(hass):
"""Test we get the form."""
await setup.async_setup_component(hass, "persistent_notification", {})
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == "form"
assert result["errors"] is None
with patch(
"homeassistant.components.zerproc.config_flow.pyzerproc.discover",
return_value=["Light1", "Light2"],
), patch(
"homeassistant.components.zerproc.async_setup", return_value=True
) as mock_setup, patch(
"homeassistant.components.zerproc.async_setup_entry", return_value=True,
) as mock_setup_entry:
result2 = await hass.config_entries.flow.async_configure(result["flow_id"], {},)
assert result2["type"] == "create_entry"
assert result2["title"] == "Zerproc"
assert result2["data"] == {}
await hass.async_block_till_done()
assert len(mock_setup.mock_calls) == 1
assert len(mock_setup_entry.mock_calls) == 1
async def test_flow_no_devices_found(hass):
"""Test we get the form."""
await setup.async_setup_component(hass, "persistent_notification", {})
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == "form"
assert result["errors"] is None
with patch(
"homeassistant.components.zerproc.config_flow.pyzerproc.discover",
return_value=[],
), patch(
"homeassistant.components.zerproc.async_setup", return_value=True
) as mock_setup, patch(
"homeassistant.components.zerproc.async_setup_entry", return_value=True,
) as mock_setup_entry:
result2 = await hass.config_entries.flow.async_configure(result["flow_id"], {},)
assert result2["type"] == "abort"
assert result2["reason"] == "no_devices_found"
await hass.async_block_till_done()
assert len(mock_setup.mock_calls) == 0
assert len(mock_setup_entry.mock_calls) == 0
async def test_flow_exceptions_caught(hass):
"""Test we get the form."""
await setup.async_setup_component(hass, "persistent_notification", {})
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == "form"
assert result["errors"] is None
with patch(
"homeassistant.components.zerproc.config_flow.pyzerproc.discover",
side_effect=pyzerproc.ZerprocException("TEST"),
), patch(
"homeassistant.components.zerproc.async_setup", return_value=True
) as mock_setup, patch(
"homeassistant.components.zerproc.async_setup_entry", return_value=True,
) as mock_setup_entry:
result2 = await hass.config_entries.flow.async_configure(result["flow_id"], {},)
assert result2["type"] == "abort"
assert result2["reason"] == "no_devices_found"
await hass.async_block_till_done()
assert len(mock_setup.mock_calls) == 0
assert len(mock_setup_entry.mock_calls) == 0

View File

@ -0,0 +1,316 @@
"""Test the zerproc lights."""
from asynctest import patch
import pytest
import pyzerproc
from homeassistant import setup
from homeassistant.components.light import (
ATTR_BRIGHTNESS,
ATTR_HS_COLOR,
ATTR_RGB_COLOR,
ATTR_XY_COLOR,
SCAN_INTERVAL,
SUPPORT_BRIGHTNESS,
SUPPORT_COLOR,
)
from homeassistant.components.zerproc.light import DOMAIN
from homeassistant.const import (
ATTR_ENTITY_ID,
ATTR_FRIENDLY_NAME,
ATTR_SUPPORTED_FEATURES,
STATE_OFF,
STATE_ON,
STATE_UNAVAILABLE,
)
import homeassistant.util.dt as dt_util
from tests.common import MockConfigEntry, async_fire_time_changed
@pytest.fixture
async def mock_light(hass):
"""Create a mock light entity."""
await setup.async_setup_component(hass, "persistent_notification", {})
mock_entry = MockConfigEntry(domain=DOMAIN)
mock_entry.add_to_hass(hass)
light = pyzerproc.Light("AA:BB:CC:DD:EE:FF", "LEDBlue-CCDDEEFF")
mock_state = pyzerproc.LightState(False, (0, 0, 0))
with patch(
"homeassistant.components.zerproc.light.pyzerproc.discover",
return_value=[light],
), patch.object(light, "connect"), patch.object(
light, "get_state", return_value=mock_state
):
await hass.config_entries.async_setup(mock_entry.entry_id)
await hass.async_block_till_done()
return light
async def test_init(hass):
"""Test platform setup."""
await setup.async_setup_component(hass, "persistent_notification", {})
mock_entry = MockConfigEntry(domain=DOMAIN)
mock_entry.add_to_hass(hass)
mock_light_1 = pyzerproc.Light("AA:BB:CC:DD:EE:FF", "LEDBlue-CCDDEEFF")
mock_light_2 = pyzerproc.Light("11:22:33:44:55:66", "LEDBlue-33445566")
mock_state_1 = pyzerproc.LightState(False, (0, 0, 0))
mock_state_2 = pyzerproc.LightState(True, (0, 80, 255))
with patch(
"homeassistant.components.zerproc.light.pyzerproc.discover",
return_value=[mock_light_1, mock_light_2],
), patch.object(mock_light_1, "connect"), patch.object(
mock_light_2, "connect"
), patch.object(
mock_light_1, "get_state", return_value=mock_state_1
), patch.object(
mock_light_2, "get_state", return_value=mock_state_2
):
await hass.config_entries.async_setup(mock_entry.entry_id)
await hass.async_block_till_done()
state = hass.states.get("light.ledblue_ccddeeff")
assert state.state == STATE_OFF
assert state.attributes == {
ATTR_FRIENDLY_NAME: "LEDBlue-CCDDEEFF",
ATTR_SUPPORTED_FEATURES: SUPPORT_BRIGHTNESS | SUPPORT_COLOR,
}
state = hass.states.get("light.ledblue_33445566")
assert state.state == STATE_ON
assert state.attributes == {
ATTR_FRIENDLY_NAME: "LEDBlue-33445566",
ATTR_SUPPORTED_FEATURES: SUPPORT_BRIGHTNESS | SUPPORT_COLOR,
ATTR_BRIGHTNESS: 255,
ATTR_HS_COLOR: (221.176, 100.0),
ATTR_RGB_COLOR: (0, 80, 255),
ATTR_XY_COLOR: (0.138, 0.08),
}
with patch.object(hass.loop, "stop"), patch.object(
mock_light_1, "disconnect"
) as mock_disconnect_1, patch.object(
mock_light_2, "disconnect"
) as mock_disconnect_2:
await hass.async_stop()
assert mock_disconnect_1.called
assert mock_disconnect_2.called
async def test_discovery_exception(hass):
"""Test platform setup."""
await setup.async_setup_component(hass, "persistent_notification", {})
mock_entry = MockConfigEntry(domain=DOMAIN)
mock_entry.add_to_hass(hass)
with patch(
"homeassistant.components.zerproc.light.pyzerproc.discover",
side_effect=pyzerproc.ZerprocException("TEST"),
):
await hass.config_entries.async_setup(mock_entry.entry_id)
await hass.async_block_till_done()
# The exception should be captured and no entities should be added
assert len(hass.data[DOMAIN]["addresses"]) == 0
async def test_connect_exception(hass):
"""Test platform setup."""
await setup.async_setup_component(hass, "persistent_notification", {})
mock_entry = MockConfigEntry(domain=DOMAIN)
mock_entry.add_to_hass(hass)
mock_light = pyzerproc.Light("AA:BB:CC:DD:EE:FF", "LEDBlue-CCDDEEFF")
with patch(
"homeassistant.components.zerproc.light.pyzerproc.discover",
return_value=[mock_light],
), patch.object(
mock_light, "connect", side_effect=pyzerproc.ZerprocException("TEST")
):
await hass.config_entries.async_setup(mock_entry.entry_id)
await hass.async_block_till_done()
# The exception should be captured and no entities should be added
assert len(hass.data[DOMAIN]["addresses"]) == 0
async def test_light_turn_on(hass, mock_light):
"""Test ZerprocLight turn_on."""
utcnow = dt_util.utcnow()
with patch.object(mock_light, "turn_on") as mock_turn_on:
await hass.services.async_call(
"light",
"turn_on",
{ATTR_ENTITY_ID: "light.ledblue_ccddeeff"},
blocking=True,
)
await hass.async_block_till_done()
mock_turn_on.assert_called()
with patch.object(mock_light, "set_color") as mock_set_color:
await hass.services.async_call(
"light",
"turn_on",
{ATTR_ENTITY_ID: "light.ledblue_ccddeeff", ATTR_BRIGHTNESS: 25},
blocking=True,
)
await hass.async_block_till_done()
mock_set_color.assert_called_with(25, 25, 25)
# Make sure no discovery calls are made while we emulate time passing
with patch("homeassistant.components.zerproc.light.pyzerproc.discover"):
with patch.object(
mock_light,
"get_state",
return_value=pyzerproc.LightState(True, (175, 150, 220)),
):
utcnow = utcnow + SCAN_INTERVAL
async_fire_time_changed(hass, utcnow)
await hass.async_block_till_done()
with patch.object(mock_light, "set_color") as mock_set_color:
await hass.services.async_call(
"light",
"turn_on",
{ATTR_ENTITY_ID: "light.ledblue_ccddeeff", ATTR_BRIGHTNESS: 25},
blocking=True,
)
await hass.async_block_till_done()
mock_set_color.assert_called_with(19, 17, 25)
with patch.object(mock_light, "set_color") as mock_set_color:
await hass.services.async_call(
"light",
"turn_on",
{ATTR_ENTITY_ID: "light.ledblue_ccddeeff", ATTR_HS_COLOR: (50, 50)},
blocking=True,
)
await hass.async_block_till_done()
mock_set_color.assert_called_with(220, 201, 110)
with patch.object(
mock_light,
"get_state",
return_value=pyzerproc.LightState(True, (75, 75, 75)),
):
utcnow = utcnow + SCAN_INTERVAL
async_fire_time_changed(hass, utcnow)
await hass.async_block_till_done()
with patch.object(mock_light, "set_color") as mock_set_color:
await hass.services.async_call(
"light",
"turn_on",
{ATTR_ENTITY_ID: "light.ledblue_ccddeeff", ATTR_HS_COLOR: (50, 50)},
blocking=True,
)
await hass.async_block_till_done()
mock_set_color.assert_called_with(75, 68, 37)
with patch.object(mock_light, "set_color") as mock_set_color:
await hass.services.async_call(
"light",
"turn_on",
{
ATTR_ENTITY_ID: "light.ledblue_ccddeeff",
ATTR_BRIGHTNESS: 200,
ATTR_HS_COLOR: (75, 75),
},
blocking=True,
)
await hass.async_block_till_done()
mock_set_color.assert_called_with(162, 200, 50)
async def test_light_turn_off(hass, mock_light):
"""Test ZerprocLight turn_on."""
with patch.object(mock_light, "turn_off") as mock_turn_off:
await hass.services.async_call(
"light",
"turn_off",
{ATTR_ENTITY_ID: "light.ledblue_ccddeeff"},
blocking=True,
)
await hass.async_block_till_done()
mock_turn_off.assert_called()
async def test_light_update(hass, mock_light):
"""Test ZerprocLight update."""
utcnow = dt_util.utcnow()
state = hass.states.get("light.ledblue_ccddeeff")
assert state.state == STATE_OFF
assert state.attributes == {
ATTR_FRIENDLY_NAME: "LEDBlue-CCDDEEFF",
ATTR_SUPPORTED_FEATURES: SUPPORT_BRIGHTNESS | SUPPORT_COLOR,
}
# Make sure no discovery calls are made while we emulate time passing
with patch("homeassistant.components.zerproc.light.pyzerproc.discover"):
# Test an exception during discovery
with patch.object(
mock_light, "get_state", side_effect=pyzerproc.ZerprocException("TEST")
):
utcnow = utcnow + SCAN_INTERVAL
async_fire_time_changed(hass, utcnow)
await hass.async_block_till_done()
state = hass.states.get("light.ledblue_ccddeeff")
assert state.state == STATE_UNAVAILABLE
assert state.attributes == {
ATTR_FRIENDLY_NAME: "LEDBlue-CCDDEEFF",
ATTR_SUPPORTED_FEATURES: SUPPORT_BRIGHTNESS | SUPPORT_COLOR,
}
with patch.object(
mock_light,
"get_state",
return_value=pyzerproc.LightState(False, (200, 128, 100)),
):
utcnow = utcnow + SCAN_INTERVAL
async_fire_time_changed(hass, utcnow)
await hass.async_block_till_done()
state = hass.states.get("light.ledblue_ccddeeff")
assert state.state == STATE_OFF
assert state.attributes == {
ATTR_FRIENDLY_NAME: "LEDBlue-CCDDEEFF",
ATTR_SUPPORTED_FEATURES: SUPPORT_BRIGHTNESS | SUPPORT_COLOR,
}
with patch.object(
mock_light,
"get_state",
return_value=pyzerproc.LightState(True, (175, 150, 220)),
):
utcnow = utcnow + SCAN_INTERVAL
async_fire_time_changed(hass, utcnow)
await hass.async_block_till_done()
state = hass.states.get("light.ledblue_ccddeeff")
assert state.state == STATE_ON
assert state.attributes == {
ATTR_FRIENDLY_NAME: "LEDBlue-CCDDEEFF",
ATTR_SUPPORTED_FEATURES: SUPPORT_BRIGHTNESS | SUPPORT_COLOR,
ATTR_BRIGHTNESS: 220,
ATTR_HS_COLOR: (261.429, 31.818),
ATTR_RGB_COLOR: (202, 173, 255),
ATTR_XY_COLOR: (0.291, 0.232),
}