Use Voluptuous for Luci and Netgear device trackers (#3123)

* Use Voluptuous for Luci and NEtgear device trackers

* str_schema shortcut

* Undo str_schema
This commit is contained in:
Johann Kellerman 2016-09-05 19:37:36 +02:00 committed by Teagan Glenn
parent e0a6d7941c
commit 17a2cac7e1
2 changed files with 35 additions and 30 deletions

View File

@ -11,10 +11,11 @@ import threading
from datetime import timedelta
import requests
import voluptuous as vol
from homeassistant.components.device_tracker import DOMAIN
import homeassistant.helpers.config_validation as cv
from homeassistant.components.device_tracker import DOMAIN, PLATFORM_SCHEMA
from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_USERNAME
from homeassistant.helpers import validate_config
from homeassistant.util import Throttle
# Return cached results if last scan was less then this time ago.
@ -22,14 +23,15 @@ MIN_TIME_BETWEEN_SCANS = timedelta(seconds=5)
_LOGGER = logging.getLogger(__name__)
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
vol.Required(CONF_HOST): cv.string,
vol.Required(CONF_USERNAME): cv.string,
vol.Required(CONF_PASSWORD): cv.string
})
def get_scanner(hass, config):
"""Validate the configuration and return a Luci scanner."""
if not validate_config(config,
{DOMAIN: [CONF_HOST, CONF_USERNAME, CONF_PASSWORD]},
_LOGGER):
return None
scanner = LuciDeviceScanner(config[DOMAIN])
return scanner if scanner.success_init else None
@ -93,7 +95,7 @@ class LuciDeviceScanner(object):
return False
with self.lock:
_LOGGER.info("Checking ARP")
_LOGGER.info('Checking ARP')
url = 'http://{}/cgi-bin/luci/rpc/sys'.format(self.host)
result = _req_json_rpc(url, 'net.arptable',
@ -117,19 +119,19 @@ def _req_json_rpc(url, method, *args, **kwargs):
try:
res = requests.post(url, data=data, timeout=5, **kwargs)
except requests.exceptions.Timeout:
_LOGGER.exception("Connection to the router timed out")
_LOGGER.exception('Connection to the router timed out')
return
if res.status_code == 200:
try:
result = res.json()
except ValueError:
# If json decoder could not parse the response
_LOGGER.exception("Failed to parse response from luci")
_LOGGER.exception('Failed to parse response from luci')
return
try:
return result['result']
except KeyError:
_LOGGER.exception("No result in response from luci")
_LOGGER.exception('No result in response from luci')
return
elif res.status_code == 401:
# Authentication error
@ -138,7 +140,7 @@ def _req_json_rpc(url, method, *args, **kwargs):
"please check your username and password")
return
else:
_LOGGER.error("Invalid response from luci: %s", res)
_LOGGER.error('Invalid response from luci: %s', res)
def _get_token(host, username, password):

View File

@ -8,9 +8,12 @@ import logging
import threading
from datetime import timedelta
from homeassistant.components.device_tracker import DOMAIN
from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_USERNAME, \
CONF_PORT
import voluptuous as vol
import homeassistant.helpers.config_validation as cv
from homeassistant.components.device_tracker import DOMAIN, PLATFORM_SCHEMA
from homeassistant.const import (
CONF_HOST, CONF_PASSWORD, CONF_USERNAME, CONF_PORT)
from homeassistant.util import Throttle
# Return cached results if last scan was less then this time ago.
@ -19,6 +22,17 @@ MIN_TIME_BETWEEN_SCANS = timedelta(seconds=5)
_LOGGER = logging.getLogger(__name__)
REQUIREMENTS = ['pynetgear==0.3.3']
DEFAULT_HOST = 'routerlogin.net'
DEFAULT_USER = 'admin'
DEFAULT_PORT = 5000
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
vol.Optional(CONF_HOST, default=DEFAULT_HOST): cv.string,
vol.Optional(CONF_USERNAME, default=DEFAULT_USER): cv.string,
vol.Required(CONF_PASSWORD): cv.string,
vol.Optional(CONF_PORT, default=DEFAULT_PORT): cv.port
})
def get_scanner(hass, config):
"""Validate the configuration and returns a Netgear scanner."""
@ -28,10 +42,6 @@ def get_scanner(hass, config):
password = info.get(CONF_PASSWORD)
port = info.get(CONF_PORT)
if password is not None and host is None:
_LOGGER.warning('Found username or password but no host')
return None
scanner = NetgearDeviceScanner(host, username, password, port)
return scanner if scanner.success_init else None
@ -47,16 +57,9 @@ class NetgearDeviceScanner(object):
self.last_results = []
self.lock = threading.Lock()
if host is None:
self._api = pynetgear.Netgear()
elif username is None:
self._api = pynetgear.Netgear(password, host)
elif port is None:
self._api = pynetgear.Netgear(password, host, username)
else:
self._api = pynetgear.Netgear(password, host, username, port)
self._api = pynetgear.Netgear(password, host, username, port)
_LOGGER.info("Logging in")
_LOGGER.info('Logging in')
results = self._api.get_attached_devices()
@ -65,7 +68,7 @@ class NetgearDeviceScanner(object):
if self.success_init:
self.last_results = results
else:
_LOGGER.error("Failed to Login")
_LOGGER.error('Failed to Login')
def scan_devices(self):
"""Scan for new devices and return a list with found device IDs."""
@ -91,7 +94,7 @@ class NetgearDeviceScanner(object):
return
with self.lock:
_LOGGER.info("Scanning")
_LOGGER.info('Scanning')
results = self._api.get_attached_devices()