evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add RFXtrx Rfy venetian blinds tilt control (#44309)

Browse source 
* Add tilt control for RFXtrx Rfy venetian blinds

* Update Rfy cover test

* Update the required version of pyRFXtrx

* Update required pyRFXtrx version to 0.26.1

* Revert "Update required pyRFXtrx version to 0.26.1"

This reverts commit d54f1645d5.

* Revert "Update the required version of pyRFXtrx"

This reverts commit ac36d65326.

* Update required version of pyRFXtrx to 0.26.1

* @dzukero
Update required version of pyRFXtrx to 0.26.1

* Make requested changes from review

* Fix isort

* Remove set tilt position support

* Remove set tilt position support per review
This commit is contained in:
dzukero 2021-01-03 13:59:22 +02:00 • committed by GitHub
parent cc21639f00
commit 2ed7b90027
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
10 changed files with 375 additions and 8 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

25
homeassistant/components/rfxtrx/config_flow.py
View file

@ -40,6 +40,10 @@ from .const import (
CONF_REMOVE_DEVICE,
CONF_REPLACE_DEVICE,
CONF_SIGNAL_REPETITIONS,
CONF_VENETIAN_BLIND_MODE,
CONST_VENETIAN_BLIND_MODE_DEFAULT,
CONST_VENETIAN_BLIND_MODE_EU,
CONST_VENETIAN_BLIND_MODE_US,
DEVICE_PACKET_TYPE_LIGHTING4,
)
from .cover import supported as cover_supported
@ -218,6 +222,10 @@ class OptionsFlow(config_entries.OptionsFlow):
device[CONF_COMMAND_ON] = command_on
if command_off:
device[CONF_COMMAND_OFF] = command_off
if user_input.get(CONF_VENETIAN_BLIND_MODE):
device[CONF_VENETIAN_BLIND_MODE] = user_input[
CONF_VENETIAN_BLIND_MODE
]
self.update_config_data(
global_options=self._global_options, devices=devices
@ -282,6 +290,23 @@ class OptionsFlow(config_entries.OptionsFlow):
}
)
if isinstance(self._selected_device_object.device, rfxtrxmod.RfyDevice):
data_schema.update(
{
vol.Optional(
CONF_VENETIAN_BLIND_MODE,
default=device_data.get(
CONF_VENETIAN_BLIND_MODE, CONST_VENETIAN_BLIND_MODE_DEFAULT
),
): vol.In(
[
CONST_VENETIAN_BLIND_MODE_DEFAULT,
CONST_VENETIAN_BLIND_MODE_US,
CONST_VENETIAN_BLIND_MODE_EU,
]
),
}
)
devices = {
entry.id: entry.name_by_user if entry.name_by_user else entry.name
for entry in self._device_entries

5
homeassistant/components/rfxtrx/const.py
View file

@ -6,10 +6,15 @@ CONF_AUTOMATIC_ADD = "automatic_add"
CONF_SIGNAL_REPETITIONS = "signal_repetitions"
CONF_DEBUG = "debug"
CONF_OFF_DELAY = "off_delay"
CONF_VENETIAN_BLIND_MODE = "venetian_blind_mode"
CONF_REMOVE_DEVICE = "remove_device"
CONF_REPLACE_DEVICE = "replace_device"
CONST_VENETIAN_BLIND_MODE_DEFAULT = "Unknown"
CONST_VENETIAN_BLIND_MODE_EU = "EU"
CONST_VENETIAN_BLIND_MODE_US = "US"
COMMAND_ON_LIST = [
"On",
"Up",

84
homeassistant/components/rfxtrx/cover.py
View file

@ -1,7 +1,15 @@
"""Support for RFXtrx covers."""
import logging
from homeassistant.components.cover import CoverEntity
from homeassistant.components.cover import (
SUPPORT_CLOSE,
SUPPORT_CLOSE_TILT,
SUPPORT_OPEN,
SUPPORT_OPEN_TILT,
SUPPORT_STOP,
SUPPORT_STOP_TILT,
CoverEntity,
)
from homeassistant.const import CONF_DEVICES, STATE_OPEN
from homeassistant.core import callback
@ -14,7 +22,13 @@ from . import (
get_device_id,
get_rfx_object,
)
from .const import COMMAND_OFF_LIST, COMMAND_ON_LIST
from .const import (
COMMAND_OFF_LIST,
COMMAND_ON_LIST,
CONF_VENETIAN_BLIND_MODE,
CONST_VENETIAN_BLIND_MODE_EU,
CONST_VENETIAN_BLIND_MODE_US,
)
_LOGGER = logging.getLogger(__name__)
@ -50,7 +64,10 @@ async def async_setup_entry(
device_ids.add(device_id)
entity = RfxtrxCover(
event.device, device_id, entity_info[CONF_SIGNAL_REPETITIONS]
event.device,
device_id,
signal_repetitions=entity_info[CONF_SIGNAL_REPETITIONS],
venetian_blind_mode=entity_info.get(CONF_VENETIAN_BLIND_MODE),
)
entities.append(entity)
@ -86,6 +103,18 @@ async def async_setup_entry(
class RfxtrxCover(RfxtrxCommandEntity, CoverEntity):
"""Representation of a RFXtrx cover."""
def __init__(
self,
device,
device_id,
signal_repetitions,
event=None,
venetian_blind_mode=None,
):
"""Initialize the RFXtrx cover device."""
super().__init__(device, device_id, signal_repetitions, event)
self._venetian_blind_mode = venetian_blind_mode
async def async_added_to_hass(self):
"""Restore device state."""
await super().async_added_to_hass()
@ -95,6 +124,21 @@ class RfxtrxCover(RfxtrxCommandEntity, CoverEntity):
if old_state is not None:
self._state = old_state.state == STATE_OPEN
@property
def supported_features(self):
"""Flag supported features."""
supported_features = SUPPORT_OPEN | SUPPORT_CLOSE | SUPPORT_STOP
if self._venetian_blind_mode in (
CONST_VENETIAN_BLIND_MODE_US,
CONST_VENETIAN_BLIND_MODE_EU,
):
supported_features |= (
SUPPORT_OPEN_TILT | SUPPORT_CLOSE_TILT | SUPPORT_STOP_TILT
)
return supported_features
@property
def is_closed(self):
"""Return if the cover is closed."""
@ -102,13 +146,23 @@ class RfxtrxCover(RfxtrxCommandEntity, CoverEntity):
async def async_open_cover(self, **kwargs):
"""Move the cover up."""
await self._async_send(self._device.send_open)
if self._venetian_blind_mode == CONST_VENETIAN_BLIND_MODE_US:
await self._async_send(self._device.send_up05sec)
elif self._venetian_blind_mode == CONST_VENETIAN_BLIND_MODE_EU:
await self._async_send(self._device.send_up2sec)
else:
await self._async_send(self._device.send_open)
self._state = True
self.async_write_ha_state()
async def async_close_cover(self, **kwargs):
"""Move the cover down."""
await self._async_send(self._device.send_close)
if self._venetian_blind_mode == CONST_VENETIAN_BLIND_MODE_US:
await self._async_send(self._device.send_down05sec)
elif self._venetian_blind_mode == CONST_VENETIAN_BLIND_MODE_EU:
await self._async_send(self._device.send_down2sec)
else:
await self._async_send(self._device.send_close)
self._state = False
self.async_write_ha_state()
@ -118,6 +172,26 @@ class RfxtrxCover(RfxtrxCommandEntity, CoverEntity):
self._state = True
self.async_write_ha_state()
async def async_open_cover_tilt(self, **kwargs):
"""Tilt the cover up."""
if self._venetian_blind_mode == CONST_VENETIAN_BLIND_MODE_US:
await self._async_send(self._device.send_up2sec)
elif self._venetian_blind_mode == CONST_VENETIAN_BLIND_MODE_EU:
await self._async_send(self._device.send_up05sec)
async def async_close_cover_tilt(self, **kwargs):
"""Tilt the cover down."""
if self._venetian_blind_mode == CONST_VENETIAN_BLIND_MODE_US:
await self._async_send(self._device.send_down2sec)
elif self._venetian_blind_mode == CONST_VENETIAN_BLIND_MODE_EU:
await self._async_send(self._device.send_down05sec)
async def async_stop_cover_tilt(self, **kwargs):
"""Stop the cover tilt."""
await self._async_send(self._device.send_stop)
self._state = True
self.async_write_ha_state()
def _apply_event(self, event):
"""Apply command from rfxtrx."""
super()._apply_event(event)

2
homeassistant/components/rfxtrx/manifest.json
View file

@ -2,7 +2,7 @@
"domain": "rfxtrx",
"name": "RFXCOM RFXtrx",
"documentation": "https://www.home-assistant.io/integrations/rfxtrx",
"requirements": ["pyRFXtrx==0.26"],
"requirements": ["pyRFXtrx==0.26.1"],
"codeowners": ["@danielhiversen", "@elupus", "@RobBie1221"],
"config_flow": true
}

1
homeassistant/components/rfxtrx/strings.json
View file

@ -56,6 +56,7 @@
"command_on": "Data bits value for command on",
"command_off": "Data bits value for command off",
"signal_repetitions": "Number of signal repetitions",
"venetian_blind_mode": "Venetian blind mode",
"replace_device": "Select device to replace"
},
"title": "Configure device options"

1
homeassistant/components/rfxtrx/translations/en.json
View file

@ -64,6 +64,7 @@
"off_delay": "Off delay",
"off_delay_enabled": "Enable off delay",
"replace_device": "Select device to replace",
"venetian_blind_mode": "Venetian blind mode (tilt by: US - long press, EU - short press)",
"signal_repetitions": "Number of signal repetitions"
},
"title": "Configure device options"

2
requirements_all.txt
View file

@ -1229,7 +1229,7 @@ pyHS100==0.3.5.2
pyMetno==0.8.1
# homeassistant.components.rfxtrx
pyRFXtrx==0.26
pyRFXtrx==0.26.1
# homeassistant.components.switchmate
# pySwitchmate==0.4.6

2
requirements_test_all.txt
View file

@ -619,7 +619,7 @@ pyHS100==0.3.5.2
pyMetno==0.8.1
# homeassistant.components.rfxtrx
pyRFXtrx==0.26
pyRFXtrx==0.26.1
# homeassistant.components.tibber
pyTibber==0.16.0

84
tests/components/rfxtrx/test_config_flow.py
View file

@ -1103,6 +1103,90 @@ async def test_options_add_and_configure_device(hass):
assert "delay_off" not in entry.data["devices"]["0913000022670e013970"]
async def test_options_configure_rfy_cover_device(hass):
"""Test we can configure the venetion blind mode of an Rfy cover."""
await setup.async_setup_component(hass, "persistent_notification", {})
entry = MockConfigEntry(
domain=DOMAIN,
data={
"host": None,
"port": None,
"device": "/dev/tty123",
"automatic_add": False,
"devices": {},
},
unique_id=DOMAIN,
)
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
result = await hass.config_entries.options.async_init(entry.entry_id)
assert result["type"] == "form"
assert result["step_id"] == "prompt_options"
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={
"automatic_add": True,
"event_code": "071a000001020301",
},
)
assert result["type"] == "form"
assert result["step_id"] == "set_device_options"
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={
"fire_event": False,
"venetian_blind_mode": "EU",
},
)
await hass.async_block_till_done()
assert entry.data["devices"]["071a000001020301"]["venetian_blind_mode"] == "EU"
device_registry = await async_get_device_registry(hass)
device_entries = async_entries_for_config_entry(device_registry, entry.entry_id)
assert device_entries[0].id
result = await hass.config_entries.options.async_init(entry.entry_id)
assert result["type"] == "form"
assert result["step_id"] == "prompt_options"
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={
"automatic_add": False,
"device": device_entries[0].id,
},
)
assert result["type"] == "form"
assert result["step_id"] == "set_device_options"
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={
"fire_event": False,
"venetian_blind_mode": "EU",
},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
await hass.async_block_till_done()
assert entry.data["devices"]["071a000001020301"]["venetian_blind_mode"] == "EU"
def test_get_serial_by_id_no_dir():
"""Test serial by id conversion if there's no /dev/serial/by-id."""
p1 = patch("os.path.isdir", MagicMock(return_value=False))

177
tests/components/rfxtrx/test_cover.py
View file

@ -140,3 +140,180 @@ async def test_duplicate_cover(hass, rfxtrx):
assert state
assert state.state == "closed"
assert state.attributes.get("friendly_name") == "LightwaveRF, Siemens 0213c7:242"
async def test_rfy_cover(hass, rfxtrx):
"""Test Rfy venetian blind covers."""
entry_data = create_rfx_test_cfg(
devices={
"071a000001020301": {
"signal_repetitions": 1,
"venetian_blind_mode": "Unknown",
},
"071a000001020302": {"signal_repetitions": 1, "venetian_blind_mode": "US"},
"071a000001020303": {"signal_repetitions": 1, "venetian_blind_mode": "EU"},
}
)
mock_entry = MockConfigEntry(domain="rfxtrx", unique_id=DOMAIN, data=entry_data)
mock_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_entry.entry_id)
await hass.async_block_till_done()
# Test a blind with no venetian mode setting
state = hass.states.get("cover.rfy_010203_1")
assert state
await hass.services.async_call(
"cover",
"stop_cover",
{"entity_id": "cover.rfy_010203_1"},
blocking=True,
)
await hass.services.async_call(
"cover",
"open_cover",
{"entity_id": "cover.rfy_010203_1"},
blocking=True,
)
await hass.services.async_call(
"cover",
"close_cover",
{"entity_id": "cover.rfy_010203_1"},
blocking=True,
)
await hass.services.async_call(
"cover",
"open_cover_tilt",
{"entity_id": "cover.rfy_010203_1"},
blocking=True,
)
await hass.services.async_call(
"cover",
"close_cover_tilt",
{"entity_id": "cover.rfy_010203_1"},
blocking=True,
)
assert rfxtrx.transport.send.mock_calls == [
call(bytearray(b"\x08\x1a\x00\x00\x01\x02\x03\x01\x00")),
call(bytearray(b"\x08\x1a\x00\x01\x01\x02\x03\x01\x01")),
call(bytearray(b"\x08\x1a\x00\x02\x01\x02\x03\x01\x03")),
]
# Test a blind with venetian mode set to US
state = hass.states.get("cover.rfy_010203_2")
assert state
rfxtrx.transport.send.mock_calls = []
await hass.services.async_call(
"cover",
"stop_cover",
{"entity_id": "cover.rfy_010203_2"},
blocking=True,
)
await hass.services.async_call(
"cover",
"open_cover",
{"entity_id": "cover.rfy_010203_2"},
blocking=True,
)
await hass.services.async_call(
"cover",
"close_cover",
{"entity_id": "cover.rfy_010203_2"},
blocking=True,
)
await hass.services.async_call(
"cover",
"open_cover_tilt",
{"entity_id": "cover.rfy_010203_2"},
blocking=True,
)
await hass.services.async_call(
"cover",
"close_cover_tilt",
{"entity_id": "cover.rfy_010203_2"},
blocking=True,
)
await hass.services.async_call(
"cover",
"stop_cover_tilt",
{"entity_id": "cover.rfy_010203_2"},
blocking=True,
)
assert rfxtrx.transport.send.mock_calls == [
call(bytearray(b"\x08\x1a\x00\x00\x01\x02\x03\x02\x00")),
call(bytearray(b"\x08\x1a\x00\x01\x01\x02\x03\x02\x0F")),
call(bytearray(b"\x08\x1a\x00\x02\x01\x02\x03\x02\x10")),
call(bytearray(b"\x08\x1a\x00\x03\x01\x02\x03\x02\x11")),
call(bytearray(b"\x08\x1a\x00\x04\x01\x02\x03\x02\x12")),
call(bytearray(b"\x08\x1a\x00\x00\x01\x02\x03\x02\x00")),
]
# Test a blind with venetian mode set to EU
state = hass.states.get("cover.rfy_010203_3")
assert state
rfxtrx.transport.send.mock_calls = []
await hass.services.async_call(
"cover",
"stop_cover",
{"entity_id": "cover.rfy_010203_3"},
blocking=True,
)
await hass.services.async_call(
"cover",
"open_cover",
{"entity_id": "cover.rfy_010203_3"},
blocking=True,
)
await hass.services.async_call(
"cover",
"close_cover",
{"entity_id": "cover.rfy_010203_3"},
blocking=True,
)
await hass.services.async_call(
"cover",
"open_cover_tilt",
{"entity_id": "cover.rfy_010203_3"},
blocking=True,
)
await hass.services.async_call(
"cover",
"close_cover_tilt",
{"entity_id": "cover.rfy_010203_3"},
blocking=True,
)
await hass.services.async_call(
"cover",
"stop_cover_tilt",
{"entity_id": "cover.rfy_010203_3"},
blocking=True,
)
assert rfxtrx.transport.send.mock_calls == [
call(bytearray(b"\x08\x1a\x00\x00\x01\x02\x03\x03\x00")),
call(bytearray(b"\x08\x1a\x00\x01\x01\x02\x03\x03\x11")),
call(bytearray(b"\x08\x1a\x00\x02\x01\x02\x03\x03\x12")),
call(bytearray(b"\x08\x1a\x00\x03\x01\x02\x03\x03\x0F")),
call(bytearray(b"\x08\x1a\x00\x04\x01\x02\x03\x03\x10")),
call(bytearray(b"\x08\x1a\x00\x00\x01\x02\x03\x03\x00")),
]