jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-04-23 08:47:57 +00:00
Code Issues Packages Projects Releases Wiki Activity

Config validation for rfxtrx sensor (#1780)

Browse Source
This commit is contained in:
Daniel Høyer Iversen 2016-04-11 01:05:32 +02:00 committed by Paulus Schoutsen
parent 4f834ba3f1
commit 769d958464
4 changed files with 111 additions and 122 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
homeassistant/components/rfxtrx.py
View File

@ -38,16 +38,22 @@ _LOGGER = logging.getLogger(__name__)
RFXOBJECT = None
def _validate_packetid(value):
def validate_packetid(value):
"""Validate that value is a valid packet id for rfxtrx."""
if get_rfx_object(value):
return value
else:
raise vol.Invalid('invalid packet id for {}'.format(value))
# Share between rfxtrx platforms
VALID_DEVICE_ID = vol.All(cv.string, vol.Lower)
VALID_SENSOR_DEVICE_ID = vol.All(VALID_DEVICE_ID,
vol.truth(lambda val:
val.startswith('sensor_')))
DEVICE_SCHEMA = vol.Schema({
vol.Required(ATTR_NAME): cv.string,
vol.Required(ATTR_PACKETID): _validate_packetid,
vol.Required(ATTR_PACKETID): validate_packetid,
vol.Optional(ATTR_FIREEVENT, default=False): cv.boolean,
})
@ -61,7 +67,7 @@ DEFAULT_SCHEMA = vol.Schema({
CONFIG_SCHEMA = vol.Schema({
DOMAIN: vol.Schema({
vol.Required(ATTR_DEVICE): cv.string,
vol.Required(ATTR_DEVICE): VALID_DEVICE_ID,
vol.Optional(ATTR_DEBUG, default=False): cv.boolean,
vol.Optional(ATTR_DUMMY, default=False): cv.boolean,
}),

40
homeassistant/components/sensor/rfxtrx.py
View File

@ -6,13 +6,16 @@ https://home-assistant.io/components/sensor.rfxtrx/
"""
import logging
from collections import OrderedDict
import voluptuous as vol
import homeassistant.components.rfxtrx as rfxtrx
from homeassistant.const import TEMP_CELCIUS
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.entity import Entity
from homeassistant.util import slugify
from homeassistant.components.rfxtrx import (
ATTR_PACKETID, ATTR_NAME, ATTR_DATA_TYPE)
ATTR_AUTOMATIC_ADD, ATTR_PACKETID, ATTR_NAME,
CONF_DEVICES, ATTR_DATA_TYPE)
DEPENDENCIES = ['rfxtrx']
@ -26,19 +29,48 @@ DATA_TYPES = OrderedDict([
('Total usage', 'W')])
_LOGGER = logging.getLogger(__name__)
DEVICE_SCHEMA = vol.Schema({
vol.Optional(ATTR_NAME, default=None): cv.string,
vol.Required(ATTR_PACKETID): rfxtrx.validate_packetid,
vol.Optional(ATTR_DATA_TYPE, default=None):
vol.In(list(DATA_TYPES.keys())),
})
def _valid_device(value):
"""Validate a dictionary of devices definitions."""
config = OrderedDict()
for key, device in value.items():
try:
key = rfxtrx.VALID_SENSOR_DEVICE_ID(key)
config[key] = DEVICE_SCHEMA(device)
if not config[key][ATTR_NAME]:
config[key][ATTR_NAME] = key
except vol.MultipleInvalid as ex:
raise vol.Invalid('Rfxtrx sensor {} is invalid: {}'
.format(key, ex))
return config
PLATFORM_SCHEMA = vol.Schema({
vol.Required("platform"): rfxtrx.DOMAIN,
vol.Required(CONF_DEVICES): vol.All(dict, _valid_device),
vol.Optional(ATTR_AUTOMATIC_ADD, default=False): cv.boolean,
}, extra=vol.ALLOW_EXTRA)
def setup_platform(hass, config, add_devices_callback, discovery_info=None):
"""Setup the RFXtrx platform."""
from RFXtrx import SensorEvent
sensors = []
for device_id, entity_info in config.get('devices', {}).items():
for device_id, entity_info in config['devices'].items():
if device_id in rfxtrx.RFX_DEVICES:
continue
_LOGGER.info("Add %s rfxtrx.sensor", entity_info[ATTR_NAME])
event = rfxtrx.get_rfx_object(entity_info[ATTR_PACKETID])
new_sensor = RfxtrxSensor(event, entity_info[ATTR_NAME],
entity_info.get(ATTR_DATA_TYPE, None))
entity_info[ATTR_DATA_TYPE])
rfxtrx.RFX_DEVICES[slugify(device_id)] = new_sensor
sensors.append(new_sensor)
@ -62,7 +94,7 @@ def setup_platform(hass, config, add_devices_callback, discovery_info=None):
return
# Add entity if not exist and the automatic_add is True
if config.get('automatic_add', True):
if config[ATTR_AUTOMATIC_ADD]:
pkt_id = "".join("{0:02x}".format(x) for x in event.data)
entity_name = "%s : %s" % (device_id, pkt_id)
_LOGGER.info(

166
tests/components/sensor/test_rfxtrx.py
View File

@ -1,8 +1,7 @@
"""The tests for the Rfxtrx sensor platform."""
import unittest
from homeassistant.components.sensor import rfxtrx
from homeassistant.bootstrap import _setup_component
from homeassistant.components import rfxtrx as rfxtrx_core
from homeassistant.const import TEMP_CELCIUS
@ -15,6 +14,7 @@ class TestSensorRfxtrx(unittest.TestCase):
def setUp(self):
"""Setup things to be run when tests are started."""
self.hass = get_test_home_assistant(0)
self.hass.config.components = ['rfxtrx']
def tearDown(self):
"""Stop everything that was started."""
@ -24,35 +24,25 @@ class TestSensorRfxtrx(unittest.TestCase):
def test_default_config(self):
"""Test with 0 sensor."""
config = {'devices': {}}
devices = []
def add_dev_callback(devs):
"""Add a callback to add devices."""
for dev in devs:
devices.append(dev)
rfxtrx.setup_platform(self.hass, config, add_dev_callback)
self.assertEqual(0, len(devices))
self.assertTrue(_setup_component(self.hass, 'sensor', {
'sensor': {'platform': 'rfxtrx',
'devices':
{}}}))
self.assertEqual(0, len(rfxtrx_core.RFX_DEVICES))
def test_one_sensor(self):
"""Test with 1 sensor."""
config = {'devices':
{'sensor_0502': {
'name': 'Test',
'packetid': '0a52080705020095220269',
'data_type': 'Temperature'}}}
devices = []
self.assertTrue(_setup_component(self.hass, 'sensor', {
'sensor': {'platform': 'rfxtrx',
'devices':
{'sensor_0502': {
'name': 'Test',
'packetid': '0a52080705020095220269',
'data_type': 'Temperature'}}}}))
def add_dev_callback(devs):
"""Add a callback to add devices."""
for dev in devs:
devices.append(dev)
rfxtrx.setup_platform(self.hass, config, add_dev_callback)
self.assertEqual(1, len(devices))
entity = devices[0]
self.assertEqual(1, len(rfxtrx_core.RFX_DEVICES))
print(rfxtrx_core.RFX_DEVICES)
entity = rfxtrx_core.RFX_DEVICES['sensor_0502']
self.assertEqual('Test', entity.name)
self.assertEqual(TEMP_CELCIUS, entity.unit_of_measurement)
self.assertEqual(14.9, entity.state)
@ -64,31 +54,25 @@ class TestSensorRfxtrx(unittest.TestCase):
def test_several_sensors(self):
"""Test with 3 sensors."""
config = {'devices':
{'sensor_0502': {
'name': 'Test',
'packetid': '0a52080705020095220269',
'data_type': 'Temperature'},
'sensor_0601': {
'name': 'Bath_Humidity',
'packetid': '0a520802060100ff0e0269',
'data_type': 'Humidity'},
'sensor_0601 2': {
'name': 'Bath',
'packetid': '0a520802060100ff0e0269'}}}
devices = []
self.assertTrue(_setup_component(self.hass, 'sensor', {
'sensor': {'platform': 'rfxtrx',
'devices':
{'sensor_0502': {
'name': 'Test',
'packetid': '0a52080705020095220269',
'data_type': 'Temperature'},
'sensor_0601': {
'name': 'Bath_Humidity',
'packetid': '0a520802060100ff0e0269',
'data_type': 'Humidity'},
'sensor_0601 2': {
'name': 'Bath',
'packetid': '0a520802060100ff0e0269'}}}}))
def add_dev_callback(devs):
"""Add a callback to add devices."""
for dev in devs:
devices.append(dev)
rfxtrx.setup_platform(self.hass, config, add_dev_callback)
self.assertEqual(3, len(devices))
self.assertEqual(3, len(rfxtrx_core.RFX_DEVICES))
device_num = 0
for entity in devices:
for id in rfxtrx_core.RFX_DEVICES:
entity = rfxtrx_core.RFX_DEVICES[id]
if entity.name == 'Bath_Humidity':
device_num = device_num + 1
self.assertEqual('%', entity.unit_of_measurement)
@ -125,23 +109,17 @@ class TestSensorRfxtrx(unittest.TestCase):
def test_discover_sensor(self):
"""Test with discovery of sensor."""
config = {'devices': {}}
devices = []
def add_dev_callback(devs):
"""Add a callback to add devices."""
for dev in devs:
devices.append(dev)
rfxtrx.setup_platform(self.hass, config, add_dev_callback)
self.assertTrue(_setup_component(self.hass, 'sensor', {
'sensor': {'platform': 'rfxtrx',
'automatic_add': True,
'devices': {}}}))
event = rfxtrx_core.get_rfx_object('0a520801070100b81b0279')
event.data = bytearray(b'\nR\x08\x01\x07\x01\x00\xb8\x1b\x02y')
rfxtrx_core.RECEIVED_EVT_SUBSCRIBERS[0](event)
entity = devices[0]
entity = rfxtrx_core.RFX_DEVICES['sensor_0701']
self.assertEqual(1, len(rfxtrx_core.RFX_DEVICES))
self.assertEqual(1, len(devices))
self.assertEqual({'Humidity status': 'normal',
'Temperature': 18.4,
'Rssi numeric': 7, 'Humidity': 27,
@ -153,14 +131,12 @@ class TestSensorRfxtrx(unittest.TestCase):
rfxtrx_core.RECEIVED_EVT_SUBSCRIBERS[0](event)
self.assertEqual(1, len(rfxtrx_core.RFX_DEVICES))
self.assertEqual(1, len(devices))
event = rfxtrx_core.get_rfx_object('0a52080405020095240279')
event.data = bytearray(b'\nR\x08\x04\x05\x02\x00\x95$\x02y')
rfxtrx_core.RECEIVED_EVT_SUBSCRIBERS[0](event)
entity = devices[1]
entity = rfxtrx_core.RFX_DEVICES['sensor_0502']
self.assertEqual(2, len(rfxtrx_core.RFX_DEVICES))
self.assertEqual(2, len(devices))
self.assertEqual({'Humidity status': 'normal',
'Temperature': 14.9,
'Rssi numeric': 7, 'Humidity': 36,
@ -173,9 +149,8 @@ class TestSensorRfxtrx(unittest.TestCase):
event = rfxtrx_core.get_rfx_object('0a52085e070100b31b0279')
event.data = bytearray(b'\nR\x08^\x07\x01\x00\xb3\x1b\x02y')
rfxtrx_core.RECEIVED_EVT_SUBSCRIBERS[0](event)
entity = devices[0]
entity = rfxtrx_core.RFX_DEVICES['sensor_0701']
self.assertEqual(2, len(rfxtrx_core.RFX_DEVICES))
self.assertEqual(2, len(devices))
self.assertEqual({'Humidity status': 'normal',
'Temperature': 17.9,
'Rssi numeric': 7, 'Humidity': 27,
@ -189,71 +164,56 @@ class TestSensorRfxtrx(unittest.TestCase):
event = rfxtrx_core.get_rfx_object('0b1100cd0213c7f210010f70')
rfxtrx_core.RECEIVED_EVT_SUBSCRIBERS[0](event)
self.assertEqual(2, len(rfxtrx_core.RFX_DEVICES))
self.assertEqual(2, len(devices))
def test_discover_sensor_noautoadd(self):
"""Test with discover of sensor when auto add is False."""
config = {'automatic_add': False, 'devices': {}}
devices = []
self.assertTrue(_setup_component(self.hass, 'sensor', {
'sensor': {'platform': 'rfxtrx',
'automatic_add': False,
'devices': {}}}))
def add_dev_callback(devs):
"""Add a callback to add devices."""
for dev in devs:
devices.append(dev)
rfxtrx.setup_platform(self.hass, config, add_dev_callback)
event = rfxtrx_core.get_rfx_object('0a520801070100b81b0279')
event.data = bytearray(b'\nR\x08\x01\x07\x01\x00\xb8\x1b\x02y')
self.assertEqual(0, len(rfxtrx_core.RFX_DEVICES))
rfxtrx_core.RECEIVED_EVT_SUBSCRIBERS[0](event)
self.assertEqual(0, len(rfxtrx_core.RFX_DEVICES))
self.assertEqual(0, len(devices))
rfxtrx_core.RECEIVED_EVT_SUBSCRIBERS[0](event)
self.assertEqual(0, len(rfxtrx_core.RFX_DEVICES))
self.assertEqual(0, len(devices))
event = rfxtrx_core.get_rfx_object('0a52080405020095240279')
event.data = bytearray(b'\nR\x08\x04\x05\x02\x00\x95$\x02y')
rfxtrx_core.RECEIVED_EVT_SUBSCRIBERS[0](event)
self.assertEqual(0, len(rfxtrx_core.RFX_DEVICES))
self.assertEqual(0, len(devices))
event = rfxtrx_core.get_rfx_object('0a52085e070100b31b0279')
event.data = bytearray(b'\nR\x08^\x07\x01\x00\xb3\x1b\x02y')
rfxtrx_core.RECEIVED_EVT_SUBSCRIBERS[0](event)
self.assertEqual(0, len(rfxtrx_core.RFX_DEVICES))
self.assertEqual(0, len(devices))
def test_update_of_sensors(self):
"""Test with 3 sensors."""
config = {'devices':
{'sensor_0502': {
'name': 'Test',
'packetid': '0a52080705020095220269',
'data_type': 'Temperature'},
'sensor_0601': {
'name': 'Bath_Humidity',
'packetid': '0a520802060100ff0e0269',
'data_type': 'Humidity'},
'sensor_0601 2': {
'name': 'Bath',
'packetid': '0a520802060100ff0e0269'}}}
devices = []
self.assertTrue(_setup_component(self.hass, 'sensor', {
'sensor': {'platform': 'rfxtrx',
'devices':
{'sensor_0502': {
'name': 'Test',
'packetid': '0a52080705020095220269',
'data_type': 'Temperature'},
'sensor_0601': {
'name': 'Bath_Humidity',
'packetid': '0a520802060100ff0e0269',
'data_type': 'Humidity'},
'sensor_0601 2': {
'name': 'Bath',
'packetid': '0a520802060100ff0e0269'}}}}))
def add_dev_callback(devs):
"""Add a callback to add devices."""
for dev in devs:
devices.append(dev)
rfxtrx.setup_platform(self.hass, config, add_dev_callback)
self.assertEqual(3, len(devices))
self.assertEqual(3, len(rfxtrx_core.RFX_DEVICES))
device_num = 0
for entity in devices:
for id in rfxtrx_core.RFX_DEVICES:
entity = rfxtrx_core.RFX_DEVICES[id]
if entity.name == 'Bath_Humidity':
device_num = device_num + 1
self.assertEqual('%', entity.unit_of_measurement)
@ -291,18 +251,17 @@ class TestSensorRfxtrx(unittest.TestCase):
event = rfxtrx_core.get_rfx_object('0a520802060101ff0f0269')
event.data = bytearray(b'\nR\x08\x01\x07\x01\x00\xb8\x1b\x02y')
rfxtrx_core.RECEIVED_EVT_SUBSCRIBERS[0](event)
entity = devices[0]
rfxtrx_core.RECEIVED_EVT_SUBSCRIBERS[0](event)
event = rfxtrx_core.get_rfx_object('0a52080705020085220269')
event.data = bytearray(b'\nR\x08\x04\x05\x02\x00\x95$\x02y')
rfxtrx_core.RECEIVED_EVT_SUBSCRIBERS[0](event)
self.assertEqual(3, len(devices))
self.assertEqual(3, len(rfxtrx_core.RFX_DEVICES))
device_num = 0
for entity in devices:
for id in rfxtrx_core.RFX_DEVICES:
entity = rfxtrx_core.RFX_DEVICES[id]
if entity.name == 'Bath_Humidity':
device_num = device_num + 1
self.assertEqual('%', entity.unit_of_measurement)
@ -337,5 +296,4 @@ class TestSensorRfxtrx(unittest.TestCase):
self.assertEqual(3, device_num)
self.assertEqual(3, len(devices))
self.assertEqual(3, len(rfxtrx_core.RFX_DEVICES))

15
tests/components/test_rfxtrx.py
View File

@ -5,7 +5,6 @@ import time
from homeassistant.bootstrap import _setup_component
from homeassistant.components import rfxtrx as rfxtrx
from homeassistant.components.sensor import rfxtrx as rfxtrx_sensor
from tests.common import get_test_home_assistant
@ -33,21 +32,15 @@ class TestRFXTRX(unittest.TestCase):
'dummy': True}
}))
config = {'devices': {}}
devices = []
def add_dev_callback(devs):
"""Add a callback to add devices."""
for dev in devs:
devices.append(dev)
rfxtrx_sensor.setup_platform(self.hass, config, add_dev_callback)
self.assertTrue(_setup_component(self.hass, 'sensor', {
'sensor': {'platform': 'rfxtrx',
'automatic_add': True,
'devices': {}}}))
while len(rfxtrx.RFX_DEVICES) < 2:
time.sleep(0.1)
self.assertEqual(len(rfxtrx.RFXOBJECT.sensors()), 2)
self.assertEqual(len(devices), 2)
def test_valid_config(self):
"""Test configuration."""