jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-08 13:57:10 +00:00
Code Issues Packages Projects Releases Wiki Activity

corrected flake8 warnings

Browse Source
This commit is contained in:
Krzysztof Koziarek 2015-10-24 11:20:57 +02:00
parent 29c9c5a7ec
commit 50fbd83b3d
1 changed files with 48 additions and 40 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

88
homeassistant/components/device_tracker/ubus.py
View File

@ -46,13 +46,14 @@ class UbusDeviceScanner(object):
Configure your routers' ubus ACL based on following instructions: Configure your routers' ubus ACL based on following instructions:
http://wiki.openwrt.org/doc/techref/ubus http://wiki.openwrt.org/doc/techref/ubus
Read only access will be fine. Read only access will be fine.
To use this class you have to install rpcd-mod-file package in your OpenWrt router: To use this class you have to install rpcd-mod-file package
in your OpenWrt router:
opkg install rpcd-mod-file opkg install rpcd-mod-file
""" """
def __init__(self, config): def __init__(self, config):
@ -64,9 +65,9 @@ class UbusDeviceScanner(object):
self.last_results = {} self.last_results = {}
self.url = 'http://{}/ubus'.format(host) self.url = 'http://{}/ubus'.format(host)
self.session_id= _get_session_id(self.url, username, password) self.session_id = _get_session_id(self.url, username, password)
self.hostapd = [] self.hostapd = []
self.leasefile = None self.leasefile = None
self.mac2name = None self.mac2name = None
self.success_init = self.session_id is not None self.success_init = self.session_id is not None
@ -84,24 +85,28 @@ class UbusDeviceScanner(object):
with self.lock: with self.lock:
if self.leasefile is None: if self.leasefile is None:
result = _req_json_rpc(self.url, self.session_id, 'call', 'uci', 'get', config="dhcp", type="dnsmasq") result = _req_json_rpc(self.url, self.session_id,
'call', 'uci', 'get',
config="dhcp", type="dnsmasq")
if result: if result:
self.leasefile=next (iter (result["values"].values()))["leasefile"] self.leasefile = next(iter(result["values"].
values()))["leasefile"]
else: else:
return return
if self.mac2name is None: if self.mac2name is None:
result = _req_json_rpc(self.url, self.session_id, 'call', 'file', 'read', path=self.leasefile) result = _req_json_rpc(self.url, self.session_id,
'call', 'file', 'read',
path=self.leasefile)
if result: if result:
self.mac2name = dict() self.mac2name = dict()
for line in result["data"].splitlines(): for line in result["data"].splitlines():
[time, mac, ip, name, lid] = line.split(" ") [time, mac, ip, name, lid] = line.split(" ")
self.mac2name[mac.upper()] = name self.mac2name[mac.upper()] = name
else: else:
# Error, handled in the _req_json_rpc # Error, handled in the _req_json_rpc
return return
return self.mac2name.get(device.upper(), None) return self.mac2name.get(device.upper(), None)
@Throttle(MIN_TIME_BETWEEN_SCANS) @Throttle(MIN_TIME_BETWEEN_SCANS)
@ -115,57 +120,60 @@ class UbusDeviceScanner(object):
with self.lock: with self.lock:
_LOGGER.info("Checking ARP") _LOGGER.info("Checking ARP")
if not self.hostapd: if not self.hostapd:
hostapd = _req_json_rpc(self.url, self.session_id, 'list', 'hostapd.*', '') hostapd = _req_json_rpc(self.url, self.session_id,
'list', 'hostapd.*', '')
for key in hostapd.keys(): for key in hostapd.keys():
self.hostapd.append(key) self.hostapd.append(key)
self.last_results = [] self.last_results = []
results = 0 results = 0
for hostapd in self.hostapd: for hostapd in self.hostapd:
result = _req_json_rpc(self.url, self.session_id, 'call', hostapd, 'get_clients') result = _req_json_rpc(self.url, self.session_id,
'call', hostapd, 'get_clients')
if result: if result:
results = results + 1 results = results + 1
for key in result["clients"].keys(): for key in result["clients"].keys():
self.last_results.append(key) self.last_results.append(key)
if results: if results:
return True return True
else: else:
return False return False
def _req_json_rpc(url, session_id, rpcmethod, subsystem, method, **params): def _req_json_rpc(url, session_id, rpcmethod, subsystem, method, **params):
""" Perform one JSON RPC operation. """ """ Perform one JSON RPC operation. """
data = json.dumps({ "jsonrpc": "2.0", data = json.dumps({"jsonrpc": "2.0",
"id": 1, "id": 1,
"method": rpcmethod, "method": rpcmethod,
"params": [ session_id, "params": [session_id,
subsystem, subsystem,
method, method,
params] params]
}) })
try: try:
res = requests.post(url, data=data, timeout=5) res = requests.post(url, data=data, timeout=5)
except requests.exceptions.Timeout: except requests.exceptions.Timeout:
return return
if res.status_code == 200: if res.status_code == 200:
response = res.json() response = res.json()
if (rpcmethod == "call"): if (rpcmethod == "call"):
return response["result"][1] return response["result"][1]
else: else:
return response["result"] return response["result"]
def _get_session_id(url, username, password): def _get_session_id(url, username, password):
""" Get authentication token for the given host+username+password. """ """ Get authentication token for the given host+username+password. """
res = _req_json_rpc(url, "00000000000000000000000000000000", 'call', 'session', 'login', username=username, password=password) res = _req_json_rpc(url, "00000000000000000000000000000000", 'call',
'session', 'login', username=username,
password=password)
return res["ubus_rpc_session"] return res["ubus_rpc_session"]
# root@dom:~# ubus call uci get '{ "config": "dhcp", "type": "dnsmasq" }'