Move Broadlink services to component

homeassistant/components/broadlink/__init__.py

@@ -1 +1,86 @@
 """The broadlink component."""
+import asyncio
+from base64 import b64decode, b64encode
+import logging
+import re
+
+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__)
+
+
+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]]
+            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:
+                await hass.async_add_executor_job(
+                    device.send_data, packet)
+        hass.services.async_register(
+            DOMAIN, SERVICE_SEND, _send_packet,
+            schema=SERVICE_SEND_SCHEMA)

homeassistant/components/broadlink/const.py

@@ -0,0 +1,7 @@
+"""Constants for broadlink platform."""
+CONF_PACKET = 'packet'
+
+DOMAIN = 'broadlink'
+
+SERVICE_LEARN = 'learn'
+SERVICE_SEND = 'send'

homeassistant/components/broadlink/services.yaml

@@ -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"}

homeassistant/components/broadlink/switch.py

@@ -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 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
 
 REQUIREMENTS = ['broadlink==0.9.0']
 
@@ -75,57 +75,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)]:
@@ -134,13 +83,8 @@ def _get_mp1_slot_name(switch_friendly_name, slot):
 
     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(

tests/components/broadlink/__init__.py

@@ -0,0 +1 @@
+"""The tests for broadlink platforms."""

tests/components/broadlink/test_init.py

@@ -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)