evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add OZW WS API support for locks (#41186)

Browse source 
* add WS API support for locks and simplify lock service calls

* move return outside of try blocK

* fix rebase conflicts
This commit is contained in:
Raman Gupta 2020-10-06 08:20:48 -04:00 • committed by GitHub
parent e1578e5389
commit 06306f5dfe
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
4 changed files with 158 additions and 74 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

42
homeassistant/components/ozw/lock.py
View file

@ -1,7 +1,9 @@
"""Representation of Z-Wave locks."""
import logging
from openzwavemqtt.const import CommandClass, ValueIndex
from openzwavemqtt.const import ATTR_CODE_SLOT
from openzwavemqtt.exceptions import BaseOZWError
from openzwavemqtt.util.lock import clear_usercode, set_usercode
import voluptuous as vol
from homeassistant.components.lock import DOMAIN as LOCK_DOMAIN, LockEntity
@ -12,7 +14,6 @@ from homeassistant.helpers.dispatcher import async_dispatcher_connect
from .const import DATA_UNSUBSCRIBE, DOMAIN
from .entity import ZWaveDeviceEntity
ATTR_CODE_SLOT = "code_slot"
ATTR_USERCODE = "usercode"
SERVICE_SET_USERCODE = "set_usercode"
@ -54,6 +55,17 @@ async def async_setup_entry(hass, config_entry, async_add_entities):
)
def _call_util_lock_function(function, *args):
"""Call an openzwavemqtt.util.lock function and return success of call."""
try:
function(*args)
except BaseOZWError as err:
_LOGGER.error("%s: %s", type(err), err.args[0])
return False
return True
class ZWaveLock(ZWaveDeviceEntity, LockEntity):
"""Representation of a Z-Wave lock."""
@ -73,25 +85,15 @@ class ZWaveLock(ZWaveDeviceEntity, LockEntity):
@callback
def async_set_usercode(self, code_slot, usercode):
"""Set the usercode to index X on the lock."""
value = self.values.primary.node.get_value(CommandClass.USER_CODE, code_slot)
if len(str(usercode)) < 4:
_LOGGER.error(
"Invalid code provided: (%s) user code must be at least 4 digits",
usercode,
)
return
value.send_value(usercode)
_LOGGER.debug("User code at slot %s set", code_slot)
if _call_util_lock_function(
set_usercode, self.values.primary.node, code_slot, usercode
):
_LOGGER.debug("User code at slot %s set", code_slot)
@callback
def async_clear_usercode(self, code_slot):
"""Clear usercode in slot X on the lock."""
value = self.values.primary.node.get_value(
CommandClass.USER_CODE, ValueIndex.CLEAR_USER_CODE
)
value.send_value(code_slot)
# Sending twice because the first time it doesn't take
value.send_value(code_slot)
_LOGGER.info("Usercode at slot %s is cleared", code_slot)
if _call_util_lock_function(
clear_usercode, self.values.primary.node, code_slot
):
_LOGGER.info("Usercode at slot %s is cleared", code_slot)

144
homeassistant/components/ozw/websocket_api.py
View file

@ -1,5 +1,6 @@
"""Web socket API for OpenZWave."""
from openzwavemqtt.const import (
ATTR_CODE_SLOT,
ATTR_LABEL,
ATTR_POSITION,
ATTR_VALUE,
@ -7,6 +8,7 @@ from openzwavemqtt.const import (
EVENT_NODE_CHANGED,
)
from openzwavemqtt.exceptions import NotFoundError, NotSupportedError
from openzwavemqtt.util.lock import clear_usercode, get_code_slots, set_usercode
from openzwavemqtt.util.node import (
get_config_parameters,
get_node_from_manager,
@ -19,6 +21,7 @@ from homeassistant.core import callback
from homeassistant.helpers import config_validation as cv
from .const import ATTR_CONFIG_PARAMETER, ATTR_CONFIG_VALUE, DOMAIN, MANAGER, OPTIONS
from .lock import ATTR_USERCODE
TYPE = "type"
ID = "id"
@ -57,6 +60,50 @@ def async_register_api(hass):
websocket_api.async_register_command(hass, websocket_refresh_node_info)
websocket_api.async_register_command(hass, websocket_get_config_parameters)
websocket_api.async_register_command(hass, websocket_set_config_parameter)
websocket_api.async_register_command(hass, websocket_set_usercode)
websocket_api.async_register_command(hass, websocket_clear_usercode)
websocket_api.async_register_command(hass, websocket_get_code_slots)
def _call_util_function(hass, connection, msg, send_result, function, *args):
"""Call an openzwavemqtt.util function."""
try:
node = get_node_from_manager(
hass.data[DOMAIN][MANAGER], msg[OZW_INSTANCE], msg[NODE_ID]
)
except NotFoundError as err:
connection.send_error(
msg[ID],
websocket_api.const.ERR_NOT_FOUND,
err.args[0],
)
return
try:
payload = function(node, *args)
except NotFoundError as err:
connection.send_error(
msg[ID],
websocket_api.const.ERR_NOT_FOUND,
err.args[0],
)
return
except NotSupportedError as err:
connection.send_error(
msg[ID],
websocket_api.const.ERR_NOT_SUPPORTED,
err.args[0],
)
return
if send_result:
connection.send_result(
msg[ID],
payload,
)
return
connection.send_result(msg[ID])
@websocket_api.websocket_command({vol.Required(TYPE): "ozw/get_instances"})
@ -114,6 +161,49 @@ def websocket_get_nodes(hass, connection, msg):
)
@websocket_api.websocket_command(
{
vol.Required(TYPE): "ozw/set_usercode",
vol.Required(NODE_ID): vol.Coerce(int),
vol.Optional(OZW_INSTANCE, default=1): vol.Coerce(int),
vol.Required(ATTR_CODE_SLOT): vol.Coerce(int),
vol.Required(ATTR_USERCODE): cv.string,
}
)
def websocket_set_usercode(hass, connection, msg):
"""Set a usercode to a node code slot."""
_call_util_function(
hass, connection, msg, False, set_usercode, msg[ATTR_CODE_SLOT], ATTR_USERCODE
)
@websocket_api.websocket_command(
{
vol.Required(TYPE): "ozw/clear_usercode",
vol.Required(NODE_ID): vol.Coerce(int),
vol.Optional(OZW_INSTANCE, default=1): vol.Coerce(int),
vol.Required(ATTR_CODE_SLOT): vol.Coerce(int),
}
)
def websocket_clear_usercode(hass, connection, msg):
"""Clear a node code slot."""
_call_util_function(
hass, connection, msg, False, clear_usercode, msg[ATTR_CODE_SLOT]
)
@websocket_api.websocket_command(
{
vol.Required(TYPE): "ozw/get_code_slots",
vol.Required(NODE_ID): vol.Coerce(int),
vol.Optional(OZW_INSTANCE, default=1): vol.Coerce(int),
}
)
def websocket_get_code_slots(hass, connection, msg):
"""Get status of node's code slots."""
_call_util_function(hass, connection, msg, True, get_code_slots)
@websocket_api.websocket_command(
{
vol.Required(TYPE): "ozw/get_config_parameters",
@ -123,22 +213,7 @@ def websocket_get_nodes(hass, connection, msg):
)
def websocket_get_config_parameters(hass, connection, msg):
"""Get a list of configuration parameters for an OZW node instance."""
try:
node = get_node_from_manager(
hass.data[DOMAIN][MANAGER], msg[OZW_INSTANCE], msg[NODE_ID]
)
except NotFoundError as err:
connection.send_error(
msg[ID],
websocket_api.const.ERR_NOT_FOUND,
err.args[0],
)
return
connection.send_result(
msg[ID],
get_config_parameters(node),
)
_call_util_function(hass, connection, msg, True, get_config_parameters)
@websocket_api.websocket_command(
@ -169,40 +244,9 @@ def websocket_get_config_parameters(hass, connection, msg):
)
def websocket_set_config_parameter(hass, connection, msg):
"""Set a config parameter to a node."""
try:
node = get_node_from_manager(
hass.data[DOMAIN][MANAGER], msg[OZW_INSTANCE], msg[NODE_ID]
)
except NotFoundError as err:
connection.send_error(
msg[ID],
websocket_api.const.ERR_NOT_FOUND,
err.args[0],
)
return
try:
set_config_parameter(
node,
msg[PARAMETER],
msg[VALUE],
)
except NotFoundError as err:
connection.send_error(
msg[ID],
websocket_api.const.ERR_NOT_FOUND,
err.args[0],
)
return
except NotSupportedError as err:
connection.send_error(
msg[ID],
websocket_api.const.ERR_NOT_SUPPORTED,
err.args[0],
)
return
connection.send_result(msg[ID])
_call_util_function(
hass, connection, msg, False, set_config_parameter, msg[PARAMETER], msg[VALUE]
)
@websocket_api.websocket_command(

5
tests/components/ozw/test_lock.py
View file

@ -80,7 +80,4 @@ async def test_lock(hass, lock_data, sent_messages, lock_msg, caplog):
blocking=True,
)
assert len(sent_messages) == 5
assert (
"Invalid code provided: (123) user code must be at least 4 digits"
in caplog.text
)
assert "User code must be at least 4 digits" in caplog.text

41
tests/components/ozw/test_websocket_api.py
View file

@ -1,5 +1,6 @@
"""Test OpenZWave Websocket API."""
from openzwavemqtt.const import (
ATTR_CODE_SLOT,
ATTR_LABEL,
ATTR_OPTIONS,
ATTR_POSITION,
@ -8,6 +9,7 @@ from openzwavemqtt.const import (
)
from homeassistant.components.ozw.const import ATTR_CONFIG_PARAMETER
from homeassistant.components.ozw.lock import ATTR_USERCODE
from homeassistant.components.ozw.websocket_api import (
ATTR_IS_AWAKE,
ATTR_IS_BEAMING,
@ -277,6 +279,45 @@ async def test_websocket_api(hass, generic_data, hass_ws_client):
assert result["code"] == ERR_NOT_FOUND
async def test_ws_locks(hass, lock_data, hass_ws_client):
"""Test lock websocket apis."""
await setup_ozw(hass, fixture=lock_data)
client = await hass_ws_client(hass)
await client.send_json(
{
ID: 1,
TYPE: "ozw/get_code_slots",
NODE_ID: 10,
}
)
msg = await client.receive_json()
assert msg["success"]
await client.send_json(
{
ID: 2,
TYPE: "ozw/set_usercode",
NODE_ID: 10,
ATTR_CODE_SLOT: 1,
ATTR_USERCODE: "1234",
}
)
msg = await client.receive_json()
assert msg["success"]
await client.send_json(
{
ID: 3,
TYPE: "ozw/clear_usercode",
NODE_ID: 10,
ATTR_CODE_SLOT: 1,
}
)
msg = await client.receive_json()
assert msg["success"]
async def test_refresh_node(hass, generic_data, sent_messages, hass_ws_client):
"""Test the ozw refresh node api."""
receive_message = await setup_ozw(hass, fixture=generic_data)