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

Added rfxtrx rollershutter (#2030)

Browse Source 
* Added rfxtrx rollershutter

* Updated mock command with real one

* Corrected test string
This commit is contained in:
Jacob Tomlinson 2016-06-02 08:39:58 +01:00 committed by Paulus Schoutsen
parent 26d3c3b0d6
commit ca7415e935
7 changed files with 327 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
homeassistant/components/rfxtrx.py
View File

@ -14,7 +14,7 @@ from homeassistant.const import EVENT_HOMEASSISTANT_STOP
from homeassistant.helpers.entity import Entity
from homeassistant.const import (ATTR_ENTITY_ID, TEMP_CELSIUS)
REQUIREMENTS = ['pyRFXtrx==0.6.5']
REQUIREMENTS = ['pyRFXtrx==0.7.0']
DOMAIN = "rfxtrx"
@ -310,6 +310,7 @@ class RfxtrxDevice(Entity):
self.update_ha_state()
def _send_command(self, command, brightness=0):
# pylint: disable=too-many-return-statements,too-many-branches
if not self._event:
return
@ -330,4 +331,16 @@ class RfxtrxDevice(Entity):
self._state = False
self._brightness = 0
elif command == "roll_up":
for _ in range(self.signal_repetitions):
self._event.device.send_open(RFXOBJECT.transport)
elif command == "roll_down":
for _ in range(self.signal_repetitions):
self._event.device.send_close(RFXOBJECT.transport)
elif command == "stop_roll":
for _ in range(self.signal_repetitions):
self._event.device.send_stop(RFXOBJECT.transport)
self.update_ha_state()

66
homeassistant/components/rollershutter/rfxtrx.py Normal file
View File

@ -0,0 +1,66 @@
"""
Support for RFXtrx roller shutter components.
For more details about this platform, please refer to the documentation
https://home-assistant.io/components/rollershutter.rfxtrx/
"""
import homeassistant.components.rfxtrx as rfxtrx
from homeassistant.components.rollershutter import RollershutterDevice
DEPENDENCIES = ['rfxtrx']
PLATFORM_SCHEMA = rfxtrx.DEFAULT_SCHEMA
def setup_platform(hass, config, add_devices_callback, discovery_info=None):
"""Setup the Demo roller shutters."""
import RFXtrx as rfxtrxmod
# Add rollershutter from config file
rollershutters = rfxtrx.get_devices_from_config(config,
RfxtrxRollershutter)
add_devices_callback(rollershutters)
def rollershutter_update(event):
"""Callback for roller shutter updates from the RFXtrx gateway."""
if not isinstance(event.device, rfxtrxmod.LightingDevice) or \
event.device.known_to_be_dimmable or \
not event.device.known_to_be_rollershutter:
return
new_device = rfxtrx.get_new_device(event, config, RfxtrxRollershutter)
if new_device:
add_devices_callback([new_device])
rfxtrx.apply_received_command(event)
# Subscribe to main rfxtrx events
if rollershutter_update not in rfxtrx.RECEIVED_EVT_SUBSCRIBERS:
rfxtrx.RECEIVED_EVT_SUBSCRIBERS.append(rollershutter_update)
class RfxtrxRollershutter(rfxtrx.RfxtrxDevice, RollershutterDevice):
"""Representation of an rfxtrx roller shutter."""
@property
def should_poll(self):
"""No polling available in rfxtrx roller shutter."""
return False
@property
def current_position(self):
"""No position available in rfxtrx roller shutter."""
return None
def move_up(self, **kwargs):
"""Move the roller shutter up."""
self._send_command("roll_up")
def move_down(self, **kwargs):
"""Move the roller shutter down."""
self._send_command("roll_down")
def stop(self, **kwargs):
"""Stop the roller shutter."""
self._send_command("stop_roll")

3
homeassistant/components/switch/rfxtrx.py
View File

@ -28,7 +28,8 @@ def setup_platform(hass, config, add_devices_callback, discovery_info=None):
def switch_update(event):
"""Callback for sensor updates from the RFXtrx gateway."""
if not isinstance(event.device, rfxtrxmod.LightingDevice) or \
event.device.known_to_be_dimmable:
event.device.known_to_be_dimmable or \
event.device.known_to_be_rollershutter:
return
new_device = rfxtrx.get_new_device(event, config, RfxtrxSwitch)

2
requirements_all.txt
View File

@ -219,7 +219,7 @@ pushetta==1.0.15
py-cpuinfo==0.2.3
# homeassistant.components.rfxtrx
pyRFXtrx==0.6.5
pyRFXtrx==0.7.0
# homeassistant.components.notify.xmpp
pyasn1-modules==0.0.8

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

@ -227,6 +227,13 @@ class TestLightRfxtrx(unittest.TestCase):
rfxtrx_core.RECEIVED_EVT_SUBSCRIBERS[0](event)
self.assertEqual(2, len(rfxtrx_core.RFX_DEVICES))
# Trying to add a rollershutter
event = rfxtrx_core.get_rfx_object('0a1400adf394ab020e0060')
event.data = bytearray([0x0A, 0x14, 0x00, 0xAD, 0xF3, 0x94,
0xAB, 0x02, 0x0E, 0x00, 0x60])
rfxtrx_core.RECEIVED_EVT_SUBSCRIBERS[0](event)
self.assertEqual(2, len(rfxtrx_core.RFX_DEVICES))
def test_discover_light_noautoadd(self):
"""Test with discover of light when auto add is False."""
self.assertTrue(_setup_component(self.hass, 'light', {
@ -265,6 +272,12 @@ class TestLightRfxtrx(unittest.TestCase):
event = rfxtrx_core.get_rfx_object('0b1100100118cdea02010f70')
event.data = bytearray([0x0b, 0x11, 0x00, 0x10, 0x01, 0x18,
0xcd, 0xea, 0x01, 0x01, 0x0f, 0x70])
rfxtrx_core.RECEIVED_EVT_SUBSCRIBERS[0](event)
self.assertEqual(0, len(rfxtrx_core.RFX_DEVICES))
# Trying to add a rollershutter
event = rfxtrx_core.get_rfx_object('0a1400adf394ab020e0060')
event.data = bytearray([0x0A, 0x14, 0x00, 0xAD, 0xF3, 0x94,
0xAB, 0x02, 0x0E, 0x00, 0x60])
rfxtrx_core.RECEIVED_EVT_SUBSCRIBERS[0](event)
self.assertEqual(0, len(rfxtrx_core.RFX_DEVICES))

216
tests/components/rollershutter/test_rfxtrx.py Normal file
View File

@ -0,0 +1,216 @@
"""The tests for the Rfxtrx roller shutter platform."""
import unittest
from homeassistant.bootstrap import _setup_component
from homeassistant.components import rfxtrx as rfxtrx_core
from tests.common import get_test_home_assistant
class TestRollershutterRfxtrx(unittest.TestCase):
"""Test the Rfxtrx roller shutter platform."""
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."""
rfxtrx_core.RECEIVED_EVT_SUBSCRIBERS = []
rfxtrx_core.RFX_DEVICES = {}
if rfxtrx_core.RFXOBJECT:
rfxtrx_core.RFXOBJECT.close_connection()
self.hass.stop()
def test_valid_config(self):
"""Test configuration."""
self.assertTrue(_setup_component(self.hass, 'rollershutter', {
'rollershutter': {'platform': 'rfxtrx',
'automatic_add': True,
'devices':
{'0b1100cd0213c7f210010f51': {
'name': 'Test',
rfxtrx_core.ATTR_FIREEVENT: True}
}}}))
def test_invalid_config1(self):
"""Test configuration."""
self.assertFalse(_setup_component(self.hass, 'rollershutter', {
'rollershutter': {'platform': 'rfxtrx',
'automatic_add': True,
'devices':
{'2FF7f216': {
'name': 'Test',
'packetid': '0b1100cd0213c7f210010f51',
'signal_repetitions': 3}
}}}))
def test_invalid_config2(self):
"""Test configuration."""
self.assertFalse(_setup_component(self.hass, 'rollershutter', {
'rollershutter': {'platform': 'rfxtrx',
'automatic_add': True,
'invalid_key': 'afda',
'devices':
{'213c7f216': {
'name': 'Test',
'packetid': '0b1100cd0213c7f210010f51',
rfxtrx_core.ATTR_FIREEVENT: True}
}}}))
def test_invalid_config3(self):
"""Test configuration."""
self.assertFalse(_setup_component(self.hass, 'rollershutter', {
'rollershutter': {'platform': 'rfxtrx',
'automatic_add': True,
'devices':
{'213c7f216': {
'name': 'Test',
'packetid': 'AA1100cd0213c7f210010f51',
rfxtrx_core.ATTR_FIREEVENT: True}
}}}))
def test_invalid_config4(self):
"""Test configuration."""
self.assertFalse(_setup_component(self.hass, 'rollershutter', {
'rollershutter': {'platform': 'rfxtrx',
'automatic_add': True,
'devices':
{'213c7f216': {
'name': 'Test',
rfxtrx_core.ATTR_FIREEVENT: True}
}}}))
def test_default_config(self):
"""Test with 0 roller shutter."""
self.assertTrue(_setup_component(self.hass, 'rollershutter', {
'rollershutter': {'platform': 'rfxtrx',
'devices': {}}}))
self.assertEqual(0, len(rfxtrx_core.RFX_DEVICES))
def test_one_rollershutter(self):
"""Test with 1 roller shutter."""
self.assertTrue(_setup_component(self.hass, 'rollershutter', {
'rollershutter': {'platform': 'rfxtrx',
'devices':
{'0b1400cd0213c7f210010f51': {
'name': 'Test'
}}}}))
import RFXtrx as rfxtrxmod
rfxtrx_core.RFXOBJECT =\
rfxtrxmod.Core("", transport_protocol=rfxtrxmod.DummyTransport)
self.assertEqual(1, len(rfxtrx_core.RFX_DEVICES))
for id in rfxtrx_core.RFX_DEVICES:
entity = rfxtrx_core.RFX_DEVICES[id]
self.assertEqual(entity.signal_repetitions, 1)
self.assertFalse(entity.should_fire_event)
self.assertFalse(entity.should_poll)
entity.move_up()
entity.move_down()
entity.stop()
def test_several_rollershutters(self):
"""Test with 3 roller shutters."""
self.assertTrue(_setup_component(self.hass, 'rollershutter', {
'rollershutter': {'platform': 'rfxtrx',
'signal_repetitions': 3,
'devices':
{'0b1100cd0213c7f230010f71': {
'name': 'Test'},
'0b1100100118cdea02010f70': {
'name': 'Bath'},
'0b1100101118cdea02010f70': {
'name': 'Living'}
}}}))
self.assertEqual(3, len(rfxtrx_core.RFX_DEVICES))
device_num = 0
for id in rfxtrx_core.RFX_DEVICES:
entity = rfxtrx_core.RFX_DEVICES[id]
self.assertEqual(entity.signal_repetitions, 3)
if entity.name == 'Living':
device_num = device_num + 1
elif entity.name == 'Bath':
device_num = device_num + 1
elif entity.name == 'Test':
device_num = device_num + 1
self.assertEqual(3, device_num)
def test_discover_rollershutter(self):
"""Test with discovery of roller shutters."""
self.assertTrue(_setup_component(self.hass, 'rollershutter', {
'rollershutter': {'platform': 'rfxtrx',
'automatic_add': True,
'devices': {}}}))
event = rfxtrx_core.get_rfx_object('0a140002f38cae010f0070')
event.data = bytearray([0x0A, 0x14, 0x00, 0x02, 0xF3, 0x8C,
0xAE, 0x01, 0x0F, 0x00, 0x70])
for evt_sub in rfxtrx_core.RECEIVED_EVT_SUBSCRIBERS:
evt_sub(event)
self.assertEqual(1, len(rfxtrx_core.RFX_DEVICES))
event = rfxtrx_core.get_rfx_object('0a1400adf394ab020e0060')
event.data = bytearray([0x0A, 0x14, 0x00, 0xAD, 0xF3, 0x94,
0xAB, 0x02, 0x0E, 0x00, 0x60])
for evt_sub in rfxtrx_core.RECEIVED_EVT_SUBSCRIBERS:
evt_sub(event)
self.assertEqual(2, len(rfxtrx_core.RFX_DEVICES))
# Trying to add a sensor
event = rfxtrx_core.get_rfx_object('0a52085e070100b31b0279')
event.data = bytearray(b'\nR\x08^\x07\x01\x00\xb3\x1b\x02y')
for evt_sub in rfxtrx_core.RECEIVED_EVT_SUBSCRIBERS:
evt_sub(event)
self.assertEqual(2, len(rfxtrx_core.RFX_DEVICES))
# Trying to add a light
event = rfxtrx_core.get_rfx_object('0b1100100118cdea02010f70')
event.data = bytearray([0x0b, 0x11, 0x11, 0x10, 0x01, 0x18,
0xcd, 0xea, 0x01, 0x02, 0x0f, 0x70])
for evt_sub in rfxtrx_core.RECEIVED_EVT_SUBSCRIBERS:
evt_sub(event)
self.assertEqual(2, len(rfxtrx_core.RFX_DEVICES))
def test_discover_rollershutter_noautoadd(self):
"""Test with discovery of roller shutter when auto add is False."""
self.assertTrue(_setup_component(self.hass, 'rollershutter', {
'rollershutter': {'platform': 'rfxtrx',
'automatic_add': False,
'devices': {}}}))
event = rfxtrx_core.get_rfx_object('0a1400adf394ab010d0060')
event.data = bytearray([0x0A, 0x14, 0x00, 0xAD, 0xF3, 0x94,
0xAB, 0x01, 0x0D, 0x00, 0x60])
for evt_sub in rfxtrx_core.RECEIVED_EVT_SUBSCRIBERS:
evt_sub(event)
self.assertEqual(0, len(rfxtrx_core.RFX_DEVICES))
event = rfxtrx_core.get_rfx_object('0a1400adf394ab020e0060')
event.data = bytearray([0x0A, 0x14, 0x00, 0xAD, 0xF3, 0x94,
0xAB, 0x02, 0x0E, 0x00, 0x60])
for evt_sub in rfxtrx_core.RECEIVED_EVT_SUBSCRIBERS:
evt_sub(event)
self.assertEqual(0, len(rfxtrx_core.RFX_DEVICES))
# Trying to add a sensor
event = rfxtrx_core.get_rfx_object('0a52085e070100b31b0279')
event.data = bytearray(b'\nR\x08^\x07\x01\x00\xb3\x1b\x02y')
for evt_sub in rfxtrx_core.RECEIVED_EVT_SUBSCRIBERS:
evt_sub(event)
self.assertEqual(0, len(rfxtrx_core.RFX_DEVICES))
# Trying to add a light
event = rfxtrx_core.get_rfx_object('0b1100100118cdea02010f70')
event.data = bytearray([0x0b, 0x11, 0x11, 0x10, 0x01,
0x18, 0xcd, 0xea, 0x01, 0x02, 0x0f, 0x70])
for evt_sub in rfxtrx_core.RECEIVED_EVT_SUBSCRIBERS:
evt_sub(event)
self.assertEqual(0, len(rfxtrx_core.RFX_DEVICES))

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

@ -219,6 +219,13 @@ class TestSwitchRfxtrx(unittest.TestCase):
rfxtrx_core.RECEIVED_EVT_SUBSCRIBERS[0](event)
self.assertEqual(2, len(rfxtrx_core.RFX_DEVICES))
# Trying to add a rollershutter
event = rfxtrx_core.get_rfx_object('0a1400adf394ab020e0060')
event.data = bytearray([0x0A, 0x14, 0x00, 0xAD, 0xF3, 0x94,
0xAB, 0x02, 0x0E, 0x00, 0x60])
rfxtrx_core.RECEIVED_EVT_SUBSCRIBERS[0](event)
self.assertEqual(2, len(rfxtrx_core.RFX_DEVICES))
def test_discover_switch_noautoadd(self):
"""Test with discovery of switch when auto add is False."""
self.assertTrue(_setup_component(self.hass, 'switch', {
@ -255,3 +262,10 @@ class TestSwitchRfxtrx(unittest.TestCase):
0x18, 0xcd, 0xea, 0x01, 0x02, 0x0f, 0x70])
rfxtrx_core.RECEIVED_EVT_SUBSCRIBERS[0](event)
self.assertEqual(0, len(rfxtrx_core.RFX_DEVICES))
# Trying to add a rollershutter
event = rfxtrx_core.get_rfx_object('0a1400adf394ab020e0060')
event.data = bytearray([0x0A, 0x14, 0x00, 0xAD, 0xF3, 0x94,
0xAB, 0x02, 0x0E, 0x00, 0x60])
rfxtrx_core.RECEIVED_EVT_SUBSCRIBERS[0](event)
self.assertEqual(0, len(rfxtrx_core.RFX_DEVICES))