Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add monitored conditions for Unifi device_tracker (#15888) 

* Add support for monitored_conditions for attributes

* Update unifi tests

* Add list of available attrs
pull/15969/head
Charles Garwood 2018-08-13 15:18:25 -04:00 committed by Martin Hjelmare
parent ba2e43600e
commit 39647a15ae
2 changed files with 75 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

35
homeassistant/components/device_tracker/unifi.py
View File

@ -12,7 +12,7 @@ import homeassistant.helpers.config_validation as cv
from homeassistant.components.device_tracker import (
DOMAIN, PLATFORM_SCHEMA, DeviceScanner)
from homeassistant.const import CONF_HOST, CONF_USERNAME, CONF_PASSWORD
from homeassistant.const import CONF_VERIFY_SSL
from homeassistant.const import CONF_VERIFY_SSL, CONF_MONITORED_CONDITIONS
import homeassistant.util.dt as dt_util
REQUIREMENTS = ['pyunifi==2.13']
@ -31,6 +31,18 @@ DEFAULT_DETECTION_TIME = timedelta(seconds=300)
NOTIFICATION_ID = 'unifi_notification'
NOTIFICATION_TITLE = 'Unifi Device Tracker Setup'
AVAILABLE_ATTRS = [
'_id', '_is_guest_by_uap', '_last_seen_by_uap', '_uptime_by_uap',
'ap_mac', 'assoc_time', 'authorized', 'bssid', 'bytes-r', 'ccq',
'channel', 'essid', 'first_seen', 'hostname', 'idletime', 'ip',
'is_11r', 'is_guest', 'is_wired', 'last_seen', 'latest_assoc_time',
'mac', 'name', 'noise', 'noted', 'oui', 'powersave_enabled',
'qos_policy_applied', 'radio', 'radio_proto', 'rssi', 'rx_bytes',
'rx_bytes-r', 'rx_packets', 'rx_rate', 'signal', 'site_id',
'tx_bytes', 'tx_bytes-r', 'tx_packets', 'tx_power', 'tx_rate',
'uptime', 'user_id', 'usergroup_id', 'vlan'
]
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
vol.Optional(CONF_HOST, default=DEFAULT_HOST): cv.string,
vol.Optional(CONF_SITE_ID, default='default'): cv.string,
@ -41,6 +53,8 @@ PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
cv.boolean, cv.isfile),
vol.Optional(CONF_DETECTION_TIME, default=DEFAULT_DETECTION_TIME): vol.All(
cv.time_period, cv.positive_timedelta),
vol.Optional(CONF_MONITORED_CONDITIONS):
vol.All(cv.ensure_list, [vol.In(AVAILABLE_ATTRS)]),
vol.Optional(CONF_SSID_FILTER): vol.All(cv.ensure_list, [cv.string])
})
@ -56,6 +70,7 @@ def get_scanner(hass, config):
port = config[DOMAIN].get(CONF_PORT)
verify_ssl = config[DOMAIN].get(CONF_VERIFY_SSL)
detection_time = config[DOMAIN].get(CONF_DETECTION_TIME)
monitored_conditions = config[DOMAIN].get(CONF_MONITORED_CONDITIONS)
ssid_filter = config[DOMAIN].get(CONF_SSID_FILTER)
try:
@ -72,18 +87,20 @@ def get_scanner(hass, config):
notification_id=NOTIFICATION_ID)
return False
return UnifiScanner(ctrl, detection_time, ssid_filter)
return UnifiScanner(ctrl, detection_time, ssid_filter,
monitored_conditions)
class UnifiScanner(DeviceScanner):
"""Provide device_tracker support from Unifi WAP client data."""
def __init__(self, controller, detection_time: timedelta,
ssid_filter) -> None:
ssid_filter, monitored_conditions) -> None:
"""Initialize the scanner."""
self._detection_time = detection_time
self._controller = controller
self._ssid_filter = ssid_filter
self._monitored_conditions = monitored_conditions
self._update()
def _update(self):
@ -125,6 +142,14 @@ class UnifiScanner(DeviceScanner):
def get_extra_attributes(self, device):
"""Return the extra attributes of the device."""
if not self._monitored_conditions:
return {}
client = self._clients.get(device, {})
_LOGGER.debug("Device mac %s attributes %s", device, client)
return client
attributes = {}
for variable in self._monitored_conditions:
if variable in client:
attributes[variable] = client[variable]
_LOGGER.debug("Device mac %s attributes %s", device, attributes)
return attributes

56
tests/components/device_tracker/test_unifi.py
View File

@ -10,7 +10,8 @@ import voluptuous as vol
from homeassistant.components.device_tracker import DOMAIN, unifi as unifi
from homeassistant.const import (CONF_HOST, CONF_USERNAME, CONF_PASSWORD,
CONF_PLATFORM, CONF_VERIFY_SSL)
CONF_PLATFORM, CONF_VERIFY_SSL,
CONF_MONITORED_CONDITIONS)
DEFAULT_DETECTION_TIME = timedelta(seconds=300)
@ -54,7 +55,7 @@ def test_config_valid_verify_ssl(hass, mock_scanner, mock_ctrl):
assert mock_scanner.call_count == 1
assert mock_scanner.call_args == mock.call(mock_ctrl.return_value,
DEFAULT_DETECTION_TIME,
None)
None, None)
def test_config_minimal(hass, mock_scanner, mock_ctrl):
@ -76,7 +77,7 @@ def test_config_minimal(hass, mock_scanner, mock_ctrl):
assert mock_scanner.call_count == 1
assert mock_scanner.call_args == mock.call(mock_ctrl.return_value,
DEFAULT_DETECTION_TIME,
None)
None, None)
def test_config_full(hass, mock_scanner, mock_ctrl):
@ -88,6 +89,7 @@ def test_config_full(hass, mock_scanner, mock_ctrl):
CONF_PASSWORD: 'password',
CONF_HOST: 'myhost',
CONF_VERIFY_SSL: False,
CONF_MONITORED_CONDITIONS: ['essid', 'signal'],
'port': 123,
'site_id': 'abcdef01',
'detection_time': 300,
@ -101,9 +103,11 @@ def test_config_full(hass, mock_scanner, mock_ctrl):
version='v4', site_id='abcdef01', ssl_verify=False)
assert mock_scanner.call_count == 1
assert mock_scanner.call_args == mock.call(mock_ctrl.return_value,
DEFAULT_DETECTION_TIME,
None)
assert mock_scanner.call_args == mock.call(
mock_ctrl.return_value,
DEFAULT_DETECTION_TIME,
None,
config[DOMAIN][CONF_MONITORED_CONDITIONS])
def test_config_error():
@ -157,7 +161,7 @@ def test_scanner_update():
'last_seen': dt_util.as_timestamp(dt_util.utcnow())},
]
ctrl.get_clients.return_value = fake_clients
unifi.UnifiScanner(ctrl, DEFAULT_DETECTION_TIME, None)
unifi.UnifiScanner(ctrl, DEFAULT_DETECTION_TIME, None, None)
assert ctrl.get_clients.call_count == 1
assert ctrl.get_clients.call_args == mock.call()
@ -167,7 +171,7 @@ def test_scanner_update_error():
ctrl = mock.MagicMock()
ctrl.get_clients.side_effect = APIError(
'/', 500, 'foo', {}, None)
unifi.UnifiScanner(ctrl, DEFAULT_DETECTION_TIME, None)
unifi.UnifiScanner(ctrl, DEFAULT_DETECTION_TIME, None, None)
def test_scan_devices():
@ -180,7 +184,7 @@ def test_scan_devices():
'last_seen': dt_util.as_timestamp(dt_util.utcnow())},
]
ctrl.get_clients.return_value = fake_clients
scanner = unifi.UnifiScanner(ctrl, DEFAULT_DETECTION_TIME, None)
scanner = unifi.UnifiScanner(ctrl, DEFAULT_DETECTION_TIME, None, None)
assert set(scanner.scan_devices()) == set(['123', '234'])
@ -200,7 +204,8 @@ def test_scan_devices_filtered():
ssid_filter = ['foonet', 'barnet']
ctrl.get_clients.return_value = fake_clients
scanner = unifi.UnifiScanner(ctrl, DEFAULT_DETECTION_TIME, ssid_filter)
scanner = unifi.UnifiScanner(ctrl, DEFAULT_DETECTION_TIME, ssid_filter,
None)
assert set(scanner.scan_devices()) == set(['123', '234', '890'])
@ -221,8 +226,37 @@ def test_get_device_name():
'last_seen': '1504786810'},
]
ctrl.get_clients.return_value = fake_clients
scanner = unifi.UnifiScanner(ctrl, DEFAULT_DETECTION_TIME, None)
scanner = unifi.UnifiScanner(ctrl, DEFAULT_DETECTION_TIME, None, None)
assert scanner.get_device_name('123') == 'foobar'
assert scanner.get_device_name('234') == 'Nice Name'
assert scanner.get_device_name('456') is None
assert scanner.get_device_name('unknown') is None
def test_monitored_conditions():
"""Test the filtering of attributes."""
ctrl = mock.MagicMock()
fake_clients = [
{'mac': '123',
'hostname': 'foobar',
'essid': 'barnet',
'signal': -60,
'last_seen': dt_util.as_timestamp(dt_util.utcnow())},
{'mac': '234',
'name': 'Nice Name',
'essid': 'barnet',
'signal': -42,
'last_seen': dt_util.as_timestamp(dt_util.utcnow())},
{'mac': '456',
'hostname': 'wired',
'essid': 'barnet',
'last_seen': dt_util.as_timestamp(dt_util.utcnow())},
]
ctrl.get_clients.return_value = fake_clients
scanner = unifi.UnifiScanner(ctrl, DEFAULT_DETECTION_TIME, None,
['essid', 'signal'])
assert scanner.get_extra_attributes('123') == {'essid': 'barnet',
'signal': -60}
assert scanner.get_extra_attributes('234') == {'essid': 'barnet',
'signal': -42}
assert scanner.get_extra_attributes('456') == {'essid': 'barnet'}