jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-04-24 09:17:53 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #1905 from home-assistant/rfxtrx_sensor_fix

Browse Source 
Rfxtrx: Simplified configuration
This commit is contained in:
Daniel Høyer Iversen 2016-04-26 11:54:11 +02:00
commit 6717215438
6 changed files with 380 additions and 281 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

222
homeassistant/components/rfxtrx.py
View File

@ -5,61 +5,102 @@ For more details about this component, please refer to the documentation at
https://home-assistant.io/components/rfxtrx/
"""
import logging
from collections import OrderedDict
import voluptuous as vol
import homeassistant.helpers.config_validation as cv
from homeassistant.util import slugify
from homeassistant.const import EVENT_HOMEASSISTANT_STOP
from homeassistant.helpers.entity import Entity
from homeassistant.const import ATTR_ENTITY_ID
from homeassistant.const import (ATTR_ENTITY_ID, TEMP_CELSIUS)
REQUIREMENTS = ['pyRFXtrx==0.6.5']
DOMAIN = "rfxtrx"
DEFAULT_SIGNAL_REPETITIONS = 1
ATTR_AUTOMATIC_ADD = 'automatic_add'
ATTR_DEVICE = 'device'
ATTR_DEBUG = 'debug'
ATTR_STATE = 'state'
ATTR_NAME = 'name'
ATTR_PACKETID = 'packetid'
ATTR_FIREEVENT = 'fire_event'
ATTR_DATA_TYPE = 'data_type'
ATTR_DUMMY = 'dummy'
CONF_SIGNAL_REPETITIONS = 'signal_repetitions'
CONF_DEVICES = 'devices'
DEFAULT_SIGNAL_REPETITIONS = 1
EVENT_BUTTON_PRESSED = 'button_pressed'
DATA_TYPES = OrderedDict([
('Temperature', TEMP_CELSIUS),
('Humidity', '%'),
('Barometer', ''),
('Wind direction', ''),
('Rain rate', ''),
('Energy usage', 'W'),
('Total usage', 'W')])
RECEIVED_EVT_SUBSCRIBERS = []
RFX_DEVICES = {}
_LOGGER = logging.getLogger(__name__)
RFXOBJECT = None
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))
def _valid_device(value, device_type):
"""Validate a dictionary of devices definitions."""
config = OrderedDict()
for key, device in value.items():
# 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_')))
# Still accept old configuration
if 'packetid' in device.keys():
msg = 'You are using an outdated configuration of the rfxtrx ' +\
'device, {}.'.format(key) +\
' Your new config should be:\n {}: \n name: {}'\
.format(device.get('packetid'),
device.get(ATTR_NAME, 'deivce_name'))
_LOGGER.warning(msg)
key = device.get('packetid')
device.pop('packetid')
if get_rfx_object(key) is None:
raise vol.Invalid('Rfxtrx device {} is invalid: '
'Invalid device id for {}'.format(key, value))
if device_type == 'sensor':
config[key] = DEVICE_SCHEMA_SENSOR(device)
elif device_type == 'light_switch':
config[key] = DEVICE_SCHEMA(device)
else:
raise vol.Invalid('Rfxtrx device is invalid')
if not config[key][ATTR_NAME]:
config[key][ATTR_NAME] = key
return config
def valid_sensor(value):
"""Validate sensor configuration."""
return _valid_device(value, "sensor")
def _valid_light_switch(value):
return _valid_device(value, "light_switch")
DEVICE_SCHEMA = vol.Schema({
vol.Required(ATTR_NAME): cv.string,
vol.Required(ATTR_PACKETID): validate_packetid,
vol.Optional(ATTR_FIREEVENT, default=False): cv.boolean,
})
DEVICE_SCHEMA_SENSOR = vol.Schema({
vol.Optional(ATTR_NAME, default=None): cv.string,
vol.Optional(ATTR_DATA_TYPE, default=[]):
vol.All(cv.ensure_list, [vol.In(DATA_TYPES.keys())]),
})
DEFAULT_SCHEMA = vol.Schema({
vol.Required("platform"): DOMAIN,
vol.Required(CONF_DEVICES): {cv.slug: DEVICE_SCHEMA},
vol.Optional(CONF_DEVICES, default={}): vol.All(dict, _valid_light_switch),
vol.Optional(ATTR_AUTOMATIC_ADD, default=False): cv.boolean,
vol.Optional(CONF_SIGNAL_REPETITIONS, default=DEFAULT_SIGNAL_REPETITIONS):
vol.Coerce(int),
@ -82,11 +123,7 @@ def setup(hass, config):
# Log RFXCOM event
if not event.device.id_string:
return
entity_id = slugify(event.device.id_string.lower())
packet_id = "".join("{0:02x}".format(x) for x in event.data)
entity_name = "%s : %s" % (entity_id, packet_id)
_LOGGER.info("Receive RFXCOM event from %s => %s",
event.device, entity_name)
_LOGGER.info("Receive RFXCOM event from %s", event.device)
# Callback to HA registered components.
for subscriber in RECEIVED_EVT_SUBSCRIBERS:
@ -121,19 +158,16 @@ def get_rfx_object(packetid):
import RFXtrx as rfxtrxmod
binarypacket = bytearray.fromhex(packetid)
pkt = rfxtrxmod.lowlevel.parse(binarypacket)
if pkt is not None:
if isinstance(pkt, rfxtrxmod.lowlevel.SensorPacket):
obj = rfxtrxmod.SensorEvent(pkt)
elif isinstance(pkt, rfxtrxmod.lowlevel.Status):
obj = rfxtrxmod.StatusEvent(pkt)
else:
obj = rfxtrxmod.ControlEvent(pkt)
return obj
return None
if pkt is None:
return None
if isinstance(pkt, rfxtrxmod.lowlevel.SensorPacket):
obj = rfxtrxmod.SensorEvent(pkt)
elif isinstance(pkt, rfxtrxmod.lowlevel.Status):
obj = rfxtrxmod.StatusEvent(pkt)
else:
obj = rfxtrxmod.ControlEvent(pkt)
return obj
def get_devices_from_config(config, device):
@ -141,7 +175,9 @@ def get_devices_from_config(config, device):
signal_repetitions = config[CONF_SIGNAL_REPETITIONS]
devices = []
for device_id, entity_info in config[CONF_DEVICES].items():
for packet_id, entity_info in config[CONF_DEVICES].items():
event = get_rfx_object(packet_id)
device_id = slugify(event.device.id_string.lower())
if device_id in RFX_DEVICES:
continue
_LOGGER.info("Add %s rfxtrx", entity_info[ATTR_NAME])
@ -150,8 +186,7 @@ def get_devices_from_config(config, device):
fire_event = entity_info[ATTR_FIREEVENT]
datas = {ATTR_STATE: False, ATTR_FIREEVENT: fire_event}
rfxobject = get_rfx_object(entity_info[ATTR_PACKETID])
new_device = device(entity_info[ATTR_NAME], rfxobject, datas,
new_device = device(entity_info[ATTR_NAME], event, datas,
signal_repetitions)
RFX_DEVICES[device_id] = new_device
devices.append(new_device)
@ -161,83 +196,80 @@ def get_devices_from_config(config, device):
def get_new_device(event, config, device):
"""Add entity if not exist and the automatic_add is True."""
device_id = slugify(event.device.id_string.lower())
if device_id not in RFX_DEVICES:
automatic_add = config[ATTR_AUTOMATIC_ADD]
if not automatic_add:
return
if device_id in RFX_DEVICES:
return
_LOGGER.info(
"Automatic add %s rfxtrx device (Class: %s Sub: %s)",
device_id,
event.device.__class__.__name__,
event.device.subtype
)
pkt_id = "".join("{0:02x}".format(x) for x in event.data)
entity_name = "%s : %s" % (device_id, pkt_id)
datas = {ATTR_STATE: False, ATTR_FIREEVENT: False}
signal_repetitions = config[CONF_SIGNAL_REPETITIONS]
new_device = device(entity_name, event, datas,
signal_repetitions)
RFX_DEVICES[device_id] = new_device
return new_device
if not config[ATTR_AUTOMATIC_ADD]:
return
_LOGGER.info(
"Automatic add %s rfxtrx device (Class: %s Sub: %s)",
device_id,
event.device.__class__.__name__,
event.device.subtype
)
pkt_id = "".join("{0:02x}".format(x) for x in event.data)
datas = {ATTR_STATE: False, ATTR_FIREEVENT: False}
signal_repetitions = config[CONF_SIGNAL_REPETITIONS]
new_device = device(pkt_id, event, datas,
signal_repetitions)
RFX_DEVICES[device_id] = new_device
return new_device
def apply_received_command(event):
"""Apply command from rfxtrx."""
device_id = slugify(event.device.id_string.lower())
# Check if entity exists or previously added automatically
if device_id in RFX_DEVICES:
_LOGGER.debug(
"EntityID: %s device_update. Command: %s",
device_id,
event.values['Command']
if device_id not in RFX_DEVICES:
return
_LOGGER.debug(
"EntityID: %s device_update. Command: %s",
device_id,
event.values['Command']
)
if event.values['Command'] == 'On'\
or event.values['Command'] == 'Off':
# Update the rfxtrx device state
is_on = event.values['Command'] == 'On'
RFX_DEVICES[device_id].update_state(is_on)
elif hasattr(RFX_DEVICES[device_id], 'brightness')\
and event.values['Command'] == 'Set level':
_brightness = (event.values['Dim level'] * 255 // 100)
# Update the rfxtrx device state
is_on = _brightness > 0
RFX_DEVICES[device_id].update_state(is_on, _brightness)
# Fire event
if RFX_DEVICES[device_id].should_fire_event:
RFX_DEVICES[device_id].hass.bus.fire(
EVENT_BUTTON_PRESSED, {
ATTR_ENTITY_ID:
RFX_DEVICES[device_id].entity_id,
ATTR_STATE: event.values['Command'].lower()
}
)
if event.values['Command'] == 'On'\
or event.values['Command'] == 'Off':
# Update the rfxtrx device state
is_on = event.values['Command'] == 'On'
# pylint: disable=protected-access
RFX_DEVICES[device_id]._state = is_on
RFX_DEVICES[device_id].update_ha_state()
elif hasattr(RFX_DEVICES[device_id], 'brightness')\
and event.values['Command'] == 'Set level':
# pylint: disable=protected-access
RFX_DEVICES[device_id]._brightness = \
(event.values['Dim level'] * 255 // 100)
# Update the rfxtrx device state
is_on = RFX_DEVICES[device_id]._brightness > 0
RFX_DEVICES[device_id]._state = is_on
RFX_DEVICES[device_id].update_ha_state()
# Fire event
if RFX_DEVICES[device_id].should_fire_event:
RFX_DEVICES[device_id].hass.bus.fire(
EVENT_BUTTON_PRESSED, {
ATTR_ENTITY_ID:
RFX_DEVICES[device_id].entity_id,
ATTR_STATE: event.values['Command'].lower()
}
)
class RfxtrxDevice(Entity):
"""Represents a Rfxtrx device.
Contains the common logic for all Rfxtrx devices.
Contains the common logic for Rfxtrx lights and switches.
"""
def __init__(self, name, event, datas, signal_repetitions):
"""Initialize the device."""
self.signal_repetitions = signal_repetitions
self._name = name
self._event = event
self._state = datas[ATTR_STATE]
self._should_fire_event = datas[ATTR_FIREEVENT]
self.signal_repetitions = signal_repetitions
self._brightness = 0
@property
@ -269,6 +301,12 @@ class RfxtrxDevice(Entity):
"""Turn the device off."""
self._send_command("turn_off")
def update_state(self, state, brightness=0):
"""Update det state of the device."""
self._state = state
self._brightness = brightness
self.update_ha_state()
def _send_command(self, command, brightness=0):
if not self._event:
return

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

@ -5,74 +5,52 @@ For more details about this platform, please refer to the documentation at
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_CELSIUS
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.entity import Entity
from homeassistant.util import slugify
from homeassistant.components.rfxtrx import (
ATTR_AUTOMATIC_ADD, ATTR_PACKETID, ATTR_NAME,
CONF_DEVICES, ATTR_DATA_TYPE)
ATTR_AUTOMATIC_ADD, ATTR_NAME,
CONF_DEVICES, ATTR_DATA_TYPE, DATA_TYPES)
DEPENDENCIES = ['rfxtrx']
DATA_TYPES = OrderedDict([
('Temperature', TEMP_CELSIUS),
('Humidity', '%'),
('Barometer', ''),
('Wind direction', ''),
('Rain rate', ''),
('Energy usage', 'W'),
('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(CONF_DEVICES, default={}): vol.All(dict, rfxtrx.valid_sensor),
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."""
# pylint: disable=too-many-locals
from RFXtrx import SensorEvent
sensors = []
for device_id, entity_info in config['devices'].items():
for packet_id, entity_info in config['devices'].items():
event = rfxtrx.get_rfx_object(packet_id)
device_id = "sensor_" + slugify(event.device.id_string.lower())
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[ATTR_DATA_TYPE])
rfxtrx.RFX_DEVICES[slugify(device_id)] = new_sensor
sensors.append(new_sensor)
sub_sensors = {}
data_types = entity_info[ATTR_DATA_TYPE]
if len(data_types) == 0:
for data_type in DATA_TYPES:
if data_type in event.values:
data_types = [data_type]
break
for _data_type in data_types:
new_sensor = RfxtrxSensor(event, entity_info[ATTR_NAME],
_data_type)
sensors.append(new_sensor)
sub_sensors[_data_type] = new_sensor
rfxtrx.RFX_DEVICES[device_id] = sub_sensors
add_devices_callback(sensors)
@ -84,27 +62,27 @@ def setup_platform(hass, config, add_devices_callback, discovery_info=None):
device_id = "sensor_" + slugify(event.device.id_string.lower())
if device_id in rfxtrx.RFX_DEVICES:
rfxtrx.RFX_DEVICES[device_id].event = event
k = 2
_device_id = device_id + "_" + str(k)
while _device_id in rfxtrx.RFX_DEVICES:
rfxtrx.RFX_DEVICES[_device_id].event = event
k = k + 1
_device_id = device_id + "_" + str(k)
sensors = rfxtrx.RFX_DEVICES[device_id]
for key in sensors:
sensors[key].event = event
return
# Add entity if not exist and the automatic_add is 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(
"Automatic add rfxtrx.sensor: (%s : %s)",
device_id,
pkt_id)
if not config[ATTR_AUTOMATIC_ADD]:
return
new_sensor = RfxtrxSensor(event, entity_name)
rfxtrx.RFX_DEVICES[device_id] = new_sensor
add_devices_callback([new_sensor])
pkt_id = "".join("{0:02x}".format(x) for x in event.data)
_LOGGER.info("Automatic add rfxtrx.sensor: %s",
device_id)
for data_type in DATA_TYPES:
if data_type in event.values:
new_sensor = RfxtrxSensor(event, pkt_id, data_type)
break
sub_sensors = {}
sub_sensors[new_sensor.data_type] = new_sensor
rfxtrx.RFX_DEVICES[device_id] = sub_sensors
add_devices_callback([new_sensor])
if sensor_update not in rfxtrx.RECEIVED_EVT_SUBSCRIBERS:
rfxtrx.RECEIVED_EVT_SUBSCRIBERS.append(sensor_update)
@ -113,21 +91,15 @@ def setup_platform(hass, config, add_devices_callback, discovery_info=None):
class RfxtrxSensor(Entity):
"""Representation of a RFXtrx sensor."""
def __init__(self, event, name, data_type=None):
def __init__(self, event, name, data_type):
"""Initialize the sensor."""
self.event = event
self._unit_of_measurement = None
self._data_type = None
self.data_type = None
self._name = name
if data_type:
self._data_type = data_type
self.data_type = data_type
self._unit_of_measurement = DATA_TYPES[data_type]
return
for data_type in DATA_TYPES:
if data_type in self.event.values:
self._unit_of_measurement = DATA_TYPES[data_type]
self._data_type = data_type
break
def __str__(self):
"""Return the name of the sensor."""
@ -136,8 +108,8 @@ class RfxtrxSensor(Entity):
@property
def state(self):
"""Return the state of the sensor."""
if self._data_type:
return self.event.values[self._data_type]
if self.data_type:
return self.event.values[self.data_type]
return None
@property

69
tests/components/light/test_rfxtrx.py
View File

@ -29,9 +29,8 @@ class TestLightRfxtrx(unittest.TestCase):
'light': {'platform': 'rfxtrx',
'automatic_add': True,
'devices':
{'213c7f216': {
{'0b1100cd0213c7f210010f51': {
'name': 'Test',
'packetid': '0b1100cd0213c7f210010f51',
rfxtrx_core.ATTR_FIREEVENT: True}}}}))
self.assertTrue(_setup_component(self.hass, 'light', {
@ -62,7 +61,7 @@ class TestLightRfxtrx(unittest.TestCase):
'devices': {}}}))
self.assertEqual(0, len(rfxtrx_core.RFX_DEVICES))
def test_one_light(self):
def test_old_config(self):
"""Test with 1 light."""
self.assertTrue(_setup_component(self.hass, 'light', {
'light': {'platform': 'rfxtrx',
@ -76,7 +75,50 @@ class TestLightRfxtrx(unittest.TestCase):
rfxtrxmod.Core("", transport_protocol=rfxtrxmod.DummyTransport)
self.assertEqual(1, len(rfxtrx_core.RFX_DEVICES))
entity = rfxtrx_core.RFX_DEVICES['123efab1']
entity = rfxtrx_core.RFX_DEVICES['213c7f216']
self.assertEqual('Test', entity.name)
self.assertEqual('off', entity.state)
self.assertTrue(entity.assumed_state)
self.assertEqual(entity.signal_repetitions, 1)
self.assertFalse(entity.should_fire_event)
self.assertFalse(entity.should_poll)
self.assertFalse(entity.is_on)
entity.turn_on()
self.assertTrue(entity.is_on)
self.assertEqual(entity.brightness, 255)
entity.turn_off()
self.assertFalse(entity.is_on)
self.assertEqual(entity.brightness, 0)
entity.turn_on(brightness=100)
self.assertTrue(entity.is_on)
self.assertEqual(entity.brightness, 100)
entity.turn_on(brightness=10)
self.assertTrue(entity.is_on)
self.assertEqual(entity.brightness, 10)
entity.turn_on(brightness=255)
self.assertTrue(entity.is_on)
self.assertEqual(entity.brightness, 255)
def test_one_light(self):
"""Test with 1 light."""
self.assertTrue(_setup_component(self.hass, 'light', {
'light': {'platform': 'rfxtrx',
'devices':
{'0b1100cd0213c7f210010f51': {
'name': 'Test'}}}}))
import RFXtrx as rfxtrxmod
rfxtrx_core.RFXOBJECT =\
rfxtrxmod.Core("", transport_protocol=rfxtrxmod.DummyTransport)
self.assertEqual(1, len(rfxtrx_core.RFX_DEVICES))
entity = rfxtrx_core.RFX_DEVICES['213c7f216']
self.assertEqual('Test', entity.name)
self.assertEqual('off', entity.state)
self.assertTrue(entity.assumed_state)
@ -112,15 +154,12 @@ class TestLightRfxtrx(unittest.TestCase):
'light': {'platform': 'rfxtrx',
'signal_repetitions': 3,
'devices':
{'123efab1': {
'name': 'Test',
'packetid': '0b1100cd0213c7f230010f71'},
'118cdea2': {
'name': 'Bath',
'packetid': '0b1100100118cdea02010f70'},
'213c7f216': {
'name': 'Living',
'packetid': '0b1100100118cdea02010f70'}}}}))
{'0b1100cd0213c7f230010f71': {
'name': 'Test'},
'0b1100100118cdea02010f70': {
'name': 'Bath'},
'0b1100101118cdea02010f70': {
'name': 'Living'}}}}))
self.assertEqual(3, len(rfxtrx_core.RFX_DEVICES))
device_num = 0
@ -155,7 +194,7 @@ class TestLightRfxtrx(unittest.TestCase):
rfxtrx_core.RECEIVED_EVT_SUBSCRIBERS[0](event)
entity = rfxtrx_core.RFX_DEVICES['0e611622']
self.assertEqual(1, len(rfxtrx_core.RFX_DEVICES))
self.assertEqual('<Entity 0e611622 : 0b11009e00e6116202020070: on>',
self.assertEqual('<Entity 0b11009e00e6116202020070: on>',
entity.__str__())
event = rfxtrx_core.get_rfx_object('0b11009e00e6116201010070')
@ -171,7 +210,7 @@ class TestLightRfxtrx(unittest.TestCase):
rfxtrx_core.RECEIVED_EVT_SUBSCRIBERS[0](event)
entity = rfxtrx_core.RFX_DEVICES['118cdea2']
self.assertEqual(2, len(rfxtrx_core.RFX_DEVICES))
self.assertEqual('<Entity 118cdea2 : 0b1100120118cdea02020070: on>',
self.assertEqual('<Entity 0b1100120118cdea02020070: on>',
entity.__str__())
# trying to add a sensor

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

@ -30,7 +30,7 @@ class TestSensorRfxtrx(unittest.TestCase):
{}}}))
self.assertEqual(0, len(rfxtrx_core.RFX_DEVICES))
def test_one_sensor(self):
def test_old_config_sensor(self):
"""Test with 1 sensor."""
self.assertTrue(_setup_component(self.hass, 'sensor', {
'sensor': {'platform': 'rfxtrx',
@ -41,8 +41,46 @@ class TestSensorRfxtrx(unittest.TestCase):
'data_type': 'Temperature'}}}}))
self.assertEqual(1, len(rfxtrx_core.RFX_DEVICES))
print(rfxtrx_core.RFX_DEVICES)
entity = rfxtrx_core.RFX_DEVICES['sensor_0502']
entity = rfxtrx_core.RFX_DEVICES['sensor_0502']['Temperature']
self.assertEqual('Test', entity.name)
self.assertEqual(TEMP_CELSIUS, entity.unit_of_measurement)
self.assertEqual(14.9, entity.state)
self.assertEqual({'Humidity status': 'normal', 'Temperature': 14.9,
'Rssi numeric': 6, 'Humidity': 34,
'Battery numeric': 9,
'Humidity status numeric': 2},
entity.device_state_attributes)
def test_one_sensor(self):
"""Test with 1 sensor."""
self.assertTrue(_setup_component(self.hass, 'sensor', {
'sensor': {'platform': 'rfxtrx',
'devices':
{'0a52080705020095220269': {
'name': 'Test',
'data_type': 'Temperature'}}}}))
self.assertEqual(1, len(rfxtrx_core.RFX_DEVICES))
entity = rfxtrx_core.RFX_DEVICES['sensor_0502']['Temperature']
self.assertEqual('Test', entity.name)
self.assertEqual(TEMP_CELSIUS, entity.unit_of_measurement)
self.assertEqual(14.9, entity.state)
self.assertEqual({'Humidity status': 'normal', 'Temperature': 14.9,
'Rssi numeric': 6, 'Humidity': 34,
'Battery numeric': 9,
'Humidity status numeric': 2},
entity.device_state_attributes)
def test_one_sensor_no_datatype(self):
"""Test with 1 sensor."""
self.assertTrue(_setup_component(self.hass, 'sensor', {
'sensor': {'platform': 'rfxtrx',
'devices':
{'0a52080705020095220269': {
'name': 'Test'}}}}))
self.assertEqual(1, len(rfxtrx_core.RFX_DEVICES))
entity = rfxtrx_core.RFX_DEVICES['sensor_0502']['Temperature']
self.assertEqual('Test', entity.name)
self.assertEqual(TEMP_CELSIUS, entity.unit_of_measurement)
self.assertEqual(14.9, entity.state)
@ -57,44 +95,44 @@ class TestSensorRfxtrx(unittest.TestCase):
self.assertTrue(_setup_component(self.hass, 'sensor', {
'sensor': {'platform': 'rfxtrx',
'devices':
{'sensor_0502': {
{'0a52080705020095220269': {
'name': 'Test',
'packetid': '0a52080705020095220269',
'data_type': 'Temperature'},
'sensor_0601': {
'name': 'Bath_Humidity',
'packetid': '0a520802060100ff0e0269',
'data_type': 'Humidity'},
'sensor_0601 2': {
'0a520802060100ff0e0269': {
'name': 'Bath',
'packetid': '0a520802060100ff0e0269'}}}}))
'data_type': ['Temperature', 'Humidity']
}}}}))
self.assertEqual(3, len(rfxtrx_core.RFX_DEVICES))
self.assertEqual(2, len(rfxtrx_core.RFX_DEVICES))
device_num = 0
for id in rfxtrx_core.RFX_DEVICES:
entity = rfxtrx_core.RFX_DEVICES[id]
if entity.name == 'Bath_Humidity':
if id == 'sensor_0601':
device_num = device_num + 1
self.assertEqual('%', entity.unit_of_measurement)
self.assertEqual(14, entity.state)
self.assertEqual(len(rfxtrx_core.RFX_DEVICES[id]), 2)
_entity_temp = rfxtrx_core.RFX_DEVICES[id]['Temperature']
_entity_hum = rfxtrx_core.RFX_DEVICES[id]['Humidity']
self.assertEqual('%', _entity_hum.unit_of_measurement)
self.assertEqual(14, _entity_hum.state)
self.assertEqual({'Battery numeric': 9, 'Temperature': 25.5,
'Humidity': 14, 'Humidity status': 'normal',
'Humidity status numeric': 2,
'Rssi numeric': 6},
entity.device_state_attributes)
self.assertEqual('Bath_Humidity', entity.__str__())
elif entity.name == 'Bath':
device_num = device_num + 1
self.assertEqual(TEMP_CELSIUS, entity.unit_of_measurement)
self.assertEqual(25.5, entity.state)
_entity_hum.device_state_attributes)
self.assertEqual('Bath', _entity_hum.__str__())
self.assertEqual(TEMP_CELSIUS,
_entity_temp.unit_of_measurement)
self.assertEqual(25.5, _entity_temp.state)
self.assertEqual({'Battery numeric': 9, 'Temperature': 25.5,
'Humidity': 14, 'Humidity status': 'normal',
'Humidity status numeric': 2,
'Rssi numeric': 6},
entity.device_state_attributes)
self.assertEqual('Bath', entity.__str__())
elif entity.name == 'Test':
_entity_temp.device_state_attributes)
self.assertEqual('Bath', _entity_temp.__str__())
elif id == 'sensor_0502':
device_num = device_num + 1
entity = rfxtrx_core.RFX_DEVICES[id]['Temperature']
self.assertEqual(TEMP_CELSIUS, entity.unit_of_measurement)
self.assertEqual(14.9, entity.state)
self.assertEqual({'Humidity status': 'normal',
@ -105,7 +143,7 @@ class TestSensorRfxtrx(unittest.TestCase):
entity.device_state_attributes)
self.assertEqual('Test', entity.__str__())
self.assertEqual(3, device_num)
self.assertEqual(2, device_num)
def test_discover_sensor(self):
"""Test with discovery of sensor."""
@ -118,7 +156,7 @@ class TestSensorRfxtrx(unittest.TestCase):
event.data = bytearray(b'\nR\x08\x01\x07\x01\x00\xb8\x1b\x02y')
rfxtrx_core.RECEIVED_EVT_SUBSCRIBERS[0](event)
entity = rfxtrx_core.RFX_DEVICES['sensor_0701']
entity = rfxtrx_core.RFX_DEVICES['sensor_0701']['Temperature']
self.assertEqual(1, len(rfxtrx_core.RFX_DEVICES))
self.assertEqual({'Humidity status': 'normal',
'Temperature': 18.4,
@ -126,7 +164,7 @@ class TestSensorRfxtrx(unittest.TestCase):
'Battery numeric': 9,
'Humidity status numeric': 2},
entity.device_state_attributes)
self.assertEqual('sensor_0701 : 0a520801070100b81b0279',
self.assertEqual('0a520801070100b81b0279',
entity.__str__())
rfxtrx_core.RECEIVED_EVT_SUBSCRIBERS[0](event)
@ -135,7 +173,7 @@ class TestSensorRfxtrx(unittest.TestCase):
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 = rfxtrx_core.RFX_DEVICES['sensor_0502']
entity = rfxtrx_core.RFX_DEVICES['sensor_0502']['Temperature']
self.assertEqual(2, len(rfxtrx_core.RFX_DEVICES))
self.assertEqual({'Humidity status': 'normal',
'Temperature': 14.9,
@ -143,13 +181,13 @@ class TestSensorRfxtrx(unittest.TestCase):
'Battery numeric': 9,
'Humidity status numeric': 2},
entity.device_state_attributes)
self.assertEqual('sensor_0502 : 0a52080405020095240279',
self.assertEqual('0a52080405020095240279',
entity.__str__())
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 = rfxtrx_core.RFX_DEVICES['sensor_0701']
entity = rfxtrx_core.RFX_DEVICES['sensor_0701']['Temperature']
self.assertEqual(2, len(rfxtrx_core.RFX_DEVICES))
self.assertEqual({'Humidity status': 'normal',
'Temperature': 17.9,
@ -157,7 +195,7 @@ class TestSensorRfxtrx(unittest.TestCase):
'Battery numeric': 9,
'Humidity status numeric': 2},
entity.device_state_attributes)
self.assertEqual('sensor_0701 : 0a520801070100b81b0279',
self.assertEqual('0a520801070100b81b0279',
entity.__str__())
# trying to add a switch
@ -197,45 +235,44 @@ class TestSensorRfxtrx(unittest.TestCase):
self.assertTrue(_setup_component(self.hass, 'sensor', {
'sensor': {'platform': 'rfxtrx',
'devices':
{'sensor_0502': {
{'0a52080705020095220269': {
'name': 'Test',
'packetid': '0a52080705020095220269',
'data_type': 'Temperature'},
'sensor_0601': {
'name': 'Bath_Humidity',
'packetid': '0a520802060100ff0e0269',
'data_type': 'Humidity'},
'sensor_0601 2': {
'0a520802060100ff0e0269': {
'name': 'Bath',
'packetid': '0a520802060100ff0e0269'}}}}))
self.assertEqual(3, len(rfxtrx_core.RFX_DEVICES))
'data_type': ['Temperature', 'Humidity']
}}}}))
self.assertEqual(2, len(rfxtrx_core.RFX_DEVICES))
device_num = 0
for id in rfxtrx_core.RFX_DEVICES:
entity = rfxtrx_core.RFX_DEVICES[id]
if entity.name == 'Bath_Humidity':
if id == 'sensor_0601':
device_num = device_num + 1
self.assertEqual('%', entity.unit_of_measurement)
self.assertEqual(14, entity.state)
self.assertEqual(len(rfxtrx_core.RFX_DEVICES[id]), 2)
_entity_temp = rfxtrx_core.RFX_DEVICES[id]['Temperature']
_entity_hum = rfxtrx_core.RFX_DEVICES[id]['Humidity']
self.assertEqual('%', _entity_hum.unit_of_measurement)
self.assertEqual(14, _entity_hum.state)
self.assertEqual({'Battery numeric': 9, 'Temperature': 25.5,
'Humidity': 14, 'Humidity status': 'normal',
'Humidity status numeric': 2,
'Rssi numeric': 6},
entity.device_state_attributes)
self.assertEqual('Bath_Humidity', entity.__str__())
elif entity.name == 'Bath':
device_num = device_num + 1
self.assertEqual(TEMP_CELSIUS, entity.unit_of_measurement)
self.assertEqual(25.5, entity.state)
_entity_hum.device_state_attributes)
self.assertEqual('Bath', _entity_hum.__str__())
self.assertEqual(TEMP_CELSIUS,
_entity_temp.unit_of_measurement)
self.assertEqual(25.5, _entity_temp.state)
self.assertEqual({'Battery numeric': 9, 'Temperature': 25.5,
'Humidity': 14, 'Humidity status': 'normal',
'Humidity status numeric': 2,
'Rssi numeric': 6},
entity.device_state_attributes)
self.assertEqual('Bath', entity.__str__())
elif entity.name == 'Test':
_entity_temp.device_state_attributes)
self.assertEqual('Bath', _entity_temp.__str__())
elif id == 'sensor_0502':
device_num = device_num + 1
entity = rfxtrx_core.RFX_DEVICES[id]['Temperature']
self.assertEqual(TEMP_CELSIUS, entity.unit_of_measurement)
self.assertEqual(14.9, entity.state)
self.assertEqual({'Humidity status': 'normal',
@ -246,7 +283,7 @@ class TestSensorRfxtrx(unittest.TestCase):
entity.device_state_attributes)
self.assertEqual('Test', entity.__str__())
self.assertEqual(3, device_num)
self.assertEqual(2, device_num)
event = rfxtrx_core.get_rfx_object('0a520802060101ff0f0269')
event.data = bytearray(b'\nR\x08\x01\x07\x01\x00\xb8\x1b\x02y')
@ -257,33 +294,36 @@ class TestSensorRfxtrx(unittest.TestCase):
event.data = bytearray(b'\nR\x08\x04\x05\x02\x00\x95$\x02y')
rfxtrx_core.RECEIVED_EVT_SUBSCRIBERS[0](event)
self.assertEqual(3, len(rfxtrx_core.RFX_DEVICES))
self.assertEqual(2, len(rfxtrx_core.RFX_DEVICES))
device_num = 0
for id in rfxtrx_core.RFX_DEVICES:
entity = rfxtrx_core.RFX_DEVICES[id]
if entity.name == 'Bath_Humidity':
if id == 'sensor_0601':
device_num = device_num + 1
self.assertEqual('%', entity.unit_of_measurement)
self.assertEqual(15, entity.state)
self.assertEqual(len(rfxtrx_core.RFX_DEVICES[id]), 2)
_entity_temp = rfxtrx_core.RFX_DEVICES[id]['Temperature']
_entity_hum = rfxtrx_core.RFX_DEVICES[id]['Humidity']
self.assertEqual('%', _entity_hum.unit_of_measurement)
self.assertEqual(15, _entity_hum.state)
self.assertEqual({'Battery numeric': 9, 'Temperature': 51.1,
'Humidity': 15, 'Humidity status': 'normal',
'Humidity status numeric': 2,
'Rssi numeric': 6},
entity.device_state_attributes)
self.assertEqual('Bath_Humidity', entity.__str__())
elif entity.name == 'Bath':
device_num = device_num + 1
self.assertEqual(TEMP_CELSIUS, entity.unit_of_measurement)
self.assertEqual(51.1, entity.state)
_entity_hum.device_state_attributes)
self.assertEqual('Bath', _entity_hum.__str__())
self.assertEqual(TEMP_CELSIUS,
_entity_temp.unit_of_measurement)
self.assertEqual(51.1, _entity_temp.state)
self.assertEqual({'Battery numeric': 9, 'Temperature': 51.1,
'Humidity': 15, 'Humidity status': 'normal',
'Humidity status numeric': 2,
'Rssi numeric': 6},
entity.device_state_attributes)
self.assertEqual('Bath', entity.__str__())
elif entity.name == 'Test':
_entity_temp.device_state_attributes)
self.assertEqual('Bath', _entity_temp.__str__())
elif id == 'sensor_0502':
device_num = device_num + 1
entity = rfxtrx_core.RFX_DEVICES[id]['Temperature']
self.assertEqual(TEMP_CELSIUS, entity.unit_of_measurement)
self.assertEqual(13.3, entity.state)
self.assertEqual({'Humidity status': 'normal',
@ -294,6 +334,5 @@ class TestSensorRfxtrx(unittest.TestCase):
entity.device_state_attributes)
self.assertEqual('Test', entity.__str__())
self.assertEqual(3, device_num)
self.assertEqual(3, len(rfxtrx_core.RFX_DEVICES))
self.assertEqual(2, device_num)
self.assertEqual(2, len(rfxtrx_core.RFX_DEVICES))

64
tests/components/switch/test_rfxtrx.py
View File

@ -29,9 +29,8 @@ class TestSwitchRfxtrx(unittest.TestCase):
'switch': {'platform': 'rfxtrx',
'automatic_add': True,
'devices':
{'213c7f216': {
{'0b1100cd0213c7f210010f51': {
'name': 'Test',
'packetid': '0b1100cd0213c7f210010f51',
rfxtrx_core.ATTR_FIREEVENT: True}
}}}))
@ -71,17 +70,6 @@ class TestSwitchRfxtrx(unittest.TestCase):
}}}))
def test_invalid_config4(self):
self.assertFalse(_setup_component(self.hass, 'switch', {
'switch': {'platform': 'rfxtrx',
'automatic_add': True,
'devices':
{'AA3c7f216': {
'name': 'Test',
'packetid': '0b1100cd0213c7f210010f51',
rfxtrx_core.ATTR_FIREEVENT: True}
}}}))
def test_invalid_config5(self):
"""Test configuration."""
self.assertFalse(_setup_component(self.hass, 'switch', {
'switch': {'platform': 'rfxtrx',
@ -100,7 +88,7 @@ class TestSwitchRfxtrx(unittest.TestCase):
{}}}))
self.assertEqual(0, len(rfxtrx_core.RFX_DEVICES))
def test_one_switch(self):
def test_old_config(self):
"""Test with 1 switch."""
self.assertTrue(_setup_component(self.hass, 'switch', {
'switch': {'platform': 'rfxtrx',
@ -114,7 +102,34 @@ class TestSwitchRfxtrx(unittest.TestCase):
rfxtrxmod.Core("", transport_protocol=rfxtrxmod.DummyTransport)
self.assertEqual(1, len(rfxtrx_core.RFX_DEVICES))
entity = rfxtrx_core.RFX_DEVICES['123efab1']
entity = rfxtrx_core.RFX_DEVICES['213c7f216']
self.assertEqual('Test', entity.name)
self.assertEqual('off', entity.state)
self.assertTrue(entity.assumed_state)
self.assertEqual(entity.signal_repetitions, 1)
self.assertFalse(entity.should_fire_event)
self.assertFalse(entity.should_poll)
self.assertFalse(entity.is_on)
entity.turn_on()
self.assertTrue(entity.is_on)
entity.turn_off()
self.assertFalse(entity.is_on)
def test_one_switch(self):
"""Test with 1 switch."""
self.assertTrue(_setup_component(self.hass, 'switch', {
'switch': {'platform': 'rfxtrx',
'devices':
{'0b1100cd0213c7f210010f51': {
'name': 'Test'}}}}))
import RFXtrx as rfxtrxmod
rfxtrx_core.RFXOBJECT =\
rfxtrxmod.Core("", transport_protocol=rfxtrxmod.DummyTransport)
self.assertEqual(1, len(rfxtrx_core.RFX_DEVICES))
entity = rfxtrx_core.RFX_DEVICES['213c7f216']
self.assertEqual('Test', entity.name)
self.assertEqual('off', entity.state)
self.assertTrue(entity.assumed_state)
@ -134,15 +149,12 @@ class TestSwitchRfxtrx(unittest.TestCase):
'switch': {'platform': 'rfxtrx',
'signal_repetitions': 3,
'devices':
{'123efab1': {
'name': 'Test',
'packetid': '0b1100cd0213c7f230010f71'},
'118cdea2': {
'name': 'Bath',
'packetid': '0b1100100118cdea02010f70'},
'213c7f216': {
'name': 'Living',
'packetid': '0b1100100118cdea02010f70'}}}}))
{'0b1100cd0213c7f230010f71': {
'name': 'Test'},
'0b1100100118cdea02010f70': {
'name': 'Bath'},
'0b1100101118cdea02010f70': {
'name': 'Living'}}}}))
self.assertEqual(3, len(rfxtrx_core.RFX_DEVICES))
device_num = 0
@ -178,7 +190,7 @@ class TestSwitchRfxtrx(unittest.TestCase):
rfxtrx_core.RECEIVED_EVT_SUBSCRIBERS[0](event)
entity = rfxtrx_core.RFX_DEVICES['118cdea2']
self.assertEqual(1, len(rfxtrx_core.RFX_DEVICES))
self.assertEqual('<Entity 118cdea2 : 0b1100100118cdea01010f70: on>',
self.assertEqual('<Entity 0b1100100118cdea01010f70: on>',
entity.__str__())
rfxtrx_core.RECEIVED_EVT_SUBSCRIBERS[0](event)
@ -191,7 +203,7 @@ class TestSwitchRfxtrx(unittest.TestCase):
rfxtrx_core.RECEIVED_EVT_SUBSCRIBERS[0](event)
entity = rfxtrx_core.RFX_DEVICES['118cdeb2']
self.assertEqual(2, len(rfxtrx_core.RFX_DEVICES))
self.assertEqual('<Entity 118cdeb2 : 0b1100120118cdea02000070: on>',
self.assertEqual('<Entity 0b1100120118cdea02000070: on>',
entity.__str__())
# Trying to add a sensor

7
tests/components/test_rfxtrx.py
View File

@ -37,10 +37,10 @@ class TestRFXTRX(unittest.TestCase):
'automatic_add': True,
'devices': {}}}))
while len(rfxtrx.RFX_DEVICES) < 2:
while len(rfxtrx.RFX_DEVICES) < 1:
time.sleep(0.1)
self.assertEqual(len(rfxtrx.RFXOBJECT.sensors()), 2)
self.assertEqual(len(rfxtrx.RFXOBJECT.sensors()), 1)
def test_valid_config(self):
"""Test configuration."""
@ -82,9 +82,8 @@ class TestRFXTRX(unittest.TestCase):
'switch': {'platform': 'rfxtrx',
'automatic_add': True,
'devices':
{'213c7f216': {
{'0b1100cd0213c7f210010f51': {
'name': 'Test',
'packetid': '0b1100cd0213c7f210010f51',
rfxtrx.ATTR_FIREEVENT: True}
}}}))