mirror of https://github.com/home-assistant/core
Move Broadlink services to component (#21465)
* Register services in broadlink domain * Add tests for broadlink services * Resolve review comments * One more review fix * Restore auth retry * Drop unused constants * Fix flake8 errors
This commit is contained in:
Joakim Plate
Martin Hjelmare
parent
f269135ae9
commit
0a3e11aa12
|
|
@ -1 +1,111 @@
|
|||
"""The broadlink component."""
|
||||
import asyncio
|
||||
from base64 import b64decode, b64encode
|
||||
import logging
|
||||
import re
|
||||
import socket
|
||||
|
||||
from datetime import timedelta
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant.const import CONF_HOST
|
||||
import homeassistant.helpers.config_validation as cv
|
||||
from homeassistant.util.dt import utcnow
|
||||
|
||||
from .const import CONF_PACKET, DOMAIN, SERVICE_LEARN, SERVICE_SEND
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
DEFAULT_RETRY = 3
|
||||
|
||||
|
||||
def ipv4_address(value):
|
||||
"""Validate an ipv4 address."""
|
||||
regex = re.compile(r'^\d+\.\d+\.\d+\.\d+$')
|
||||
if not regex.match(value):
|
||||
raise vol.Invalid('Invalid Ipv4 address, expected a.b.c.d')
|
||||
return value
|
||||
|
||||
|
||||
def data_packet(value):
|
||||
"""Decode a data packet given for broadlink."""
|
||||
return b64decode(cv.string(value))
|
||||
|
||||
|
||||
SERVICE_SEND_SCHEMA = vol.Schema({
|
||||
vol.Required(CONF_HOST): ipv4_address,
|
||||
vol.Required(CONF_PACKET): vol.All(cv.ensure_list, [data_packet])
|
||||
})
|
||||
|
||||
SERVICE_LEARN_SCHEMA = vol.Schema({
|
||||
vol.Required(CONF_HOST): ipv4_address,
|
||||
})
|
||||
|
||||
|
||||
def async_setup_service(hass, host, device):
|
||||
"""Register a device for given host for use in services."""
|
||||
hass.data.setdefault(DOMAIN, {})[host] = device
|
||||
|
||||
if not hass.services.has_service(DOMAIN, SERVICE_LEARN):
|
||||
|
||||
async def _learn_command(call):
|
||||
"""Learn a packet from remote."""
|
||||
device = hass.data[DOMAIN][call.data[CONF_HOST]]
|
||||
|
||||
try:
|
||||
auth = await hass.async_add_executor_job(device.auth)
|
||||
except socket.timeout:
|
||||
_LOGGER.error("Failed to connect to device, timeout")
|
||||
return
|
||||
if not auth:
|
||||
_LOGGER.error("Failed to connect to device")
|
||||
return
|
||||
|
||||
await hass.async_add_executor_job(device.enter_learning)
|
||||
|
||||
_LOGGER.info("Press the key you want Home Assistant to learn")
|
||||
start_time = utcnow()
|
||||
while (utcnow() - start_time) < timedelta(seconds=20):
|
||||
packet = await hass.async_add_executor_job(
|
||||
device.check_data)
|
||||
if packet:
|
||||
data = b64encode(packet).decode('utf8')
|
||||
log_msg = "Received packet is: {}".\
|
||||
format(data)
|
||||
_LOGGER.info(log_msg)
|
||||
hass.components.persistent_notification.async_create(
|
||||
log_msg, title='Broadlink switch')
|
||||
return
|
||||
await asyncio.sleep(1)
|
||||
_LOGGER.error("No signal was received")
|
||||
hass.components.persistent_notification.async_create(
|
||||
"No signal was received", title='Broadlink switch')
|
||||
|
||||
hass.services.async_register(
|
||||
DOMAIN, SERVICE_LEARN, _learn_command,
|
||||
schema=SERVICE_LEARN_SCHEMA)
|
||||
|
||||
if not hass.services.has_service(DOMAIN, SERVICE_SEND):
|
||||
|
||||
async def _send_packet(call):
|
||||
"""Send a packet."""
|
||||
device = hass.data[DOMAIN][call.data[CONF_HOST]]
|
||||
packets = call.data[CONF_PACKET]
|
||||
for packet in packets:
|
||||
for retry in range(DEFAULT_RETRY):
|
||||
try:
|
||||
await hass.async_add_executor_job(
|
||||
device.send_data, packet)
|
||||
break
|
||||
except (socket.timeout, ValueError):
|
||||
try:
|
||||
await hass.async_add_executor_job(
|
||||
device.auth)
|
||||
except socket.timeout:
|
||||
if retry == DEFAULT_RETRY-1:
|
||||
_LOGGER.error(
|
||||
"Failed to send packet to device")
|
||||
|
||||
hass.services.async_register(
|
||||
DOMAIN, SERVICE_SEND, _send_packet,
|
||||
schema=SERVICE_SEND_SCHEMA)
|
||||
|
|
|
|||
|
|
@ -0,0 +1,7 @@
|
|||
"""Constants for broadlink platform."""
|
||||
CONF_PACKET = 'packet'
|
||||
|
||||
DOMAIN = 'broadlink'
|
||||
|
||||
SERVICE_LEARN = 'learn'
|
||||
SERVICE_SEND = 'send'
|
||||
|
|
@ -0,0 +1,9 @@
|
|||
send:
|
||||
description: Send a raw packet to device.
|
||||
fields:
|
||||
host: {description: IP address of device to send packet via. This must be an already configured device., example: "192.168.0.1"}
|
||||
packet: {description: base64 encoded packet.}
|
||||
learn:
|
||||
description: Learn a IR or RF code from remote.
|
||||
fields:
|
||||
host: {description: IP address of device to send packet via. This must be an already configured device., example: "192.168.0.1"}
|
||||
|
|
@ -1,6 +1,5 @@
|
|||
"""Support for Broadlink RM devices."""
|
||||
import asyncio
|
||||
from base64 import b64decode, b64encode
|
||||
from base64 import b64decode
|
||||
import binascii
|
||||
from datetime import timedelta
|
||||
import logging
|
||||
|
|
@ -9,13 +8,14 @@ import socket
|
|||
import voluptuous as vol
|
||||
|
||||
from homeassistant.components.switch import (
|
||||
DOMAIN, PLATFORM_SCHEMA, SwitchDevice, ENTITY_ID_FORMAT)
|
||||
ENTITY_ID_FORMAT, PLATFORM_SCHEMA, SwitchDevice)
|
||||
from homeassistant.const import (
|
||||
CONF_COMMAND_OFF, CONF_COMMAND_ON, CONF_FRIENDLY_NAME, CONF_HOST, CONF_MAC,
|
||||
CONF_SWITCHES, CONF_TIMEOUT, CONF_TYPE)
|
||||
import homeassistant.helpers.config_validation as cv
|
||||
from homeassistant.util import Throttle, slugify
|
||||
from homeassistant.util.dt import utcnow
|
||||
|
||||
from . import async_setup_service
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
|
@ -23,9 +23,6 @@ TIME_BETWEEN_UPDATES = timedelta(seconds=5)
|
|||
|
||||
DEFAULT_NAME = 'Broadlink switch'
|
||||
DEFAULT_TIMEOUT = 10
|
||||
DEFAULT_RETRY = 3
|
||||
SERVICE_LEARN = 'broadlink_learn_command'
|
||||
SERVICE_SEND = 'broadlink_send_packet'
|
||||
CONF_SLOTS = 'slots'
|
||||
|
||||
RM_TYPES = ['rm', 'rm2', 'rm_mini', 'rm_pro_phicomm', 'rm2_home_plus',
|
||||
|
|
@ -73,57 +70,6 @@ def setup_platform(hass, config, add_entities, discovery_info=None):
|
|||
config.get(CONF_MAC).encode().replace(b':', b''))
|
||||
switch_type = config.get(CONF_TYPE)
|
||||
|
||||
async def _learn_command(call):
|
||||
"""Handle a learn command."""
|
||||
try:
|
||||
auth = await hass.async_add_job(broadlink_device.auth)
|
||||
except socket.timeout:
|
||||
_LOGGER.error("Failed to connect to device, timeout")
|
||||
return
|
||||
if not auth:
|
||||
_LOGGER.error("Failed to connect to device")
|
||||
return
|
||||
|
||||
await hass.async_add_job(broadlink_device.enter_learning)
|
||||
|
||||
_LOGGER.info("Press the key you want Home Assistant to learn")
|
||||
start_time = utcnow()
|
||||
while (utcnow() - start_time) < timedelta(seconds=20):
|
||||
packet = await hass.async_add_job(
|
||||
broadlink_device.check_data)
|
||||
if packet:
|
||||
log_msg = "Received packet is: {}".\
|
||||
format(b64encode(packet).decode('utf8'))
|
||||
_LOGGER.info(log_msg)
|
||||
hass.components.persistent_notification.async_create(
|
||||
log_msg, title='Broadlink switch')
|
||||
return
|
||||
await asyncio.sleep(1, loop=hass.loop)
|
||||
_LOGGER.error("Did not received any signal")
|
||||
hass.components.persistent_notification.async_create(
|
||||
"Did not received any signal", title='Broadlink switch')
|
||||
|
||||
async def _send_packet(call):
|
||||
"""Send a packet."""
|
||||
packets = call.data.get('packet', [])
|
||||
for packet in packets:
|
||||
for retry in range(DEFAULT_RETRY):
|
||||
try:
|
||||
extra = len(packet) % 4
|
||||
if extra > 0:
|
||||
packet = packet + ('=' * (4 - extra))
|
||||
payload = b64decode(packet)
|
||||
await hass.async_add_job(
|
||||
broadlink_device.send_data, payload)
|
||||
break
|
||||
except (socket.timeout, ValueError):
|
||||
try:
|
||||
await hass.async_add_job(
|
||||
broadlink_device.auth)
|
||||
except socket.timeout:
|
||||
if retry == DEFAULT_RETRY-1:
|
||||
_LOGGER.error("Failed to send packet to device")
|
||||
|
||||
def _get_mp1_slot_name(switch_friendly_name, slot):
|
||||
"""Get slot name."""
|
||||
if not slots['slot_{}'.format(slot)]:
|
||||
|
|
@ -132,13 +78,8 @@ def setup_platform(hass, config, add_entities, discovery_info=None):
|
|||
|
||||
if switch_type in RM_TYPES:
|
||||
broadlink_device = broadlink.rm((ip_addr, 80), mac_addr, None)
|
||||
hass.services.register(DOMAIN, SERVICE_LEARN + '_' +
|
||||
slugify(ip_addr.replace('.', '_')),
|
||||
_learn_command)
|
||||
hass.services.register(DOMAIN, SERVICE_SEND + '_' +
|
||||
slugify(ip_addr.replace('.', '_')),
|
||||
_send_packet,
|
||||
vol.Schema({'packet': cv.ensure_list}))
|
||||
hass.add_job(async_setup_service, hass, ip_addr, broadlink_device)
|
||||
|
||||
switches = []
|
||||
for object_id, device_config in devices.items():
|
||||
switches.append(
|
||||
|
|
|
|||
|
|
@ -0,0 +1 @@
|
|||
"""The tests for broadlink platforms."""
|
||||
|
|
@ -0,0 +1,117 @@
|
|||
"""The tests for the broadlink component."""
|
||||
from datetime import timedelta
|
||||
from base64 import b64decode
|
||||
from unittest.mock import MagicMock, patch, call
|
||||
|
||||
import pytest
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant.util.dt import utcnow
|
||||
from homeassistant.components.broadlink import async_setup_service
|
||||
from homeassistant.components.broadlink.const import (
|
||||
DOMAIN, SERVICE_LEARN, SERVICE_SEND)
|
||||
|
||||
DUMMY_IR_PACKET = ("JgBGAJKVETkRORA6ERQRFBEUERQRFBE5ETkQOhAVEBUQFREUEBUQ"
|
||||
"OhEUERQRORE5EBURFBA6EBUQOhE5EBUQFRA6EDoRFBEADQUAAA==")
|
||||
DUMMY_HOST = "192.168.0.2"
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def dummy_broadlink():
|
||||
"""Mock broadlink module so we don't have that dependency on tests."""
|
||||
broadlink = MagicMock()
|
||||
with patch.dict('sys.modules', {
|
||||
'broadlink': broadlink,
|
||||
}):
|
||||
yield broadlink
|
||||
|
||||
|
||||
async def test_send(hass):
|
||||
"""Test send service."""
|
||||
mock_device = MagicMock()
|
||||
mock_device.send_data.return_value = None
|
||||
|
||||
async_setup_service(hass, DUMMY_HOST, mock_device)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
await hass.services.async_call(DOMAIN, SERVICE_SEND, {
|
||||
"host": DUMMY_HOST,
|
||||
"packet": (DUMMY_IR_PACKET)
|
||||
})
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert mock_device.send_data.call_count == 1
|
||||
assert mock_device.send_data.call_args == call(
|
||||
b64decode(DUMMY_IR_PACKET))
|
||||
|
||||
|
||||
async def test_learn(hass):
|
||||
"""Test learn service."""
|
||||
mock_device = MagicMock()
|
||||
mock_device.enter_learning.return_value = None
|
||||
mock_device.check_data.return_value = b64decode(DUMMY_IR_PACKET)
|
||||
|
||||
with patch.object(hass.components.persistent_notification,
|
||||
'async_create') as mock_create:
|
||||
|
||||
async_setup_service(hass, DUMMY_HOST, mock_device)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
await hass.services.async_call(DOMAIN, SERVICE_LEARN, {
|
||||
"host": DUMMY_HOST,
|
||||
})
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert mock_device.enter_learning.call_count == 1
|
||||
assert mock_device.enter_learning.call_args == call()
|
||||
|
||||
assert mock_create.call_count == 1
|
||||
assert mock_create.call_args == call(
|
||||
"Received packet is: {}".format(DUMMY_IR_PACKET),
|
||||
title='Broadlink switch')
|
||||
|
||||
|
||||
async def test_learn_timeout(hass):
|
||||
"""Test learn service."""
|
||||
mock_device = MagicMock()
|
||||
mock_device.enter_learning.return_value = None
|
||||
mock_device.check_data.return_value = None
|
||||
|
||||
async_setup_service(hass, DUMMY_HOST, mock_device)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
now = utcnow()
|
||||
|
||||
with patch.object(hass.components.persistent_notification,
|
||||
'async_create') as mock_create, \
|
||||
patch('homeassistant.components.broadlink.utcnow') as mock_utcnow:
|
||||
|
||||
mock_utcnow.side_effect = [now, now + timedelta(20)]
|
||||
|
||||
await hass.services.async_call(DOMAIN, SERVICE_LEARN, {
|
||||
"host": DUMMY_HOST,
|
||||
})
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert mock_device.enter_learning.call_count == 1
|
||||
assert mock_device.enter_learning.call_args == call()
|
||||
|
||||
assert mock_create.call_count == 1
|
||||
assert mock_create.call_args == call(
|
||||
"No signal was received",
|
||||
title='Broadlink switch')
|
||||
|
||||
|
||||
async def test_ipv4():
|
||||
"""Test ipv4 parsing."""
|
||||
from homeassistant.components.broadlink import ipv4_address
|
||||
|
||||
schema = vol.Schema(ipv4_address)
|
||||
|
||||
for value in ('invalid', '1', '192', '192.168',
|
||||
'192.168.0', '192.168.0.A'):
|
||||
with pytest.raises(vol.MultipleInvalid):
|
||||
schema(value)
|
||||
|
||||
for value in ('192.168.0.1', '10.0.0.1'):
|
||||
schema(value)
|
||||
Loading…
Reference in New Issue