Add ness alarm control panel using nessclient (#18463)

* Add ness alarm control panel using nessclient

* indenting

* .

* Remove availability functionality, will improve and add back in another PR

* Use call_count

* lint

* lint

* Review changes

* Lint

* Bump nessclient to 0.9.8

* Bump nessclient to 0.9.9

* Remove from .coveragerc
This commit is contained in:
Nick Whyte 2019-01-02 02:08:13 +11:00 committed by Martin Hjelmare
parent 61d5b3028d
commit 5aa2bd81cf
6 changed files with 566 additions and 0 deletions

View file

@ -213,6 +213,10 @@ homeassistant/components/melissa.py @kennedyshead
homeassistant/components/*/melissa.py @kennedyshead
homeassistant/components/*/mystrom.py @fabaff
# N
homeassistant/components/ness_alarm.py @nickw444
homeassistant/components/*/ness_alarm.py @nickw444
# O
homeassistant/components/openuv/* @bachya
homeassistant/components/*/openuv.py @bachya

View file

@ -0,0 +1,107 @@
"""
Support for Ness D8X/D16X alarm panel.
For more details about this platform, please refer to the documentation at
https://home-assistant.io/components/alarm_control_panel.ness_alarm/
"""
import logging
import homeassistant.components.alarm_control_panel as alarm
from homeassistant.components.ness_alarm import (
DATA_NESS, SIGNAL_ARMING_STATE_CHANGED)
from homeassistant.const import (
STATE_ALARM_ARMED_AWAY, STATE_ALARM_ARMING,
STATE_ALARM_TRIGGERED, STATE_ALARM_PENDING, STATE_ALARM_DISARMED)
from homeassistant.core import callback
from homeassistant.helpers.dispatcher import async_dispatcher_connect
_LOGGER = logging.getLogger(__name__)
DEPENDENCIES = ['ness_alarm']
async def async_setup_platform(hass, config, async_add_entities,
discovery_info=None):
"""Set up the Ness Alarm alarm control panel devices."""
if discovery_info is None:
return
device = NessAlarmPanel(hass.data[DATA_NESS], 'Alarm Panel')
async_add_entities([device])
class NessAlarmPanel(alarm.AlarmControlPanel):
"""Representation of a Ness alarm panel."""
def __init__(self, client, name):
"""Initialize the alarm panel."""
self._client = client
self._name = name
self._state = None
async def async_added_to_hass(self):
"""Register callbacks."""
async_dispatcher_connect(
self.hass, SIGNAL_ARMING_STATE_CHANGED,
self._handle_arming_state_change)
@property
def name(self):
"""Return the name of the device."""
return self._name
@property
def should_poll(self):
"""Return the polling state."""
return False
@property
def code_format(self):
"""Return the regex for code format or None if no code is required."""
return 'Number'
@property
def state(self):
"""Return the state of the device."""
return self._state
async def async_alarm_disarm(self, code=None):
"""Send disarm command."""
await self._client.disarm(code)
async def async_alarm_arm_away(self, code=None):
"""Send arm away command."""
await self._client.arm_away(code)
async def async_alarm_arm_home(self, code=None):
"""Send arm home command."""
await self._client.arm_home(code)
async def async_alarm_trigger(self, code=None):
"""Send trigger/panic command."""
await self._client.panic(code)
@callback
def _handle_arming_state_change(self, arming_state):
"""Handle arming state update."""
from nessclient import ArmingState
if arming_state == ArmingState.UNKNOWN:
self._state = None
elif arming_state == ArmingState.DISARMED:
self._state = STATE_ALARM_DISARMED
elif arming_state == ArmingState.ARMING:
self._state = STATE_ALARM_ARMING
elif arming_state == ArmingState.EXIT_DELAY:
self._state = STATE_ALARM_ARMING
elif arming_state == ArmingState.ARMED:
self._state = STATE_ALARM_ARMED_AWAY
elif arming_state == ArmingState.ENTRY_DELAY:
self._state = STATE_ALARM_PENDING
elif arming_state == ArmingState.TRIGGERED:
self._state = STATE_ALARM_TRIGGERED
else:
_LOGGER.warning("Unhandled arming state: %s", arming_state)
self.async_schedule_update_ha_state()

View file

@ -0,0 +1,81 @@
"""
Support for Ness D8X/D16X zone states - represented as binary sensors.
For more details about this platform, please refer to the documentation at
https://home-assistant.io/components/binary_sensor.ness_alarm/
"""
import logging
from homeassistant.components.binary_sensor import BinarySensorDevice
from homeassistant.components.ness_alarm import (
CONF_ZONES, CONF_ZONE_TYPE, CONF_ZONE_NAME, CONF_ZONE_ID,
SIGNAL_ZONE_CHANGED, ZoneChangedData)
from homeassistant.core import callback
from homeassistant.helpers.dispatcher import async_dispatcher_connect
DEPENDENCIES = ['ness_alarm']
_LOGGER = logging.getLogger(__name__)
async def async_setup_platform(hass, config, async_add_entities,
discovery_info=None):
"""Set up the Ness Alarm binary sensor devices."""
if not discovery_info:
return
configured_zones = discovery_info[CONF_ZONES]
devices = []
for zone_config in configured_zones:
zone_type = zone_config[CONF_ZONE_TYPE]
zone_name = zone_config[CONF_ZONE_NAME]
zone_id = zone_config[CONF_ZONE_ID]
device = NessZoneBinarySensor(zone_id=zone_id, name=zone_name,
zone_type=zone_type)
devices.append(device)
async_add_entities(devices)
class NessZoneBinarySensor(BinarySensorDevice):
"""Representation of an Ness alarm zone as a binary sensor."""
def __init__(self, zone_id, name, zone_type):
"""Initialize the binary_sensor."""
self._zone_id = zone_id
self._name = name
self._type = zone_type
self._state = 0
async def async_added_to_hass(self):
"""Register callbacks."""
async_dispatcher_connect(
self.hass, SIGNAL_ZONE_CHANGED, self._handle_zone_change)
@property
def name(self):
"""Return the name of the entity."""
return self._name
@property
def should_poll(self):
"""No polling needed."""
return False
@property
def is_on(self):
"""Return true if sensor is on."""
return self._state == 1
@property
def device_class(self):
"""Return the class of this sensor, from DEVICE_CLASSES."""
return self._type
@callback
def _handle_zone_change(self, data: ZoneChangedData):
"""Handle zone state update."""
if self._zone_id == data.zone_id:
self._state = data.state
self.async_schedule_update_ha_state()

View file

@ -0,0 +1,121 @@
"""
Support for Ness D8X/D16X devices.
For more details about this component, please refer to the documentation at
https://home-assistant.io/components/ness_alarm/
"""
import logging
from collections import namedtuple
import voluptuous as vol
from homeassistant.components.binary_sensor import DEVICE_CLASSES
from homeassistant.const import EVENT_HOMEASSISTANT_STOP
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.discovery import async_load_platform
from homeassistant.helpers.dispatcher import async_dispatcher_send
REQUIREMENTS = ['nessclient==0.9.9']
_LOGGER = logging.getLogger(__name__)
DOMAIN = 'ness_alarm'
DATA_NESS = 'ness_alarm'
CONF_DEVICE_HOST = 'host'
CONF_DEVICE_PORT = 'port'
CONF_ZONES = 'zones'
CONF_ZONE_NAME = 'name'
CONF_ZONE_TYPE = 'type'
CONF_ZONE_ID = 'id'
ATTR_CODE = 'code'
ATTR_OUTPUT_ID = 'output_id'
ATTR_STATE = 'state'
DEFAULT_ZONES = []
SIGNAL_ZONE_CHANGED = 'ness_alarm.zone_changed'
SIGNAL_ARMING_STATE_CHANGED = 'ness_alarm.arming_state_changed'
ZoneChangedData = namedtuple('ZoneChangedData', ['zone_id', 'state'])
DEFAULT_ZONE_TYPE = 'motion'
ZONE_SCHEMA = vol.Schema({
vol.Required(CONF_ZONE_NAME): cv.string,
vol.Required(CONF_ZONE_ID): cv.positive_int,
vol.Optional(CONF_ZONE_TYPE, default=DEFAULT_ZONE_TYPE):
vol.In(DEVICE_CLASSES)})
CONFIG_SCHEMA = vol.Schema({
DOMAIN: vol.Schema({
vol.Required(CONF_DEVICE_HOST): cv.string,
vol.Required(CONF_DEVICE_PORT): cv.port,
vol.Optional(CONF_ZONES, default=DEFAULT_ZONES):
vol.All(cv.ensure_list, [ZONE_SCHEMA]),
}),
}, extra=vol.ALLOW_EXTRA)
SERVICE_PANIC = 'panic'
SERVICE_AUX = 'aux'
SERVICE_SCHEMA_PANIC = vol.Schema({
vol.Required(ATTR_CODE): cv.string,
})
SERVICE_SCHEMA_AUX = vol.Schema({
vol.Required(ATTR_OUTPUT_ID): cv.positive_int,
vol.Optional(ATTR_STATE, default=True): cv.boolean,
})
async def async_setup(hass, config):
"""Set up the Ness Alarm platform."""
from nessclient import Client, ArmingState
conf = config[DOMAIN]
zones = conf[CONF_ZONES]
host = conf[CONF_DEVICE_HOST]
port = conf[CONF_DEVICE_PORT]
client = Client(host=host, port=port, loop=hass.loop)
hass.data[DATA_NESS] = client
async def _close(event):
await client.close()
hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, _close)
hass.async_create_task(
async_load_platform(hass, 'binary_sensor', DOMAIN, {CONF_ZONES: zones},
config))
hass.async_create_task(
async_load_platform(hass, 'alarm_control_panel', DOMAIN, {}, config))
def on_zone_change(zone_id: int, state: bool):
"""Receives and propagates zone state updates."""
async_dispatcher_send(hass, SIGNAL_ZONE_CHANGED, ZoneChangedData(
zone_id=zone_id,
state=state,
))
def on_state_change(arming_state: ArmingState):
"""Receives and propagates arming state updates."""
async_dispatcher_send(hass, SIGNAL_ARMING_STATE_CHANGED, arming_state)
client.on_zone_change(on_zone_change)
client.on_state_change(on_state_change)
# Force update for current arming status and current zone states
hass.loop.create_task(client.keepalive())
hass.loop.create_task(client.update())
async def handle_panic(call):
await client.panic(call.data[ATTR_CODE])
async def handle_aux(call):
await client.aux(call.data[ATTR_OUTPUT_ID], call.data[ATTR_STATE])
hass.services.async_register(DOMAIN, SERVICE_PANIC, handle_panic,
schema=SERVICE_SCHEMA_PANIC)
hass.services.async_register(DOMAIN, SERVICE_AUX, handle_aux,
schema=SERVICE_SCHEMA_AUX)
return True

View file

@ -698,6 +698,9 @@ nanoleaf==0.4.1
# homeassistant.components.device_tracker.keenetic_ndms2
ndms2_client==0.0.6
# homeassistant.components.ness_alarm
nessclient==0.9.9
# homeassistant.components.sensor.netdata
netdata==0.1.2

View file

@ -0,0 +1,250 @@
"""Tests for the ness_alarm component."""
from enum import Enum
import pytest
from asynctest import patch, MagicMock
from homeassistant.components import alarm_control_panel
from homeassistant.components.ness_alarm import (
DOMAIN, CONF_DEVICE_PORT, CONF_DEVICE_HOST, CONF_ZONE_NAME, CONF_ZONES,
CONF_ZONE_ID, SERVICE_AUX, SERVICE_PANIC,
ATTR_CODE, ATTR_OUTPUT_ID)
from homeassistant.const import (
STATE_ALARM_ARMING, SERVICE_ALARM_DISARM, ATTR_ENTITY_ID,
SERVICE_ALARM_ARM_AWAY, SERVICE_ALARM_ARM_HOME, SERVICE_ALARM_TRIGGER,
STATE_ALARM_DISARMED, STATE_ALARM_ARMED_AWAY, STATE_ALARM_PENDING,
STATE_ALARM_TRIGGERED, STATE_UNKNOWN)
from homeassistant.setup import async_setup_component
from tests.common import MockDependency
VALID_CONFIG = {
DOMAIN: {
CONF_DEVICE_HOST: 'alarm.local',
CONF_DEVICE_PORT: 1234,
CONF_ZONES: [
{
CONF_ZONE_NAME: 'Zone 1',
CONF_ZONE_ID: 1,
},
{
CONF_ZONE_NAME: 'Zone 2',
CONF_ZONE_ID: 2,
}
]
}
}
async def test_setup_platform(hass, mock_nessclient):
"""Test platform setup."""
await async_setup_component(hass, DOMAIN, VALID_CONFIG)
assert hass.services.has_service(DOMAIN, 'panic')
assert hass.services.has_service(DOMAIN, 'aux')
await hass.async_block_till_done()
assert hass.states.get('alarm_control_panel.alarm_panel') is not None
assert hass.states.get('binary_sensor.zone_1') is not None
assert hass.states.get('binary_sensor.zone_2') is not None
assert mock_nessclient.keepalive.call_count == 1
assert mock_nessclient.update.call_count == 1
async def test_panic_service(hass, mock_nessclient):
"""Test calling panic service."""
await async_setup_component(hass, DOMAIN, VALID_CONFIG)
await hass.services.async_call(
DOMAIN, SERVICE_PANIC, blocking=True, service_data={
ATTR_CODE: '1234'
})
mock_nessclient.panic.assert_awaited_once_with('1234')
async def test_aux_service(hass, mock_nessclient):
"""Test calling aux service."""
await async_setup_component(hass, DOMAIN, VALID_CONFIG)
await hass.services.async_call(
DOMAIN, SERVICE_AUX, blocking=True, service_data={
ATTR_OUTPUT_ID: 1
})
mock_nessclient.aux.assert_awaited_once_with(1, True)
async def test_dispatch_state_change(hass, mock_nessclient):
"""Test calling aux service."""
await async_setup_component(hass, DOMAIN, VALID_CONFIG)
await hass.async_block_till_done()
on_state_change = mock_nessclient.on_state_change.call_args[0][0]
on_state_change(MockArmingState.ARMING)
await hass.async_block_till_done()
assert hass.states.is_state('alarm_control_panel.alarm_panel',
STATE_ALARM_ARMING)
async def test_alarm_disarm(hass, mock_nessclient):
"""Test disarm."""
await async_setup_component(hass, DOMAIN, VALID_CONFIG)
await hass.async_block_till_done()
await hass.services.async_call(
alarm_control_panel.DOMAIN, SERVICE_ALARM_DISARM, blocking=True,
service_data={
ATTR_ENTITY_ID: 'alarm_control_panel.alarm_panel',
ATTR_CODE: '1234'
})
mock_nessclient.disarm.assert_called_once_with('1234')
async def test_alarm_arm_away(hass, mock_nessclient):
"""Test disarm."""
await async_setup_component(hass, DOMAIN, VALID_CONFIG)
await hass.async_block_till_done()
await hass.services.async_call(
alarm_control_panel.DOMAIN, SERVICE_ALARM_ARM_AWAY, blocking=True,
service_data={
ATTR_ENTITY_ID: 'alarm_control_panel.alarm_panel',
ATTR_CODE: '1234'
})
mock_nessclient.arm_away.assert_called_once_with('1234')
async def test_alarm_arm_home(hass, mock_nessclient):
"""Test disarm."""
await async_setup_component(hass, DOMAIN, VALID_CONFIG)
await hass.async_block_till_done()
await hass.services.async_call(
alarm_control_panel.DOMAIN, SERVICE_ALARM_ARM_HOME, blocking=True,
service_data={
ATTR_ENTITY_ID: 'alarm_control_panel.alarm_panel',
ATTR_CODE: '1234'
})
mock_nessclient.arm_home.assert_called_once_with('1234')
async def test_alarm_trigger(hass, mock_nessclient):
"""Test disarm."""
await async_setup_component(hass, DOMAIN, VALID_CONFIG)
await hass.async_block_till_done()
await hass.services.async_call(
alarm_control_panel.DOMAIN, SERVICE_ALARM_TRIGGER, blocking=True,
service_data={
ATTR_ENTITY_ID: 'alarm_control_panel.alarm_panel',
ATTR_CODE: '1234'
})
mock_nessclient.panic.assert_called_once_with('1234')
async def test_dispatch_zone_change(hass, mock_nessclient):
"""Test zone change events dispatch a signal to subscribers."""
await async_setup_component(hass, DOMAIN, VALID_CONFIG)
await hass.async_block_till_done()
on_zone_change = mock_nessclient.on_zone_change.call_args[0][0]
on_zone_change(1, True)
await hass.async_block_till_done()
assert hass.states.is_state('binary_sensor.zone_1', 'on')
assert hass.states.is_state('binary_sensor.zone_2', 'off')
async def test_arming_state_change(hass, mock_nessclient):
"""Test arming state change handing."""
states = [
(MockArmingState.UNKNOWN, STATE_UNKNOWN),
(MockArmingState.DISARMED, STATE_ALARM_DISARMED),
(MockArmingState.ARMING, STATE_ALARM_ARMING),
(MockArmingState.EXIT_DELAY, STATE_ALARM_ARMING),
(MockArmingState.ARMED, STATE_ALARM_ARMED_AWAY),
(MockArmingState.ENTRY_DELAY, STATE_ALARM_PENDING),
(MockArmingState.TRIGGERED, STATE_ALARM_TRIGGERED),
]
await async_setup_component(hass, DOMAIN, VALID_CONFIG)
await hass.async_block_till_done()
assert hass.states.is_state('alarm_control_panel.alarm_panel',
STATE_UNKNOWN)
on_state_change = mock_nessclient.on_state_change.call_args[0][0]
for arming_state, expected_state in states:
on_state_change(arming_state)
await hass.async_block_till_done()
assert hass.states.is_state('alarm_control_panel.alarm_panel',
expected_state)
class MockArmingState(Enum):
"""Mock nessclient.ArmingState enum."""
UNKNOWN = 'UNKNOWN'
DISARMED = 'DISARMED'
ARMING = 'ARMING'
EXIT_DELAY = 'EXIT_DELAY'
ARMED = 'ARMED'
ENTRY_DELAY = 'ENTRY_DELAY'
TRIGGERED = 'TRIGGERED'
class MockClient:
"""Mock nessclient.Client stub."""
async def panic(self, code):
"""Handle panic."""
pass
async def disarm(self, code):
"""Handle disarm."""
pass
async def arm_away(self, code):
"""Handle arm_away."""
pass
async def arm_home(self, code):
"""Handle arm_home."""
pass
async def aux(self, output_id, state):
"""Handle auxiliary control."""
pass
async def keepalive(self):
"""Handle keepalive."""
pass
async def update(self):
"""Handle update."""
pass
def on_zone_change(self):
"""Handle on_zone_change."""
pass
def on_state_change(self):
"""Handle on_state_change."""
pass
async def close(self):
"""Handle close."""
pass
@pytest.fixture
def mock_nessclient():
"""Mock the nessclient Client constructor.
Replaces nessclient.Client with a Mock which always returns the same
MagicMock() instance.
"""
_mock_instance = MagicMock(MockClient())
_mock_factory = MagicMock()
_mock_factory.return_value = _mock_instance
with MockDependency('nessclient'), \
patch('nessclient.Client', new=_mock_factory, create=True), \
patch('nessclient.ArmingState', new=MockArmingState):
yield _mock_instance