Initial support for Google Wifi/OnHub (#8485)
* Initial support for Google Wifi/OnHub * Moved state logic to update function of API class - Throttle added to update - State logic implementation is cleaner - Modified tests to work with the new throttle on updatepull/8545/head
Kevin Fronczak
Pascal Vizeli
parent
d54a634f11
commit
1a86fa5a02
|
|
@ -0,0 +1,201 @@
|
|||
"""
|
||||
Support for retreiving status info from Google Wifi/OnHub routers.
|
||||
|
||||
For more details about this platform, please refer to the documentation at
|
||||
https://home-assistant.io/components/sensor.google_wifi/
|
||||
"""
|
||||
import logging
|
||||
from datetime import timedelta
|
||||
|
||||
import voluptuous as vol
|
||||
import requests
|
||||
|
||||
import homeassistant.util.dt as dt
|
||||
import homeassistant.helpers.config_validation as cv
|
||||
from homeassistant.helpers.entity import Entity
|
||||
from homeassistant.components.sensor import PLATFORM_SCHEMA
|
||||
from homeassistant.util import Throttle
|
||||
from homeassistant.const import (
|
||||
CONF_NAME, CONF_HOST, CONF_MONITORED_CONDITIONS,
|
||||
STATE_UNKNOWN)
|
||||
|
||||
MIN_TIME_BETWEEN_UPDATES = timedelta(seconds=1)
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
ENDPOINT = '/api/v1/status'
|
||||
|
||||
ATTR_CURRENT_VERSION = 'current_version'
|
||||
ATTR_NEW_VERSION = 'new_version'
|
||||
ATTR_UPTIME = 'uptime'
|
||||
ATTR_LAST_RESTART = 'last_restart'
|
||||
ATTR_LOCAL_IP = 'local_ip'
|
||||
ATTR_STATUS = 'status'
|
||||
|
||||
DEFAULT_NAME = 'google_wifi'
|
||||
DEFAULT_HOST = 'testwifi.here'
|
||||
|
||||
MONITORED_CONDITIONS = {
|
||||
ATTR_CURRENT_VERSION: [
|
||||
'Current Version',
|
||||
None,
|
||||
'mdi:checkbox-marked-circle-outline'
|
||||
],
|
||||
ATTR_NEW_VERSION: [
|
||||
'New Version',
|
||||
None,
|
||||
'mdi:update'
|
||||
],
|
||||
ATTR_UPTIME: [
|
||||
'Uptime',
|
||||
'days',
|
||||
'mdi:timelapse'
|
||||
],
|
||||
ATTR_LAST_RESTART: [
|
||||
'Last Network Restart',
|
||||
None,
|
||||
'mdi:restart'
|
||||
],
|
||||
ATTR_LOCAL_IP: [
|
||||
'Local IP Address',
|
||||
None,
|
||||
'mdi:access-point-network'
|
||||
],
|
||||
ATTR_STATUS: [
|
||||
'Status',
|
||||
None,
|
||||
'mdi:google'
|
||||
]
|
||||
}
|
||||
|
||||
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
|
||||
vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
|
||||
vol.Optional(CONF_HOST, default=DEFAULT_HOST): cv.string,
|
||||
vol.Optional(CONF_MONITORED_CONDITIONS, default=MONITORED_CONDITIONS):
|
||||
vol.All(cv.ensure_list, [vol.In(MONITORED_CONDITIONS)]),
|
||||
})
|
||||
|
||||
|
||||
def setup_platform(hass, config, add_devices, discovery_info=None):
|
||||
"""Set up the Google Wifi sensor."""
|
||||
name = config.get(CONF_NAME)
|
||||
host = config.get(CONF_HOST)
|
||||
|
||||
api = GoogleWifiAPI(host)
|
||||
|
||||
sensors = [GoogleWifiSensor(hass, api, name, condition)
|
||||
for condition in config[CONF_MONITORED_CONDITIONS]]
|
||||
|
||||
add_devices(sensors, True)
|
||||
|
||||
|
||||
class GoogleWifiSensor(Entity):
|
||||
"""Representation of a Google Wifi sensor."""
|
||||
|
||||
def __init__(self, hass, api, name, variable):
|
||||
"""Initialize a Pi-Hole sensor."""
|
||||
self._hass = hass
|
||||
self._api = api
|
||||
self._name = name
|
||||
self._state = STATE_UNKNOWN
|
||||
|
||||
variable_info = MONITORED_CONDITIONS[variable]
|
||||
self._var_name = variable
|
||||
self._var_units = variable_info[1]
|
||||
self._var_icon = variable_info[2]
|
||||
|
||||
@property
|
||||
def name(self):
|
||||
"""Return the name of the sensor."""
|
||||
return "{}_{}".format(self._name, self._var_name)
|
||||
|
||||
@property
|
||||
def icon(self):
|
||||
"""Icon to use in the frontend, if any."""
|
||||
return self._var_icon
|
||||
|
||||
@property
|
||||
def unit_of_measurement(self):
|
||||
"""Return the unit the value is expressed in."""
|
||||
return self._var_units
|
||||
|
||||
@property
|
||||
def availiable(self):
|
||||
"""Return availiability of goole wifi api."""
|
||||
return self._api.availiable
|
||||
|
||||
@property
|
||||
def state(self):
|
||||
"""Return the state of the device."""
|
||||
return self._state
|
||||
|
||||
def update(self):
|
||||
"""Get the latest data from the Google Wifi API."""
|
||||
self._api.update()
|
||||
if self.availiable:
|
||||
self._state = self._api.data[self._var_name]
|
||||
else:
|
||||
self._state = STATE_UNKNOWN
|
||||
|
||||
|
||||
class GoogleWifiAPI(object):
|
||||
"""Get the latest data and update the states."""
|
||||
|
||||
def __init__(self, host):
|
||||
"""Initialize the data object."""
|
||||
uri = 'http://'
|
||||
resource = "{}{}{}".format(uri, host, ENDPOINT)
|
||||
|
||||
self._request = requests.Request('GET', resource).prepare()
|
||||
self.raw_data = None
|
||||
self.data = {
|
||||
ATTR_CURRENT_VERSION: STATE_UNKNOWN,
|
||||
ATTR_NEW_VERSION: STATE_UNKNOWN,
|
||||
ATTR_UPTIME: STATE_UNKNOWN,
|
||||
ATTR_LAST_RESTART: STATE_UNKNOWN,
|
||||
ATTR_LOCAL_IP: STATE_UNKNOWN,
|
||||
ATTR_STATUS: STATE_UNKNOWN
|
||||
}
|
||||
self.availiable = True
|
||||
self.update()
|
||||
|
||||
@Throttle(MIN_TIME_BETWEEN_UPDATES)
|
||||
def update(self):
|
||||
"""Get the latest data from the router."""
|
||||
try:
|
||||
_LOGGER.error("Before request")
|
||||
with requests.Session() as sess:
|
||||
response = sess.send(
|
||||
self._request, timeout=10)
|
||||
self.raw_data = response.json()
|
||||
_LOGGER.error(self.raw_data)
|
||||
self.data_format()
|
||||
self.availiable = True
|
||||
except ValueError:
|
||||
_LOGGER.error("Unable to fetch data from Google Wifi")
|
||||
self.availiable = False
|
||||
self.raw_data = None
|
||||
|
||||
def data_format(self):
|
||||
"""Format raw data into easily accessible dict."""
|
||||
for key, value in self.raw_data.items():
|
||||
if key == 'software':
|
||||
self.data[ATTR_CURRENT_VERSION] = value['softwareVersion']
|
||||
if value['updateNewVersion'] == '0.0.0.0':
|
||||
self.data[ATTR_NEW_VERSION] = 'Latest'
|
||||
else:
|
||||
self.data[ATTR_NEW_VERSION] = value['updateNewVersion']
|
||||
elif key == 'system':
|
||||
self.data[ATTR_UPTIME] = value['uptime'] / (3600 * 24)
|
||||
last_restart = dt.now() - timedelta(seconds=value['uptime'])
|
||||
self.data[ATTR_LAST_RESTART] = \
|
||||
last_restart.strftime("%Y-%m-%d %H:%M:%S")
|
||||
elif key == 'wan':
|
||||
if value['online']:
|
||||
self.data[ATTR_STATUS] = 'Online'
|
||||
else:
|
||||
self.data[ATTR_STATUS] = 'Offline'
|
||||
if not value['ipAddress']:
|
||||
self.data[ATTR_LOCAL_IP] = STATE_UNKNOWN
|
||||
else:
|
||||
self.data[ATTR_LOCAL_IP] = value['localIpAddress']
|
||||
|
|
@ -0,0 +1,203 @@
|
|||
"""The tests for the Google Wifi platform."""
|
||||
import unittest
|
||||
from unittest.mock import patch, Mock
|
||||
from datetime import datetime, timedelta
|
||||
import requests_mock
|
||||
|
||||
from homeassistant import core as ha
|
||||
from homeassistant.setup import setup_component
|
||||
import homeassistant.components.sensor.google_wifi as google_wifi
|
||||
from homeassistant.const import STATE_UNKNOWN
|
||||
from homeassistant.util import dt as dt_util
|
||||
|
||||
from tests.common import get_test_home_assistant, assert_setup_component
|
||||
|
||||
NAME = 'foo'
|
||||
|
||||
MOCK_DATA = ('{"software": {"softwareVersion":"initial",'
|
||||
'"updateNewVersion":"initial"},'
|
||||
'"system": {"uptime":86400},'
|
||||
'"wan": {"localIpAddress":"initial", "online":true,'
|
||||
'"ipAddress":true}}')
|
||||
|
||||
MOCK_DATA_NEXT = ('{"software": {"softwareVersion":"next",'
|
||||
'"updateNewVersion":"0.0.0.0"},'
|
||||
'"system": {"uptime":172800},'
|
||||
'"wan": {"localIpAddress":"next", "online":false,'
|
||||
'"ipAddress":false}}')
|
||||
|
||||
|
||||
class TestGoogleWifiSetup(unittest.TestCase):
|
||||
"""Tests for setting up the Google Wifi switch platform."""
|
||||
|
||||
def setUp(self):
|
||||
"""Setup things to be run when tests are started."""
|
||||
self.hass = get_test_home_assistant()
|
||||
|
||||
def tearDown(self):
|
||||
"""Stop everything that was started."""
|
||||
self.hass.stop()
|
||||
|
||||
@requests_mock.Mocker()
|
||||
def test_setup_minimum(self, mock_req):
|
||||
"""Test setup with minimum configuration."""
|
||||
resource = '{}{}{}'.format('http://',
|
||||
google_wifi.DEFAULT_HOST,
|
||||
google_wifi.ENDPOINT)
|
||||
mock_req.get(resource, status_code=200)
|
||||
self.assertTrue(setup_component(self.hass, 'sensor', {
|
||||
'sensor': {
|
||||
'platform': 'google_wifi'
|
||||
}
|
||||
}))
|
||||
|
||||
@requests_mock.Mocker()
|
||||
def test_setup_get(self, mock_req):
|
||||
"""Test setup with full configuration."""
|
||||
resource = '{}{}{}'.format('http://',
|
||||
'localhost',
|
||||
google_wifi.ENDPOINT)
|
||||
mock_req.get(resource, status_code=200)
|
||||
self.assertTrue(setup_component(self.hass, 'sensor', {
|
||||
'sensor': {
|
||||
'platform': 'google_wifi',
|
||||
'host': 'localhost',
|
||||
'name': 'Test Wifi',
|
||||
'monitored_conditions': ['current_version',
|
||||
'new_version',
|
||||
'uptime',
|
||||
'last_restart',
|
||||
'local_ip',
|
||||
'status']
|
||||
}
|
||||
}))
|
||||
assert_setup_component(6, 'sensor')
|
||||
|
||||
|
||||
class TestGoogleWifiSensor(unittest.TestCase):
|
||||
"""Tests for Google Wifi sensor platform."""
|
||||
|
||||
def setUp(self):
|
||||
"""Setup things to be run when tests are started."""
|
||||
self.hass = get_test_home_assistant()
|
||||
with requests_mock.Mocker() as mock_req:
|
||||
self.setup_api(MOCK_DATA, mock_req)
|
||||
|
||||
def tearDown(self):
|
||||
"""Stop everything that was started."""
|
||||
self.hass.stop()
|
||||
|
||||
def setup_api(self, data, mock_req):
|
||||
"""Setup API with fake data."""
|
||||
resource = '{}{}{}'.format('http://',
|
||||
'localhost',
|
||||
google_wifi.ENDPOINT)
|
||||
now = datetime(1970, month=1, day=1)
|
||||
with patch('homeassistant.util.dt.now', return_value=now):
|
||||
mock_req.get(resource, text=data, status_code=200)
|
||||
self.api = google_wifi.GoogleWifiAPI("localhost")
|
||||
self.name = NAME
|
||||
self.sensor_dict = dict()
|
||||
for condition, cond_list in google_wifi.MONITORED_CONDITIONS.items():
|
||||
sensor = google_wifi.GoogleWifiSensor(self.hass, self.api,
|
||||
self.name, condition)
|
||||
name = '{}_{}'.format(self.name, condition)
|
||||
units = cond_list[1]
|
||||
icon = cond_list[2]
|
||||
self.sensor_dict[condition] = {'sensor': sensor,
|
||||
'name': name,
|
||||
'units': units,
|
||||
'icon': icon}
|
||||
|
||||
def fake_delay(self, ha_delay):
|
||||
"""Fake delay to prevent update throttle."""
|
||||
hass_now = dt_util.utcnow()
|
||||
shifted_time = hass_now + timedelta(seconds=ha_delay)
|
||||
self.hass.bus.fire(ha.EVENT_TIME_CHANGED,
|
||||
{ha.ATTR_NOW: shifted_time})
|
||||
|
||||
def test_name(self):
|
||||
"""Test the name."""
|
||||
for name in self.sensor_dict:
|
||||
sensor = self.sensor_dict[name]['sensor']
|
||||
test_name = self.sensor_dict[name]['name']
|
||||
self.assertEqual(test_name, sensor.name)
|
||||
|
||||
def test_unit_of_measurement(self):
|
||||
"""Test the unit of measurement."""
|
||||
for name in self.sensor_dict:
|
||||
sensor = self.sensor_dict[name]['sensor']
|
||||
self.assertEqual(self.sensor_dict[name]['units'],
|
||||
sensor.unit_of_measurement)
|
||||
|
||||
def test_icon(self):
|
||||
"""Test the icon."""
|
||||
for name in self.sensor_dict:
|
||||
sensor = self.sensor_dict[name]['sensor']
|
||||
self.assertEqual(self.sensor_dict[name]['icon'], sensor.icon)
|
||||
|
||||
@requests_mock.Mocker()
|
||||
def test_state(self, mock_req):
|
||||
"""Test the initial state."""
|
||||
self.setup_api(MOCK_DATA, mock_req)
|
||||
now = datetime(1970, month=1, day=1)
|
||||
with patch('homeassistant.util.dt.now', return_value=now):
|
||||
for name in self.sensor_dict:
|
||||
sensor = self.sensor_dict[name]['sensor']
|
||||
self.fake_delay(2)
|
||||
sensor.update()
|
||||
if name == google_wifi.ATTR_LAST_RESTART:
|
||||
self.assertEqual('1969-12-31 00:00:00', sensor.state)
|
||||
elif name == google_wifi.ATTR_UPTIME:
|
||||
self.assertEqual(1, sensor.state)
|
||||
elif name == google_wifi.ATTR_STATUS:
|
||||
self.assertEqual('Online', sensor.state)
|
||||
else:
|
||||
self.assertEqual('initial', sensor.state)
|
||||
|
||||
@requests_mock.Mocker()
|
||||
def test_update_when_value_is_none(self, mock_req):
|
||||
"""Test state gets updated to unknown when sensor returns no data."""
|
||||
self.setup_api(None, mock_req)
|
||||
for name in self.sensor_dict:
|
||||
sensor = self.sensor_dict[name]['sensor']
|
||||
self.fake_delay(2)
|
||||
sensor.update()
|
||||
self.assertEqual(STATE_UNKNOWN, sensor.state)
|
||||
|
||||
@requests_mock.Mocker()
|
||||
def test_update_when_value_changed(self, mock_req):
|
||||
"""Test state gets updated when sensor returns a new status."""
|
||||
self.setup_api(MOCK_DATA_NEXT, mock_req)
|
||||
now = datetime(1970, month=1, day=1)
|
||||
with patch('homeassistant.util.dt.now', return_value=now):
|
||||
for name in self.sensor_dict:
|
||||
sensor = self.sensor_dict[name]['sensor']
|
||||
self.fake_delay(2)
|
||||
sensor.update()
|
||||
if name == google_wifi.ATTR_LAST_RESTART:
|
||||
self.assertEqual('1969-12-30 00:00:00', sensor.state)
|
||||
elif name == google_wifi.ATTR_UPTIME:
|
||||
self.assertEqual(2, sensor.state)
|
||||
elif name == google_wifi.ATTR_STATUS:
|
||||
self.assertEqual('Offline', sensor.state)
|
||||
elif name == google_wifi.ATTR_NEW_VERSION:
|
||||
self.assertEqual('Latest', sensor.state)
|
||||
elif name == google_wifi.ATTR_LOCAL_IP:
|
||||
self.assertEqual(STATE_UNKNOWN, sensor.state)
|
||||
else:
|
||||
self.assertEqual('next', sensor.state)
|
||||
|
||||
def test_update_when_unavailiable(self):
|
||||
"""Test state updates when Google Wifi unavailiable."""
|
||||
self.api.update = Mock('google_wifi.GoogleWifiAPI.update',
|
||||
side_effect=self.update_side_effect())
|
||||
for name in self.sensor_dict:
|
||||
sensor = self.sensor_dict[name]['sensor']
|
||||
sensor.update()
|
||||
self.assertEqual(STATE_UNKNOWN, sensor.state)
|
||||
|
||||
def update_side_effect(self):
|
||||
"""Mock representation of update function."""
|
||||
self.api.data = None
|
||||
self.api.availiable = False
|
||||
Loading…
Reference in New Issue