evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Cleanup old device_tracker stuff (#8627)

Browse source 
* Cleanup old device_tracker stuff

* Fix lint
This commit is contained in:
Pascal Vizeli 2017-07-24 16:45:02 +02:00 • committed by Paulus Schoutsen
parent 654ad41464
commit f86bd15580
22 changed files with 478 additions and 679 deletions
Download patch file Download diff file Expand all files Collapse all files

82
homeassistant/components/device_tracker/luci.py
View file

@ -7,8 +7,6 @@ https://home-assistant.io/components/device_tracker.luci/
import json
import logging
import re
import threading
from datetime import timedelta
import requests
import voluptuous as vol
@ -18,9 +16,6 @@ from homeassistant.exceptions import HomeAssistantError
from homeassistant.components.device_tracker import (
DOMAIN, PLATFORM_SCHEMA, DeviceScanner)
from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_USERNAME
from homeassistant.util import Throttle
MIN_TIME_BETWEEN_SCANS = timedelta(seconds=5)
_LOGGER = logging.getLogger(__name__)
@ -55,12 +50,8 @@ class LuciDeviceScanner(DeviceScanner):
self.parse_api_pattern = re.compile(r"(?P<param>\w*) = (?P<value>.*);")
self.lock = threading.Lock()
self.last_results = {}
self.refresh_token()
self.mac2name = None
self.success_init = self.token is not None
@ -75,24 +66,22 @@ class LuciDeviceScanner(DeviceScanner):
def get_device_name(self, device):
"""Return the name of the given device or None if we don't know."""
with self.lock:
if self.mac2name is None:
url = 'http://{}/cgi-bin/luci/rpc/uci'.format(self.host)
result = _req_json_rpc(url, 'get_all', 'dhcp',
params={'auth': self.token})
if result:
hosts = [x for x in result.values()
if x['.type'] == 'host' and
'mac' in x and 'name' in x]
mac2name_list = [
(x['mac'].upper(), x['name']) for x in hosts]
self.mac2name = dict(mac2name_list)
else:
# Error, handled in the _req_json_rpc
return
return self.mac2name.get(device.upper(), None)
if self.mac2name is None:
url = 'http://{}/cgi-bin/luci/rpc/uci'.format(self.host)
result = _req_json_rpc(url, 'get_all', 'dhcp',
params={'auth': self.token})
if result:
hosts = [x for x in result.values()
if x['.type'] == 'host' and
'mac' in x and 'name' in x]
mac2name_list = [
(x['mac'].upper(), x['name']) for x in hosts]
self.mac2name = dict(mac2name_list)
else:
# Error, handled in the _req_json_rpc
return
return self.mac2name.get(device.upper(), None)
@Throttle(MIN_TIME_BETWEEN_SCANS)
def _update_info(self):
"""Ensure the information from the Luci router is up to date.
@ -101,31 +90,30 @@ class LuciDeviceScanner(DeviceScanner):
if not self.success_init:
return False
with self.lock:
_LOGGER.info("Checking ARP")
_LOGGER.info("Checking ARP")
url = 'http://{}/cgi-bin/luci/rpc/sys'.format(self.host)
try:
result = _req_json_rpc(url, 'net.arptable',
params={'auth': self.token})
except InvalidLuciTokenError:
_LOGGER.info("Refreshing token")
self.refresh_token()
return False
if result:
self.last_results = []
for device_entry in result:
# Check if the Flags for each device contain
# NUD_REACHABLE and if so, add it to last_results
if int(device_entry['Flags'], 16) & 0x2:
self.last_results.append(device_entry['HW address'])
return True
url = 'http://{}/cgi-bin/luci/rpc/sys'.format(self.host)
try:
result = _req_json_rpc(url, 'net.arptable',
params={'auth': self.token})
except InvalidLuciTokenError:
_LOGGER.info("Refreshing token")
self.refresh_token()
return False
if result:
self.last_results = []
for device_entry in result:
# Check if the Flags for each device contain
# NUD_REACHABLE and if so, add it to last_results
if int(device_entry['Flags'], 16) & 0x2:
self.last_results.append(device_entry['HW address'])
return True
return False
def _req_json_rpc(url, method, *args, **kwargs):
"""Perform one JSON RPC operation."""