mirror of https://github.com/home-assistant/core
Add nx584 as a sensor platform
This allows you to get every door, window, smoke, etc zone from your security panel into HA. This uses the live eventing feature of pynx584, which means you get instantaneous signaling into HA when a door opens or something happens, which is handy for automating lights on when doors open after dark, etc. Requires update to pynx584 0.2
This commit is contained in:
Dan Smith
parent
2ba3df2257
commit
22865e5d96
|
|
@ -14,7 +14,7 @@ from homeassistant.const import (STATE_UNKNOWN, STATE_ALARM_DISARMED,
|
|||
STATE_ALARM_ARMED_AWAY)
|
||||
import homeassistant.components.alarm_control_panel as alarm
|
||||
|
||||
REQUIREMENTS = ['pynx584==0.1']
|
||||
REQUIREMENTS = ['pynx584==0.2']
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
|
|
|||
|
|
@ -0,0 +1,115 @@
|
|||
"""
|
||||
homeassistant.components.sensor.nx584
|
||||
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
|
||||
Support for exposing nx584 elements as sensors.
|
||||
|
||||
For more details about this platform, please refer to the documentation at
|
||||
https://home-assistant.io/components/sensor.nx584/
|
||||
"""
|
||||
import logging
|
||||
import threading
|
||||
import time
|
||||
|
||||
import requests
|
||||
|
||||
from homeassistant.components.binary_sensor import BinarySensorDevice
|
||||
|
||||
REQUIREMENTS = ['pynx584==0.2']
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def setup_platform(hass, config, add_devices, discovery_info=None):
|
||||
"""Setup nx584 sensors."""
|
||||
from nx584 import client as nx584_client
|
||||
|
||||
host = config.get('host', 'localhost:5007')
|
||||
exclude = config.get('exclude_zones', [])
|
||||
|
||||
if not all(isinstance(zone, int) for zone in exclude):
|
||||
_LOGGER.error('Invalid excluded zone specified (use zone number)')
|
||||
return False
|
||||
|
||||
try:
|
||||
client = nx584_client.Client('http://%s' % host)
|
||||
zones = client.list_zones()
|
||||
except requests.exceptions.ConnectionError as ex:
|
||||
_LOGGER.error('Unable to connect to NX584: %s', str(ex))
|
||||
return False
|
||||
|
||||
version = [int(v) for v in client.get_version().split('.')]
|
||||
if version < [1, 1]:
|
||||
_LOGGER.error('NX584 is too old to use for sensors (>=0.2 required)')
|
||||
return False
|
||||
|
||||
zone_sensors = {
|
||||
zone['number']: NX584ZoneSensor(zone)
|
||||
for zone in zones
|
||||
if zone['number'] not in exclude}
|
||||
if zone_sensors:
|
||||
add_devices(zone_sensors.values())
|
||||
watcher = NX584Watcher(client, zone_sensors)
|
||||
watcher.start()
|
||||
else:
|
||||
_LOGGER.warning('No zones found on NX584')
|
||||
|
||||
return True
|
||||
|
||||
|
||||
class NX584ZoneSensor(BinarySensorDevice):
|
||||
"""Represents a NX584 zone as a sensor."""
|
||||
|
||||
def __init__(self, zone):
|
||||
self._zone = zone
|
||||
|
||||
@property
|
||||
def should_poll(self):
|
||||
return False
|
||||
|
||||
@property
|
||||
def name(self):
|
||||
return self._zone['name']
|
||||
|
||||
@property
|
||||
def is_on(self):
|
||||
# True means "faulted" or "open" or "abnormal state"
|
||||
return self._zone['state']
|
||||
|
||||
|
||||
class NX584Watcher(threading.Thread):
|
||||
"""Event listener thread to process NX584 events."""
|
||||
|
||||
def __init__(self, client, zone_sensors):
|
||||
super(NX584Watcher, self).__init__()
|
||||
self.daemon = True
|
||||
self._client = client
|
||||
self._zone_sensors = zone_sensors
|
||||
|
||||
def _process_zone_event(self, event):
|
||||
zone = event['zone']
|
||||
zone_sensor = self._zone_sensors.get(zone)
|
||||
# pylint: disable=protected-access
|
||||
if not zone_sensor:
|
||||
return
|
||||
zone_sensor._zone['state'] = event['zone_state']
|
||||
zone_sensor.update_ha_state()
|
||||
|
||||
def _process_events(self, events):
|
||||
for event in events:
|
||||
if event.get('type') == 'zone_status':
|
||||
self._process_zone_event(event)
|
||||
|
||||
def _run(self):
|
||||
# Throw away any existing events so we don't replay history
|
||||
self._client.get_events()
|
||||
while True:
|
||||
events = self._client.get_events()
|
||||
if events:
|
||||
self._process_events(events)
|
||||
|
||||
def run(self):
|
||||
while True:
|
||||
try:
|
||||
self._run()
|
||||
except requests.exceptions.ConnectionError:
|
||||
_LOGGER.error('Failed to reach NX584 server')
|
||||
time.sleep(10)
|
||||
|
|
@ -161,7 +161,8 @@ pyicloud==0.7.2
|
|||
pynetgear==0.3.2
|
||||
|
||||
# homeassistant.components.alarm_control_panel.nx584
|
||||
pynx584==0.1
|
||||
# homeassistant.components.binary_sensor.nx584
|
||||
pynx584==0.2
|
||||
|
||||
# homeassistant.components.sensor.openweathermap
|
||||
pyowm==2.3.0
|
||||
|
|
|
|||
|
|
@ -0,0 +1,173 @@
|
|||
"""
|
||||
tests.components.binary_sensor.nx584
|
||||
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
|
||||
Tests for nx584 sensor.
|
||||
"""
|
||||
|
||||
import requests
|
||||
import unittest
|
||||
from unittest import mock
|
||||
|
||||
from homeassistant.components.binary_sensor import nx584
|
||||
from nx584 import client as nx584_client
|
||||
|
||||
|
||||
class StopMe(Exception):
|
||||
pass
|
||||
|
||||
|
||||
class TestNX584SensorSetup(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self._mock_client = mock.patch.object(nx584_client, 'Client')
|
||||
self._mock_client.start()
|
||||
|
||||
self.fake_zones = [
|
||||
{'name': 'front', 'number': 1},
|
||||
{'name': 'back', 'number': 2},
|
||||
{'name': 'inside', 'number': 3},
|
||||
]
|
||||
|
||||
client = nx584_client.Client.return_value
|
||||
client.list_zones.return_value = self.fake_zones
|
||||
client.get_version.return_value = '1.1'
|
||||
|
||||
def tearDown(self):
|
||||
self._mock_client.stop()
|
||||
|
||||
@mock.patch('homeassistant.components.binary_sensor.nx584.NX584Watcher')
|
||||
@mock.patch('homeassistant.components.binary_sensor.nx584.NX584ZoneSensor')
|
||||
def test_setup_no_config(self, mock_nx, mock_watcher):
|
||||
add_devices = mock.MagicMock()
|
||||
hass = mock.MagicMock()
|
||||
self.assertTrue(nx584.setup_platform(hass, {}, add_devices))
|
||||
mock_nx.assert_has_calls([
|
||||
mock.call(zone)
|
||||
for zone in self.fake_zones])
|
||||
self.assertTrue(add_devices.called)
|
||||
nx584_client.Client.assert_called_once_with('http://localhost:5007')
|
||||
|
||||
@mock.patch('homeassistant.components.binary_sensor.nx584.NX584Watcher')
|
||||
@mock.patch('homeassistant.components.binary_sensor.nx584.NX584ZoneSensor')
|
||||
def test_setup_full_config(self, mock_nx, mock_watcher):
|
||||
config = {
|
||||
'host': 'foo:123',
|
||||
'exclude_zones': [2],
|
||||
'zone_types': {3: 'motion'},
|
||||
}
|
||||
add_devices = mock.MagicMock()
|
||||
hass = mock.MagicMock()
|
||||
self.assertTrue(nx584.setup_platform(hass, config, add_devices))
|
||||
mock_nx.assert_has_calls([
|
||||
mock.call(self.fake_zones[0]),
|
||||
mock.call(self.fake_zones[2]),
|
||||
])
|
||||
self.assertTrue(add_devices.called)
|
||||
nx584_client.Client.assert_called_once_with('http://foo:123')
|
||||
self.assertTrue(mock_watcher.called)
|
||||
|
||||
def _test_assert_graceful_fail(self, config):
|
||||
hass = add_devices = mock.MagicMock()
|
||||
self.assertFalse(nx584.setup_platform(hass, config,
|
||||
add_devices))
|
||||
self.assertFalse(add_devices.called)
|
||||
|
||||
def test_setup_bad_config(self):
|
||||
bad_configs = [
|
||||
{'exclude_zones': ['a']},
|
||||
]
|
||||
for config in bad_configs:
|
||||
self._test_assert_graceful_fail(config)
|
||||
|
||||
def test_setup_connect_failed(self):
|
||||
nx584_client.Client.return_value.list_zones.side_effect = \
|
||||
requests.exceptions.ConnectionError
|
||||
self._test_assert_graceful_fail({})
|
||||
|
||||
def test_setup_version_too_old(self):
|
||||
nx584_client.Client.return_value.get_version.return_value = '1.0'
|
||||
self._test_assert_graceful_fail({})
|
||||
|
||||
def test_setup_no_zones(self):
|
||||
nx584_client.Client.return_value.list_zones.return_value = []
|
||||
hass = add_devices = mock.MagicMock()
|
||||
self.assertTrue(nx584.setup_platform(hass, {},
|
||||
add_devices))
|
||||
self.assertFalse(add_devices.called)
|
||||
|
||||
|
||||
class TestNX584ZoneSensor(unittest.TestCase):
|
||||
def test_sensor_normal(self):
|
||||
zone = {'number': 1, 'name': 'foo', 'state': True}
|
||||
sensor = nx584.NX584ZoneSensor(zone)
|
||||
self.assertEqual('foo', sensor.name)
|
||||
self.assertFalse(sensor.should_poll)
|
||||
self.assertTrue(sensor.is_on)
|
||||
|
||||
zone['state'] = False
|
||||
self.assertFalse(sensor.is_on)
|
||||
|
||||
|
||||
class TestNX584Watcher(unittest.TestCase):
|
||||
@mock.patch.object(nx584.NX584ZoneSensor, 'update_ha_state')
|
||||
def test_process_zone_event(self, mock_update):
|
||||
zone1 = {'number': 1, 'name': 'foo', 'state': True}
|
||||
zone2 = {'number': 2, 'name': 'bar', 'state': True}
|
||||
zones = {
|
||||
1: nx584.NX584ZoneSensor(zone1),
|
||||
2: nx584.NX584ZoneSensor(zone2),
|
||||
}
|
||||
watcher = nx584.NX584Watcher(None, zones)
|
||||
watcher._process_zone_event({'zone': 1, 'zone_state': False})
|
||||
self.assertFalse(zone1['state'])
|
||||
self.assertEqual(1, mock_update.call_count)
|
||||
|
||||
@mock.patch.object(nx584.NX584ZoneSensor, 'update_ha_state')
|
||||
def test_process_zone_event_missing_zone(self, mock_update):
|
||||
watcher = nx584.NX584Watcher(None, {})
|
||||
watcher._process_zone_event({'zone': 1, 'zone_state': False})
|
||||
self.assertFalse(mock_update.called)
|
||||
|
||||
def test_run_with_zone_events(self):
|
||||
empty_me = [1, 2]
|
||||
|
||||
def fake_get_events():
|
||||
"""Return nothing twice, then some events"""
|
||||
if empty_me:
|
||||
empty_me.pop()
|
||||
else:
|
||||
return fake_events
|
||||
|
||||
client = mock.MagicMock()
|
||||
fake_events = [
|
||||
{'zone': 1, 'zone_state': True, 'type': 'zone_status'},
|
||||
{'zone': 2, 'foo': False},
|
||||
]
|
||||
client.get_events.side_effect = fake_get_events
|
||||
watcher = nx584.NX584Watcher(client, {})
|
||||
|
||||
@mock.patch.object(watcher, '_process_zone_event')
|
||||
def run(fake_process):
|
||||
fake_process.side_effect = StopMe
|
||||
self.assertRaises(StopMe, watcher._run)
|
||||
fake_process.assert_called_once_with(fake_events[0])
|
||||
|
||||
run()
|
||||
self.assertEqual(3, client.get_events.call_count)
|
||||
|
||||
@mock.patch('time.sleep')
|
||||
def test_run_retries_failures(self, mock_sleep):
|
||||
empty_me = [1, 2]
|
||||
|
||||
def fake_run():
|
||||
if empty_me:
|
||||
empty_me.pop()
|
||||
raise requests.exceptions.ConnectionError()
|
||||
else:
|
||||
raise StopMe()
|
||||
|
||||
watcher = nx584.NX584Watcher(None, {})
|
||||
with mock.patch.object(watcher, '_run') as mock_inner:
|
||||
mock_inner.side_effect = fake_run
|
||||
self.assertRaises(StopMe, watcher.run)
|
||||
self.assertEqual(3, mock_inner.call_count)
|
||||
mock_sleep.assert_has_calls([mock.call(10), mock.call(10)])
|
||||
Loading…
Reference in New Issue