BT Home Hub 5 device tracker support (#2250)

.coveragerc

@@ -95,6 +95,7 @@ omit =
     homeassistant/components/device_tracker/aruba.py
     homeassistant/components/device_tracker/asuswrt.py
     homeassistant/components/device_tracker/bluetooth_tracker.py
+    homeassistant/components/device_tracker/bt_home_hub_5.py
     homeassistant/components/device_tracker/ddwrt.py
     homeassistant/components/device_tracker/fritz.py
     homeassistant/components/device_tracker/icloud.py

homeassistant/components/device_tracker/bt_home_hub_5.py

@@ -0,0 +1,141 @@
+"""
+Support for BT Home Hub 5.
+
+For more details about this platform, please refer to the documentation at
+https://home-assistant.io/components/device_tracker.bt_home_hub_5/
+"""
+import logging
+import re
+import threading
+from datetime import timedelta
+import xml.etree.ElementTree as ET
+import json
+from urllib.parse import unquote
+
+import requests
+
+from homeassistant.helpers import validate_config
+from homeassistant.components.device_tracker import DOMAIN
+from homeassistant.const import CONF_HOST
+from homeassistant.util import Throttle
+
+# Return cached results if last scan was less then this time ago.
+MIN_TIME_BETWEEN_SCANS = timedelta(seconds=10)
+
+_LOGGER = logging.getLogger(__name__)
+
+_MAC_REGEX = re.compile(r'(([0-9A-Fa-f]{1,2}\:){5}[0-9A-Fa-f]{1,2})')
+
+
+# pylint: disable=unused-argument
+def get_scanner(hass, config):
+    """Return a BT Home Hub 5 scanner if successful."""
+    if not validate_config(config,
+                           {DOMAIN: [CONF_HOST]},
+                           _LOGGER):
+        return None
+    scanner = BTHomeHub5DeviceScanner(config[DOMAIN])
+
+    return scanner if scanner.success_init else None
+
+
+class BTHomeHub5DeviceScanner(object):
+    """This class queries a BT Home Hub 5."""
+
+    def __init__(self, config):
+        """Initialise the scanner."""
+        _LOGGER.info("Initialising BT Home Hub 5")
+        self.host = config.get(CONF_HOST, '192.168.1.254')
+
+        self.lock = threading.Lock()
+
+        self.last_results = {}
+
+        self.url = 'http://{}/nonAuth/home_status.xml'.format(self.host)
+
+        # Test the router is accessible
+        data = _get_homehub_data(self.url)
+        self.success_init = data is not None
+
+    def scan_devices(self):
+        """Scan for new devices and return a list with found device IDs."""
+        self._update_info()
+
+        return (device for device in self.last_results)
+
+    def get_device_name(self, device):
+        """Return the name of the given device or None if we don't know."""
+        with self.lock:
+            # If not initialised and not already scanned and not found.
+            if device not in self.last_results:
+                self._update_info()
+
+                if not self.last_results:
+                    return None
+
+            return self.last_results.get(device)
+
+    @Throttle(MIN_TIME_BETWEEN_SCANS)
+    def _update_info(self):
+        """Ensure the information from the BT Home Hub 5 is up to date.
+
+        Return boolean if scanning successful.
+        """
+        if not self.success_init:
+            return False
+
+        with self.lock:
+            _LOGGER.info("Scanning")
+
+            data = _get_homehub_data(self.url)
+
+            if not data:
+                _LOGGER.warning('Error scanning devices')
+                return False
+
+            self.last_results = data
+
+            return True
+
+
+def _get_homehub_data(url):
+    """Retrieve data from BT Home Hub 5 and return parsed result."""
+    try:
+        response = requests.get(url, timeout=5)
+    except requests.exceptions.Timeout:
+        _LOGGER.exception("Connection to the router timed out")
+        return
+    if response.status_code == 200:
+        return _parse_homehub_response(response.text)
+    else:
+        _LOGGER.error("Invalid response from Home Hub: %s", response)
+
+
+def _parse_homehub_response(data_str):
+    """Parse the BT Home Hub 5 data format."""
+    root = ET.fromstring(data_str)
+
+    dirty_json = root.find('known_device_list').get('value')
+
+    # Normalise the JavaScript data to JSON.
+    clean_json = unquote(dirty_json.replace('\'', '\"')
+                         .replace('{', '{\"')
+                         .replace(':\"', '\":\"')
+                         .replace('\",', '\",\"'))
+
+    known_devices = [x for x in json.loads(clean_json) if x]
+
+    devices = {}
+
+    for device in known_devices:
+        name = device.get('name')
+        mac = device.get('mac')
+
+        if _MAC_REGEX.match(mac) or ',' in mac:
+            for mac_addr in mac.split(','):
+                if _MAC_REGEX.match(mac_addr):
+                    devices[mac_addr] = name
+        else:
+            devices[mac] = name
+
+    return devices

tests/components/device_tracker/test_bt_home_hub_5.py

@@ -0,0 +1,53 @@
+"""The tests for the BT Home Hub 5 device tracker platform."""
+import unittest
+from unittest.mock import patch
+
+from homeassistant.components.device_tracker import bt_home_hub_5
+from homeassistant.const import CONF_HOST
+
+patch_file = 'homeassistant.components.device_tracker.bt_home_hub_5'
+
+
+def _get_homehub_data(url):
+    return '''
+    [
+        {
+            "mac": "AA:BB:CC:DD:EE:FF,
+            "hostname": "hostname",
+            "ip": "192.168.1.43",
+            "ipv6": "",
+            "name": "hostname",
+            "activity": "1",
+            "os": "Unknown",
+            "device": "Unknown",
+            "time_first_seen": "2016/06/05 11:14:45",
+            "time_last_active": "2016/06/06 11:33:08",
+            "dhcp_option": "39043T90430T9TGK0EKGE5KGE3K904390K45GK054",
+            "port": "wl0",
+            "ipv6_ll": "fe80::gd67:ghrr:fuud:4332",
+            "activity_ip": "1",
+            "activity_ipv6_ll": "0",
+            "activity_ipv6": "0",
+            "device_oui": "NA",
+            "device_serial": "NA",
+            "device_class": "NA"
+        }
+    ]
+    '''
+
+
+class TestBTHomeHub5DeviceTracker(unittest.TestCase):
+    """Test BT Home Hub 5 device tracker platform."""
+
+    @patch('{}._get_homehub_data'.format(patch_file), new=_get_homehub_data)
+    def test_config_minimal(self):
+        """Test the setup with minimal configuration."""
+
+        config = {
+            'device_tracker': {
+                CONF_HOST: 'foo'
+            }
+        }
+        result = bt_home_hub_5.get_scanner(None, config)
+
+        self.assertIsNotNone(result)