Add Ubiquiti Unifi device tracker

Ubiquiti's Unifi WAP infrastructure has a central controller (like mfi and uvc)
that can be queried for client status. This adds a device_tracker module that
can report the state of any client connected to the controller.
This commit is contained in:
Dan Smith 2016-02-19 15:19:03 -08:00
parent f9385eb87a
commit 27f456ca70
3 changed files with 213 additions and 0 deletions

View file

@ -0,0 +1,79 @@
"""
homeassistant.components.device_tracker.unifi
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
Device tracker platform that supports scanning a Unifi WAP controller
"""
import logging
import urllib
from homeassistant.components.device_tracker import DOMAIN
from homeassistant.const import CONF_HOST, CONF_USERNAME, CONF_PASSWORD
from homeassistant.helpers import validate_config
# Unifi package doesn't list urllib3 as a requirement
REQUIREMENTS = ['urllib3', 'unifi==1.2.4']
_LOGGER = logging.getLogger(__name__)
CONF_PORT = 'port'
def get_scanner(hass, config):
""" Sets up unifi device_tracker """
from unifi.controller import Controller
if not validate_config(config, {DOMAIN: [CONF_USERNAME,
CONF_PASSWORD]},
_LOGGER):
_LOGGER.error('Invalid configuration')
return False
this_config = config[DOMAIN]
host = this_config.get(CONF_HOST, 'localhost')
username = this_config.get(CONF_USERNAME)
password = this_config.get(CONF_PASSWORD)
try:
port = int(this_config.get(CONF_PORT, 8443))
except ValueError:
_LOGGER.error('Invalid port (must be numeric like 8443)')
return False
try:
ctrl = Controller(host, username, password, port, 'v4')
except urllib.error.HTTPError as ex:
_LOGGER.error('Failed to connect to unifi: %s', ex)
return False
return UnifiScanner(ctrl)
class UnifiScanner(object):
"""Provide device_tracker support from Unifi WAP client data."""
def __init__(self, controller):
self._controller = controller
self._update()
def _update(self):
try:
clients = self._controller.get_clients()
except urllib.error.HTTPError as ex:
_LOGGER.error('Failed to scan clients: %s', ex)
clients = []
self._clients = {client['mac']: client for client in clients}
def scan_devices(self):
""" Scans for devices. """
self._update()
return self._clients.keys()
def get_device_name(self, mac):
""" Returns the name (if known) of the device.
If a name has been set in Unifi, then return that, else
return the hostname if it has been detected.
"""
client = self._clients.get(mac, {})
name = client.get('name') or client.get('hostname')
_LOGGER.debug('Device %s name %s', mac, name)
return name

View file

@ -252,6 +252,12 @@ tellive-py==0.5.2
# homeassistant.components.switch.transmission
transmissionrpc==0.11
# homeassistant.components.device_tracker.unifi
unifi==1.2.4
# homeassistant.components.device_tracker.unifi
urllib3
# homeassistant.components.camera.uvc
uvcclient==0.6

View file

@ -0,0 +1,128 @@
"""
homeassistant.components.device_tracker.unifi
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
Device tracker platform that supports scanning a Unifi WAP controller
"""
import unittest
from unittest import mock
import urllib
from homeassistant.components.device_tracker import unifi as unifi
from homeassistant.const import CONF_HOST, CONF_USERNAME, CONF_PASSWORD
from unifi import controller
class TestUnifiScanner(unittest.TestCase):
@mock.patch('homeassistant.components.device_tracker.unifi.UnifiScanner')
@mock.patch.object(controller, 'Controller')
def test_config_minimal(self, mock_ctrl, mock_scanner):
config = {
'device_tracker': {
CONF_USERNAME: 'foo',
CONF_PASSWORD: 'password',
}
}
result = unifi.get_scanner(None, config)
self.assertEqual(unifi.UnifiScanner.return_value, result)
mock_ctrl.assert_called_once_with('localhost', 'foo', 'password',
8443, 'v4')
mock_scanner.assert_called_once_with(mock_ctrl.return_value)
@mock.patch('homeassistant.components.device_tracker.unifi.UnifiScanner')
@mock.patch.object(controller, 'Controller')
def test_config_full(self, mock_ctrl, mock_scanner):
config = {
'device_tracker': {
CONF_USERNAME: 'foo',
CONF_PASSWORD: 'password',
CONF_HOST: 'myhost',
'port': 123,
}
}
result = unifi.get_scanner(None, config)
self.assertEqual(unifi.UnifiScanner.return_value, result)
mock_ctrl.assert_called_once_with('myhost', 'foo', 'password',
123, 'v4')
mock_scanner.assert_called_once_with(mock_ctrl.return_value)
@mock.patch('homeassistant.components.device_tracker.unifi.UnifiScanner')
@mock.patch.object(controller, 'Controller')
def test_config_error(self, mock_ctrl, mock_scanner):
config = {
'device_tracker': {
CONF_HOST: 'myhost',
'port': 123,
}
}
result = unifi.get_scanner(None, config)
self.assertFalse(result)
self.assertFalse(mock_ctrl.called)
@mock.patch('homeassistant.components.device_tracker.unifi.UnifiScanner')
@mock.patch.object(controller, 'Controller')
def test_config_badport(self, mock_ctrl, mock_scanner):
config = {
'device_tracker': {
CONF_USERNAME: 'foo',
CONF_PASSWORD: 'password',
CONF_HOST: 'myhost',
'port': 'foo',
}
}
result = unifi.get_scanner(None, config)
self.assertFalse(result)
self.assertFalse(mock_ctrl.called)
@mock.patch('homeassistant.components.device_tracker.unifi.UnifiScanner')
@mock.patch.object(controller, 'Controller')
def test_config_controller_failed(self, mock_ctrl, mock_scanner):
config = {
'device_tracker': {
CONF_USERNAME: 'foo',
CONF_PASSWORD: 'password',
}
}
mock_ctrl.side_effect = urllib.error.HTTPError(
'/', 500, 'foo', {}, None)
result = unifi.get_scanner(None, config)
self.assertFalse(result)
def test_scanner_update(self):
ctrl = mock.MagicMock()
fake_clients = [
{'mac': '123'},
{'mac': '234'},
]
ctrl.get_clients.return_value = fake_clients
unifi.UnifiScanner(ctrl)
ctrl.get_clients.assert_called_once_with()
def test_scanner_update_error(self):
ctrl = mock.MagicMock()
ctrl.get_clients.side_effect = urllib.error.HTTPError(
'/', 500, 'foo', {}, None)
unifi.UnifiScanner(ctrl)
def test_scan_devices(self):
ctrl = mock.MagicMock()
fake_clients = [
{'mac': '123'},
{'mac': '234'},
]
ctrl.get_clients.return_value = fake_clients
scanner = unifi.UnifiScanner(ctrl)
self.assertEqual(set(['123', '234']), set(scanner.scan_devices()))
def test_get_device_name(self):
ctrl = mock.MagicMock()
fake_clients = [
{'mac': '123', 'hostname': 'foobar'},
{'mac': '234', 'name': 'Nice Name'},
{'mac': '456'},
]
ctrl.get_clients.return_value = fake_clients
scanner = unifi.UnifiScanner(ctrl)
self.assertEqual('foobar', scanner.get_device_name('123'))
self.assertEqual('Nice Name', scanner.get_device_name('234'))
self.assertEqual(None, scanner.get_device_name('456'))
self.assertEqual(None, scanner.get_device_name('unknown'))