Add support for Vanderbilt SPC alarm panels and attached sensors (#7663)
* Add support for Vanderbilt SPC alarm panels. * Arm/disarm + read state * Autodiscover and add motion sensors * Fix code formatting. * Use asyncio.async for Python < 3.4.4. * Fix for moved aiohttp exceptions. * Add docstrings. * Fix tests and add docstrings.pull/6616/merge
Martin Berg
Paulus Schoutsen
parent
549133a062
commit
b90964faad
|
|
@ -0,0 +1,96 @@
|
|||
"""
|
||||
Support for Vanderbilt (formerly Siemens) SPC alarm systems.
|
||||
|
||||
For more details about this platform, please refer to the documentation at
|
||||
https://home-assistant.io/components/alarm_control_panel.spc/
|
||||
"""
|
||||
import asyncio
|
||||
import logging
|
||||
|
||||
import homeassistant.components.alarm_control_panel as alarm
|
||||
from homeassistant.components.spc import (
|
||||
SpcWebGateway, ATTR_DISCOVER_AREAS, DATA_API, DATA_REGISTRY)
|
||||
from homeassistant.const import (
|
||||
STATE_ALARM_ARMED_AWAY, STATE_ALARM_ARMED_HOME, STATE_ALARM_DISARMED,
|
||||
STATE_UNKNOWN)
|
||||
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
SPC_AREA_MODE_TO_STATE = {'0': STATE_ALARM_DISARMED,
|
||||
'1': STATE_ALARM_ARMED_HOME,
|
||||
'3': STATE_ALARM_ARMED_AWAY}
|
||||
|
||||
|
||||
def _get_alarm_state(spc_mode):
|
||||
return SPC_AREA_MODE_TO_STATE.get(spc_mode, STATE_UNKNOWN)
|
||||
|
||||
|
||||
@asyncio.coroutine
|
||||
def async_setup_platform(hass, config, async_add_entities,
|
||||
discovery_info=None):
|
||||
"""Set up the SPC alarm control panel platform."""
|
||||
if (discovery_info is None or
|
||||
discovery_info[ATTR_DISCOVER_AREAS] is None):
|
||||
return
|
||||
|
||||
entities = [SpcAlarm(hass=hass,
|
||||
area_id=area['id'],
|
||||
name=area['name'],
|
||||
state=_get_alarm_state(area['mode']))
|
||||
for area in discovery_info[ATTR_DISCOVER_AREAS]]
|
||||
|
||||
async_add_entities(entities)
|
||||
|
||||
|
||||
class SpcAlarm(alarm.AlarmControlPanel):
|
||||
"""Represents the SPC alarm panel."""
|
||||
|
||||
def __init__(self, hass, area_id, name, state):
|
||||
"""Initialize the SPC alarm panel."""
|
||||
self._hass = hass
|
||||
self._area_id = area_id
|
||||
self._name = name
|
||||
self._state = state
|
||||
self._api = hass.data[DATA_API]
|
||||
|
||||
hass.data[DATA_REGISTRY].register_alarm_device(area_id, self)
|
||||
|
||||
@asyncio.coroutine
|
||||
def async_update_from_spc(self, state):
|
||||
"""Update the alarm panel with a new state."""
|
||||
self._state = state
|
||||
yield from self.async_update_ha_state()
|
||||
|
||||
@property
|
||||
def should_poll(self):
|
||||
"""No polling needed."""
|
||||
return False
|
||||
|
||||
@property
|
||||
def name(self):
|
||||
"""Return the name of the device."""
|
||||
return self._name
|
||||
|
||||
@property
|
||||
def state(self):
|
||||
"""Return the state of the device."""
|
||||
return self._state
|
||||
|
||||
@asyncio.coroutine
|
||||
def async_alarm_disarm(self, code=None):
|
||||
"""Send disarm command."""
|
||||
yield from self._api.send_area_command(
|
||||
self._area_id, SpcWebGateway.AREA_COMMAND_UNSET)
|
||||
|
||||
@asyncio.coroutine
|
||||
def async_alarm_arm_home(self, code=None):
|
||||
"""Send arm home command."""
|
||||
yield from self._api.send_area_command(
|
||||
self._area_id, SpcWebGateway.AREA_COMMAND_PART_SET)
|
||||
|
||||
@asyncio.coroutine
|
||||
def async_alarm_arm_away(self, code=None):
|
||||
"""Send arm away command."""
|
||||
yield from self._api.send_area_command(
|
||||
self._area_id, SpcWebGateway.AREA_COMMAND_SET)
|
||||
|
|
@ -0,0 +1,99 @@
|
|||
"""
|
||||
Support for Vanderbilt (formerly Siemens) SPC alarm systems.
|
||||
|
||||
For more details about this platform, please refer to the documentation at
|
||||
https://home-assistant.io/components/binary_sensor.spc/
|
||||
"""
|
||||
import logging
|
||||
import asyncio
|
||||
|
||||
from homeassistant.components.spc import (
|
||||
ATTR_DISCOVER_DEVICES, DATA_REGISTRY)
|
||||
from homeassistant.components.binary_sensor import BinarySensorDevice
|
||||
from homeassistant.const import (STATE_UNAVAILABLE, STATE_ON, STATE_OFF)
|
||||
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
SPC_TYPE_TO_DEVICE_CLASS = {'0': 'motion',
|
||||
'1': 'opening',
|
||||
'3': 'smoke'}
|
||||
|
||||
|
||||
SPC_INPUT_TO_SENSOR_STATE = {'0': STATE_OFF,
|
||||
'1': STATE_ON}
|
||||
|
||||
|
||||
def _get_device_class(spc_type):
|
||||
return SPC_TYPE_TO_DEVICE_CLASS.get(spc_type, None)
|
||||
|
||||
|
||||
def _get_sensor_state(spc_input):
|
||||
return SPC_INPUT_TO_SENSOR_STATE.get(spc_input, STATE_UNAVAILABLE)
|
||||
|
||||
|
||||
def _create_sensor(hass, zone):
|
||||
return SpcBinarySensor(zone_id=zone['id'],
|
||||
name=zone['zone_name'],
|
||||
state=_get_sensor_state(zone['input']),
|
||||
device_class=_get_device_class(zone['type']),
|
||||
spc_registry=hass.data[DATA_REGISTRY])
|
||||
|
||||
|
||||
@asyncio.coroutine
|
||||
def async_setup_platform(hass, config, async_add_entities,
|
||||
discovery_info=None):
|
||||
"""Initialize the platform."""
|
||||
if (discovery_info is None or
|
||||
discovery_info[ATTR_DISCOVER_DEVICES] is None):
|
||||
return
|
||||
|
||||
async_add_entities(
|
||||
_create_sensor(hass, zone)
|
||||
for zone in discovery_info[ATTR_DISCOVER_DEVICES]
|
||||
if _get_device_class(zone['type']))
|
||||
|
||||
|
||||
class SpcBinarySensor(BinarySensorDevice):
|
||||
"""Represents a sensor based on an SPC zone."""
|
||||
|
||||
def __init__(self, zone_id, name, state, device_class, spc_registry):
|
||||
"""Initialize the sensor device."""
|
||||
self._zone_id = zone_id
|
||||
self._name = name
|
||||
self._state = state
|
||||
self._device_class = device_class
|
||||
|
||||
spc_registry.register_sensor_device(zone_id, self)
|
||||
|
||||
@asyncio.coroutine
|
||||
def async_update_from_spc(self, state):
|
||||
"""Update the state of the device."""
|
||||
self._state = state
|
||||
yield from self.async_update_ha_state()
|
||||
|
||||
@property
|
||||
def name(self):
|
||||
"""The name of the device."""
|
||||
return self._name
|
||||
|
||||
@property
|
||||
def is_on(self):
|
||||
"""Whether the device is switched on."""
|
||||
return self._state == STATE_ON
|
||||
|
||||
@property
|
||||
def hidden(self) -> bool:
|
||||
"""Whether the device is hidden by default."""
|
||||
# these type of sensors are probably mainly used for automations
|
||||
return True
|
||||
|
||||
@property
|
||||
def should_poll(self):
|
||||
"""No polling needed."""
|
||||
return False
|
||||
|
||||
@property
|
||||
def device_class(self):
|
||||
"""The device class."""
|
||||
return self._device_class
|
||||
|
|
@ -0,0 +1,279 @@
|
|||
"""
|
||||
Support for Vanderbilt (formerly Siemens) SPC alarm systems.
|
||||
|
||||
For more details about this platform, please refer to the documentation at
|
||||
https://home-assistant.io/components/spc/
|
||||
"""
|
||||
import logging
|
||||
import asyncio
|
||||
import json
|
||||
from urllib.parse import urljoin
|
||||
|
||||
import aiohttp
|
||||
import async_timeout
|
||||
import voluptuous as vol
|
||||
|
||||
import homeassistant.helpers.config_validation as cv
|
||||
from homeassistant.helpers import discovery
|
||||
from homeassistant.const import (
|
||||
STATE_UNKNOWN, STATE_ON, STATE_OFF, STATE_ALARM_ARMED_AWAY,
|
||||
STATE_ALARM_ARMED_HOME, STATE_ALARM_DISARMED, STATE_ALARM_TRIGGERED,
|
||||
STATE_UNAVAILABLE)
|
||||
|
||||
DOMAIN = 'spc'
|
||||
REQUIREMENTS = ['websockets==3.2']
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
ATTR_DISCOVER_DEVICES = 'devices'
|
||||
ATTR_DISCOVER_AREAS = 'areas'
|
||||
|
||||
CONF_WS_URL = 'ws_url'
|
||||
CONF_API_URL = 'api_url'
|
||||
|
||||
DATA_REGISTRY = 'spc_registry'
|
||||
DATA_API = 'spc_api'
|
||||
|
||||
CONFIG_SCHEMA = vol.Schema({
|
||||
DOMAIN: vol.Schema({
|
||||
vol.Required(CONF_WS_URL): cv.string,
|
||||
vol.Required(CONF_API_URL): cv.string
|
||||
}),
|
||||
}, extra=vol.ALLOW_EXTRA)
|
||||
|
||||
|
||||
@asyncio.coroutine
|
||||
def async_setup(hass, config):
|
||||
"""Setup the SPC platform."""
|
||||
hass.data[DATA_REGISTRY] = SpcRegistry()
|
||||
|
||||
api = SpcWebGateway(hass,
|
||||
config[DOMAIN].get(CONF_API_URL),
|
||||
config[DOMAIN].get(CONF_WS_URL))
|
||||
|
||||
hass.data[DATA_API] = api
|
||||
|
||||
# add sensor devices for each zone (typically motion/fire/door sensors)
|
||||
zones = yield from api.get_zones()
|
||||
if zones:
|
||||
hass.async_add_job(discovery.async_load_platform(
|
||||
hass, 'binary_sensor', DOMAIN,
|
||||
{ATTR_DISCOVER_DEVICES: zones}, config))
|
||||
|
||||
# create a separate alarm panel for each area
|
||||
areas = yield from api.get_areas()
|
||||
if areas:
|
||||
hass.async_add_job(discovery.async_load_platform(
|
||||
hass, 'alarm_control_panel', DOMAIN,
|
||||
{ATTR_DISCOVER_AREAS: areas}, config))
|
||||
|
||||
# start listening for incoming events over websocket
|
||||
api.start_listener(_async_process_message, hass.data[DATA_REGISTRY])
|
||||
|
||||
return True
|
||||
|
||||
|
||||
@asyncio.coroutine
|
||||
def _async_process_message(sia_message, spc_registry):
|
||||
spc_id = sia_message['sia_address']
|
||||
sia_code = sia_message['sia_code']
|
||||
|
||||
# BA - Burglary Alarm
|
||||
# CG - Close Area
|
||||
# NL - Perimeter Armed
|
||||
# OG - Open Area
|
||||
# ZO - Zone Open
|
||||
# ZC - Zone Close
|
||||
# ZX - Zone Short
|
||||
# ZD - Zone Disconnected
|
||||
|
||||
if sia_code in ('BA', 'CG', 'NL', 'OG', 'OQ'):
|
||||
# change in area status, notify alarm panel device
|
||||
device = spc_registry.get_alarm_device(spc_id)
|
||||
else:
|
||||
# change in zone status, notify sensor device
|
||||
device = spc_registry.get_sensor_device(spc_id)
|
||||
|
||||
sia_code_to_state_map = {'BA': STATE_ALARM_TRIGGERED,
|
||||
'CG': STATE_ALARM_ARMED_AWAY,
|
||||
'NL': STATE_ALARM_ARMED_HOME,
|
||||
'OG': STATE_ALARM_DISARMED,
|
||||
'OQ': STATE_ALARM_DISARMED,
|
||||
'ZO': STATE_ON,
|
||||
'ZC': STATE_OFF,
|
||||
'ZX': STATE_UNKNOWN,
|
||||
'ZD': STATE_UNAVAILABLE}
|
||||
|
||||
new_state = sia_code_to_state_map.get(sia_code, None)
|
||||
|
||||
if new_state and not device:
|
||||
_LOGGER.warning("No device mapping found for SPC area/zone id %s.",
|
||||
spc_id)
|
||||
elif new_state:
|
||||
yield from device.async_update_from_spc(new_state)
|
||||
|
||||
|
||||
class SpcRegistry:
|
||||
"""Maintains mappings between SPC zones/areas and HA entities."""
|
||||
|
||||
def __init__(self):
|
||||
"""Initialize the registry."""
|
||||
self._zone_id_to_sensor_map = {}
|
||||
self._area_id_to_alarm_map = {}
|
||||
|
||||
def register_sensor_device(self, zone_id, device):
|
||||
"""Add a sensor device to the registry."""
|
||||
self._zone_id_to_sensor_map[zone_id] = device
|
||||
|
||||
def get_sensor_device(self, zone_id):
|
||||
"""Retrieve a sensor device for a specific zone."""
|
||||
return self._zone_id_to_sensor_map.get(zone_id, None)
|
||||
|
||||
def register_alarm_device(self, area_id, device):
|
||||
"""Add an alarm device to the registry."""
|
||||
self._area_id_to_alarm_map[area_id] = device
|
||||
|
||||
def get_alarm_device(self, area_id):
|
||||
"""Retrieve an alarm device for a specific area."""
|
||||
return self._area_id_to_alarm_map.get(area_id, None)
|
||||
|
||||
|
||||
@asyncio.coroutine
|
||||
def _ws_process_message(message, async_callback, *args):
|
||||
if message.get('status', '') != 'success':
|
||||
_LOGGER.warning("Unsuccessful websocket message "
|
||||
"delivered, ignoring: %s", message)
|
||||
try:
|
||||
yield from async_callback(message['data']['sia'], *args)
|
||||
except: # pylint: disable=bare-except
|
||||
_LOGGER.exception("Exception in callback, ignoring.")
|
||||
|
||||
|
||||
class SpcWebGateway:
|
||||
"""Simple binding for the Lundix SPC Web Gateway REST API."""
|
||||
|
||||
AREA_COMMAND_SET = 'set'
|
||||
AREA_COMMAND_PART_SET = 'set_a'
|
||||
AREA_COMMAND_UNSET = 'unset'
|
||||
|
||||
def __init__(self, hass, api_url, ws_url):
|
||||
"""Initialize the web gateway client."""
|
||||
self._hass = hass
|
||||
self._api_url = api_url
|
||||
self._ws_url = ws_url
|
||||
self._ws = None
|
||||
|
||||
@asyncio.coroutine
|
||||
def get_zones(self):
|
||||
"""Retrieve all available zones."""
|
||||
return (yield from self._get_data('zone'))
|
||||
|
||||
@asyncio.coroutine
|
||||
def get_areas(self):
|
||||
"""Retrieve all available areas."""
|
||||
return (yield from self._get_data('area'))
|
||||
|
||||
@asyncio.coroutine
|
||||
def send_area_command(self, area_id, command):
|
||||
"""Send an area command."""
|
||||
_LOGGER.debug("Sending SPC area command '%s' to area %s.",
|
||||
command, area_id)
|
||||
resource = "area/{}/{}".format(area_id, command)
|
||||
return (yield from self._call_web_gateway(resource, use_get=False))
|
||||
|
||||
def start_listener(self, async_callback, *args):
|
||||
"""Start the websocket listener."""
|
||||
try:
|
||||
from asyncio import ensure_future
|
||||
except ImportError:
|
||||
from asyncio import async as ensure_future
|
||||
|
||||
ensure_future(self._ws_listen(async_callback, *args))
|
||||
|
||||
def _build_url(self, resource):
|
||||
return urljoin(self._api_url, "spc/{}".format(resource))
|
||||
|
||||
@asyncio.coroutine
|
||||
def _get_data(self, resource):
|
||||
data = yield from self._call_web_gateway(resource)
|
||||
if not data:
|
||||
return False
|
||||
if data['status'] != 'success':
|
||||
_LOGGER.error("SPC Web Gateway call unsuccessful "
|
||||
"for resource '%s'.", resource)
|
||||
return False
|
||||
return [item for item in data['data'][resource]]
|
||||
|
||||
@asyncio.coroutine
|
||||
def _call_web_gateway(self, resource, use_get=True):
|
||||
response = None
|
||||
session = None
|
||||
url = self._build_url(resource)
|
||||
try:
|
||||
_LOGGER.debug("Attempting to retrieve SPC data from %s.", url)
|
||||
session = aiohttp.ClientSession()
|
||||
with async_timeout.timeout(10, loop=self._hass.loop):
|
||||
action = session.get if use_get else session.put
|
||||
response = yield from action(url)
|
||||
if response.status != 200:
|
||||
_LOGGER.error("SPC Web Gateway returned http "
|
||||
"status %d, response %s.",
|
||||
response.status, (yield from response.text()))
|
||||
return False
|
||||
result = yield from response.json()
|
||||
except asyncio.TimeoutError:
|
||||
_LOGGER.error("Timeout getting SPC data from %s.", url)
|
||||
return False
|
||||
except aiohttp.ClientError:
|
||||
_LOGGER.exception("Error getting SPC data from %s.", url)
|
||||
return False
|
||||
finally:
|
||||
if session:
|
||||
yield from session.close()
|
||||
if response:
|
||||
yield from response.release()
|
||||
_LOGGER.debug("Data from SPC: %s", result)
|
||||
return result
|
||||
|
||||
@asyncio.coroutine
|
||||
def _ws_read(self):
|
||||
import websockets as wslib
|
||||
|
||||
try:
|
||||
if not self._ws:
|
||||
self._ws = yield from wslib.connect(self._ws_url)
|
||||
_LOGGER.info("Connected to websocket at %s.", self._ws_url)
|
||||
except Exception as ws_exc: # pylint: disable=broad-except
|
||||
_LOGGER.error("Failed to connect to websocket: %s", ws_exc)
|
||||
return
|
||||
|
||||
result = None
|
||||
|
||||
try:
|
||||
result = yield from self._ws.recv()
|
||||
_LOGGER.debug("Data from websocket: %s", result)
|
||||
except Exception as ws_exc: # pylint: disable=broad-except
|
||||
_LOGGER.error("Failed to read from websocket: %s", ws_exc)
|
||||
try:
|
||||
yield from self._ws.close()
|
||||
finally:
|
||||
self._ws = None
|
||||
|
||||
return result
|
||||
|
||||
@asyncio.coroutine
|
||||
def _ws_listen(self, async_callback, *args):
|
||||
try:
|
||||
while True:
|
||||
result = yield from self._ws_read()
|
||||
|
||||
if result:
|
||||
yield from _ws_process_message(json.loads(result),
|
||||
async_callback, *args)
|
||||
else:
|
||||
_LOGGER.info("Trying again in 30 seconds.")
|
||||
yield from asyncio.sleep(30)
|
||||
|
||||
finally:
|
||||
if self._ws:
|
||||
yield from self._ws.close()
|
||||
|
|
@ -873,6 +873,7 @@ wakeonlan==0.2.2
|
|||
# homeassistant.components.media_player.gpmdp
|
||||
websocket-client==0.37.0
|
||||
|
||||
# homeassistant.components.spc
|
||||
# homeassistant.components.media_player.webostv
|
||||
websockets==3.2
|
||||
|
||||
|
|
|
|||
|
|
@ -0,0 +1,64 @@
|
|||
"""Tests for Vanderbilt SPC alarm control panel platform."""
|
||||
import asyncio
|
||||
|
||||
import pytest
|
||||
|
||||
from homeassistant.components.spc import SpcRegistry
|
||||
from homeassistant.components.alarm_control_panel import spc
|
||||
from tests.common import async_test_home_assistant
|
||||
from homeassistant.const import (
|
||||
STATE_ALARM_ARMED_AWAY, STATE_ALARM_DISARMED)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def hass(loop):
|
||||
"""Home Assistant fixture with device mapping registry."""
|
||||
hass = loop.run_until_complete(async_test_home_assistant(loop))
|
||||
hass.data['spc_registry'] = SpcRegistry()
|
||||
hass.data['spc_api'] = None
|
||||
yield hass
|
||||
loop.run_until_complete(hass.async_stop())
|
||||
|
||||
|
||||
@asyncio.coroutine
|
||||
def test_setup_platform(hass):
|
||||
"""Test adding areas as separate alarm control panel devices."""
|
||||
added_entities = []
|
||||
|
||||
def add_entities(entities):
|
||||
nonlocal added_entities
|
||||
added_entities = list(entities)
|
||||
|
||||
areas = {'areas': [{
|
||||
'id': '1',
|
||||
'name': 'House',
|
||||
'mode': '3',
|
||||
'last_set_time': '1485759851',
|
||||
'last_set_user_id': '1',
|
||||
'last_set_user_name': 'Pelle',
|
||||
'last_unset_time': '1485800564',
|
||||
'last_unset_user_id': '1',
|
||||
'last_unset_user_name': 'Pelle',
|
||||
'last_alarm': '1478174896'
|
||||
}, {
|
||||
'id': '3',
|
||||
'name': 'Garage',
|
||||
'mode': '0',
|
||||
'last_set_time': '1483705803',
|
||||
'last_set_user_id': '9998',
|
||||
'last_set_user_name': 'Lisa',
|
||||
'last_unset_time': '1483705808',
|
||||
'last_unset_user_id': '9998',
|
||||
'last_unset_user_name': 'Lisa'
|
||||
}]}
|
||||
|
||||
yield from spc.async_setup_platform(hass=hass,
|
||||
config={},
|
||||
async_add_entities=add_entities,
|
||||
discovery_info=areas)
|
||||
|
||||
assert len(added_entities) == 2
|
||||
assert added_entities[0].name == 'House'
|
||||
assert added_entities[0].state == STATE_ALARM_ARMED_AWAY
|
||||
assert added_entities[1].name == 'Garage'
|
||||
assert added_entities[1].state == STATE_ALARM_DISARMED
|
||||
|
|
@ -0,0 +1,67 @@
|
|||
"""Tests for Vanderbilt SPC binary sensor platform."""
|
||||
import asyncio
|
||||
|
||||
import pytest
|
||||
|
||||
from homeassistant.components.spc import SpcRegistry
|
||||
from homeassistant.components.binary_sensor import spc
|
||||
from tests.common import async_test_home_assistant
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def hass(loop):
|
||||
"""Home Assistant fixture with device mapping registry."""
|
||||
hass = loop.run_until_complete(async_test_home_assistant(loop))
|
||||
hass.data['spc_registry'] = SpcRegistry()
|
||||
yield hass
|
||||
loop.run_until_complete(hass.async_stop())
|
||||
|
||||
|
||||
@asyncio.coroutine
|
||||
def test_setup_platform(hass):
|
||||
"""Test autodiscovery of supported device types."""
|
||||
added_entities = []
|
||||
|
||||
zones = {'devices': [{
|
||||
'id': '1',
|
||||
'type': '3',
|
||||
'zone_name': 'Kitchen smoke',
|
||||
'area': '1',
|
||||
'area_name': 'House',
|
||||
'input': '0',
|
||||
'status': '0',
|
||||
}, {
|
||||
'id': '3',
|
||||
'type': '0',
|
||||
'zone_name': 'Hallway PIR',
|
||||
'area': '1',
|
||||
'area_name': 'House',
|
||||
'input': '0',
|
||||
'status': '0',
|
||||
}, {
|
||||
'id': '5',
|
||||
'type': '1',
|
||||
'zone_name': 'Front door',
|
||||
'area': '1',
|
||||
'area_name': 'House',
|
||||
'input': '1',
|
||||
'status': '0',
|
||||
}]}
|
||||
|
||||
def add_entities(entities):
|
||||
nonlocal added_entities
|
||||
added_entities = list(entities)
|
||||
|
||||
yield from spc.async_setup_platform(hass=hass,
|
||||
config={},
|
||||
async_add_entities=add_entities,
|
||||
discovery_info=zones)
|
||||
|
||||
assert len(added_entities) == 3
|
||||
assert added_entities[0].device_class == 'smoke'
|
||||
assert added_entities[0].state == 'off'
|
||||
assert added_entities[1].device_class == 'motion'
|
||||
assert added_entities[1].state == 'off'
|
||||
assert added_entities[2].device_class == 'opening'
|
||||
assert added_entities[2].state == 'on'
|
||||
assert all(d.hidden for d in added_entities)
|
||||
|
|
@ -0,0 +1,149 @@
|
|||
"""Tests for Vanderbilt SPC component."""
|
||||
import asyncio
|
||||
|
||||
import pytest
|
||||
|
||||
from homeassistant.components import spc
|
||||
from homeassistant.bootstrap import async_setup_component
|
||||
from tests.common import async_test_home_assistant
|
||||
from tests.test_util.aiohttp import mock_aiohttp_client
|
||||
from homeassistant.const import (STATE_ALARM_ARMED_HOME, STATE_ALARM_DISARMED)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def hass(loop):
|
||||
"""Home Assistant fixture with device mapping registry."""
|
||||
hass = loop.run_until_complete(async_test_home_assistant(loop))
|
||||
hass.data[spc.DATA_REGISTRY] = spc.SpcRegistry()
|
||||
hass.data[spc.DATA_API] = None
|
||||
yield hass
|
||||
loop.run_until_complete(hass.async_stop())
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def spcwebgw(hass):
|
||||
"""Fixture for the SPC Web Gateway API configured for localhost."""
|
||||
yield spc.SpcWebGateway(hass=hass,
|
||||
api_url='http://localhost/',
|
||||
ws_url='ws://localhost/')
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def aioclient_mock():
|
||||
"""HTTP client mock for areas and zones."""
|
||||
areas = """{"status":"success","data":{"area":[{"id":"1","name":"House",
|
||||
"mode":"0","last_set_time":"1485759851","last_set_user_id":"1",
|
||||
"last_set_user_name":"Pelle","last_unset_time":"1485800564",
|
||||
"last_unset_user_id":"1","last_unset_user_name":"Pelle","last_alarm":
|
||||
"1478174896"},{"id":"3","name":"Garage","mode":"0","last_set_time":
|
||||
"1483705803","last_set_user_id":"9998","last_set_user_name":"Lisa",
|
||||
"last_unset_time":"1483705808","last_unset_user_id":"9998",
|
||||
"last_unset_user_name":"Lisa"}]}}"""
|
||||
|
||||
zones = """{"status":"success","data":{"zone":[{"id":"1","type":"3",
|
||||
"zone_name":"Kitchen smoke","area":"1","area_name":"House","input":"0",
|
||||
"logic_input":"0","status":"0","proc_state":"0","inhibit_allowed":"1",
|
||||
"isolate_allowed":"1"},{"id":"3","type":"0","zone_name":"Hallway PIR",
|
||||
"area":"1","area_name":"House","input":"0","logic_input":"0","status":
|
||||
"0","proc_state":"0","inhibit_allowed":"1","isolate_allowed":"1"},
|
||||
{"id":"5","type":"1","zone_name":"Front door","area":"1","area_name":
|
||||
"House","input":"1","logic_input":"0","status":"0","proc_state":"0",
|
||||
"inhibit_allowed":"1","isolate_allowed":"1"}]}}"""
|
||||
|
||||
with mock_aiohttp_client() as mock_session:
|
||||
mock_session.get('http://localhost/spc/area', text=areas)
|
||||
mock_session.get('http://localhost/spc/zone', text=zones)
|
||||
yield mock_session
|
||||
|
||||
|
||||
@asyncio.coroutine
|
||||
def test_update_alarm_device(hass, aioclient_mock, monkeypatch):
|
||||
"""Test that alarm panel state changes on incoming websocket data."""
|
||||
monkeypatch.setattr("homeassistant.components.spc.SpcWebGateway."
|
||||
"start_listener", lambda x, *args: None)
|
||||
config = {
|
||||
'spc': {
|
||||
'api_url': 'http://localhost/',
|
||||
'ws_url': 'ws://localhost/'
|
||||
}
|
||||
}
|
||||
yield from async_setup_component(hass, 'spc', config)
|
||||
yield from hass.async_block_till_done()
|
||||
|
||||
entity_id = 'alarm_control_panel.house'
|
||||
|
||||
assert hass.states.get(entity_id).state == STATE_ALARM_DISARMED
|
||||
|
||||
msg = {"sia_code": "NL", "sia_address": "1", "description": "House|Sam|1"}
|
||||
yield from spc._async_process_message(msg, hass.data[spc.DATA_REGISTRY])
|
||||
assert hass.states.get(entity_id).state == STATE_ALARM_ARMED_HOME
|
||||
|
||||
msg = {"sia_code": "OQ", "sia_address": "1", "description": "Sam"}
|
||||
yield from spc._async_process_message(msg, hass.data[spc.DATA_REGISTRY])
|
||||
assert hass.states.get(entity_id).state == STATE_ALARM_DISARMED
|
||||
|
||||
|
||||
@asyncio.coroutine
|
||||
def test_update_sensor_device(hass, aioclient_mock, monkeypatch):
|
||||
"""Test that sensors change state on incoming websocket data."""
|
||||
monkeypatch.setattr("homeassistant.components.spc.SpcWebGateway."
|
||||
"start_listener", lambda x, *args: None)
|
||||
config = {
|
||||
'spc': {
|
||||
'api_url': 'http://localhost/',
|
||||
'ws_url': 'ws://localhost/'
|
||||
}
|
||||
}
|
||||
yield from async_setup_component(hass, 'spc', config)
|
||||
yield from hass.async_block_till_done()
|
||||
|
||||
assert hass.states.get('binary_sensor.hallway_pir').state == 'off'
|
||||
|
||||
msg = {"sia_code": "ZO", "sia_address": "3", "description": "Hallway PIR"}
|
||||
yield from spc._async_process_message(msg, hass.data[spc.DATA_REGISTRY])
|
||||
assert hass.states.get('binary_sensor.hallway_pir').state == 'on'
|
||||
|
||||
msg = {"sia_code": "ZC", "sia_address": "3", "description": "Hallway PIR"}
|
||||
yield from spc._async_process_message(msg, hass.data[spc.DATA_REGISTRY])
|
||||
assert hass.states.get('binary_sensor.hallway_pir').state == 'off'
|
||||
|
||||
|
||||
class TestSpcRegistry:
|
||||
"""Test the device mapping registry."""
|
||||
|
||||
def test_sensor_device(self):
|
||||
"""Test retrieving device based on ID."""
|
||||
r = spc.SpcRegistry()
|
||||
r.register_sensor_device('1', 'dummy')
|
||||
assert r.get_sensor_device('1') == 'dummy'
|
||||
|
||||
def test_alarm_device(self):
|
||||
"""Test retrieving device based on zone name."""
|
||||
r = spc.SpcRegistry()
|
||||
r.register_alarm_device('Area 51', 'dummy')
|
||||
assert r.get_alarm_device('Area 51') == 'dummy'
|
||||
|
||||
|
||||
class TestSpcWebGateway:
|
||||
"""Test the SPC Web Gateway API wrapper."""
|
||||
|
||||
@asyncio.coroutine
|
||||
def test_get_areas(self, spcwebgw, aioclient_mock):
|
||||
"""Test area retrieval."""
|
||||
result = yield from spcwebgw.get_areas()
|
||||
assert aioclient_mock.call_count == 1
|
||||
assert len(list(result)) == 2
|
||||
|
||||
@asyncio.coroutine
|
||||
@pytest.mark.parametrize("url_command,command", [
|
||||
('set', spc.SpcWebGateway.AREA_COMMAND_SET),
|
||||
('unset', spc.SpcWebGateway.AREA_COMMAND_UNSET),
|
||||
('set_a', spc.SpcWebGateway.AREA_COMMAND_PART_SET)
|
||||
])
|
||||
def test_area_commands(self, spcwebgw, url_command, command):
|
||||
"""Test alarm arming/disarming."""
|
||||
with mock_aiohttp_client() as aioclient_mock:
|
||||
url = "http://localhost/spc/area/1/{}".format(url_command)
|
||||
aioclient_mock.put(url, text='{}')
|
||||
yield from spcwebgw.send_area_command('1', command)
|
||||
assert aioclient_mock.call_count == 1
|
||||
Loading…
Reference in New Issue