pep8 + pylint run fixes. Also refactored the json-rpc to happen only in one place (external library would be nice but oh well).

This commit is contained in:
Markus Stenberg 2014-04-24 16:58:15 +03:00
parent fc6c2fbc18
commit 9af3d2b914

View File

@ -456,6 +456,7 @@ class NetgearDeviceScanner(object):
else: else:
return return
class LuciDeviceScanner(object): class LuciDeviceScanner(object):
""" This class queries a wireless router running OpenWrt firmware """ This class queries a wireless router running OpenWrt firmware
for connected devices. Adapted from Tomato scanner. for connected devices. Adapted from Tomato scanner.
@ -484,28 +485,39 @@ class LuciDeviceScanner(object):
self.mac2name = None self.mac2name = None
self.success_init = self.token self.success_init = self.token
def get_token(self, host, username, password): def _req_json_rpc(self, url, method, *args, **kwargs):
data = json.dumps({'method': 'login', """ Perform one JSON RPC operation. """
'params': [username, password]}) data = json.dumps({'method': method, 'params': args})
try: try:
r = requests.post('http://{}/cgi-bin/luci/rpc/auth'.format(host), data=data, timeout=3) res = requests.post(url, data=data, **kwargs)
if r.status_code == 200: except requests.exceptions.Timeout:
token = r.json()['result'] self.logger.exception("Connection to the router timed out")
self.logger.info('Authenticated') return
return token if res.status_code == 200:
elif r.status_code == 401: try:
result = res.json()
except ValueError:
# If json decoder could not parse the response
self.logger.exception("Failed to parse response from luci")
return
try:
return result['result']
except KeyError:
self.logger.exception("No result in response from luci")
return
elif res.status_code == 401:
# Authentication error # Authentication error
self.logger.exception( self.logger.exception(
"Failed to authenticate, " "Failed to authenticate, "
"please check your username and password") "please check your username and password")
return return
else: else:
self.logger.error("Invalid response: %s" % r) self.logger.error("Invalid response from luci: {}".format(res))
except requests.exceptions.Timeout:
self.logger.exception("Connection to the router timed out") def get_token(self, host, username, password):
except ValueError: """ Get authentication token for the given host+username+password """
# If json decoder could not parse the response url = 'http://{}/cgi-bin/luci/rpc/auth'.format(host)
self.logger.exception("Failed to parse response from router") return self._req_json_rpc(url, 'login', username, password)
def scan_devices(self): def scan_devices(self):
""" Scans for new devices and return a """ Scans for new devices and return a
@ -520,37 +532,17 @@ class LuciDeviceScanner(object):
with self.lock: with self.lock:
if self.mac2name is None: if self.mac2name is None:
try: url = 'http://{}/cgi-bin/luci/rpc/uci'.format(self.host)
data = json.dumps({'method': 'get_all', result = self._req_json_rpc(url, 'get_all', 'dhcp',
'params': ['dhcp']}) params={'auth': self.token})
if result:
r = requests.post('http://{}/cgi-bin/luci/rpc/uci'.format(self.host), params={'auth': self.token}, data=data, timeout=3) hosts = [x for x in result.values()
if x['.type'] == 'host' and
# Calling and parsing the Luci api here. We only need the 'mac' in x and 'name' in x]
# wldev and dhcpd_lease values. For API description see: mac2name_list = [(x['mac'], x['name']) for x in hosts]
# http://paulusschoutsen.nl/ self.mac2name = dict(mac2name_list)
# blog/2013/10/tomato-api-documentation/
if r.status_code == 200:
self.mac2name = dict(map(lambda x:(x['mac'], x['name']),
filter(lambda x:x['.type'] == 'host' and 'mac' in x and 'name' in x,
r.json()['result'].values())))
# Passthrough
else: else:
self.logger.error("Invalid response: %s" % r) # Error, handled in the _req_json_rpc
return
except requests.exceptions.Timeout:
# We get this if we could not connect to the router or
# an invalid http_id was supplied
self.logger.exception(
"Connection to the router timed out")
return
except ValueError:
# If json decoder could not parse the response
self.logger.exception(
"Failed to parse response from router")
return return
return self.mac2name.get(device, None) return self.mac2name.get(device, None)
@ -560,46 +552,20 @@ class LuciDeviceScanner(object):
if not self.success_init: if not self.success_init:
return False return False
with self.lock: with self.lock:
# if date_updated is None or the date is too old we scan
# if date_updated is None or the date is too old we scan for new data # for new data
if (not self.date_updated or datetime.now() - self.date_updated > if (not self.date_updated or datetime.now() - self.date_updated >
MIN_TIME_BETWEEN_SCANS): MIN_TIME_BETWEEN_SCANS):
self.logger.info("Checking ARP") self.logger.info("Checking ARP")
try: url = 'http://{}/cgi-bin/luci/rpc/sys'.format(self.host)
data = json.dumps({'method': 'net.arptable', result = self._req_json_rpc(url, 'net.arptable',
'params': []}) params={'auth': self.token})
if result:
r = requests.post('http://{}/cgi-bin/luci/rpc/sys'.format(self.host), params={'auth': self.token}, data=data, timeout=3) self.last_results = [x['HW address'] for x in result]
# Calling and parsing the Luci api here. We only need the
# wldev and dhcpd_lease values. For API description see:
# http://paulusschoutsen.nl/
# blog/2013/10/tomato-api-documentation/
if r.status_code == 200:
self.last_results = list(map(lambda x:x['HW address'], r.json()['result']))
self.date_updated = datetime.now() self.date_updated = datetime.now()
return True return True
else:
self.logger.error("Invalid response: %s" % r)
return False
except requests.exceptions.Timeout:
# We get this if we could not connect to the router or
# an invalid http_id was supplied
self.logger.exception(
"Connection to the router timed out")
return False
except ValueError:
# If json decoder could not parse the response
self.logger.exception(
"Failed to parse response from router")
return False return False
return True return True