Initial

unifi-low-rssi-reconnect

@@ -0,0 +1,63 @@
+#!/usr/bin/env python
+
+import logging
+import argparse
+import os
+import sys
+import time
+from collections import defaultdict
+
+from unifi.controller import Controller
+
+parser = argparse.ArgumentParser()
+parser.add_argument('-c', '--controller', default='unifi', help='the controller address (default "unifi")')
+parser.add_argument('-u', '--username', default='admin', help='the controller usernane (default("admin")')
+parser.add_argument('-p', '--password', default='', help='the controller password')
+parser.add_argument('-m', '--minrssi', default=15, type=int, help='(dB) minimum required client RSSI (default 15)')
+parser.add_argument('-i', '--checkintv', default=5, type=int, help='(s) check interval (default 5)')
+parser.add_argument('-o', '--holdintv', default=300, type=int, help='(s) holddown interval (default 300)')
+parser.add_argument('-d', '--debug', help='enable debug output', action='store_true')
+args = parser.parse_args()
+
+logging.basicConfig(level=logging.INFO,
+        format='%(asctime)s %(levelname)s %(name)s: %(message)s',
+        datefmt='%Y-%m-%d %H:%M:%S')
+log = logging.getLogger(os.path.basename(sys.argv[0]))
+
+if args.debug:
+    log.setLevel(logging.DEBUG)
+
+c = Controller(args.controller, args.username, args.password)
+c.login()
+
+all_aps = c.get_aps()
+ap_names = dict([(ap['mac'], ap['name']) for ap in all_aps])
+
+kicks = defaultdict(int)
+held = defaultdict(bool)
+while True:
+    res = c.get_clients()
+    now = time.time()
+
+    log.debug('Got %d clients to check', len(res))
+    for sta in res:
+        name = sta['hostname']
+        rssi = sta['rssi']
+        mac = sta['mac']
+        ap_mac = sta['ap_mac']
+        ap_name = ap_names[ap_mac]
+        log.debug('%s/%s@%s %d', name, mac, ap_name, rssi)
+        if rssi < args.minrssi:
+            if now - kicks[name] > args.holdintv:
+                log.info('Disconnecting %s/%s@%s (RSSI %d dB < %d dB)',
+                        name, mac, ap_name, rssi, args.minrssi)
+                c.reconnect_client(mac)
+                kicks[name] = now
+                held[name] = False
+            elif not held[name]:
+                held[name] = True
+                log.debug('Ignoring %s/%s@%s (RSSI %d dB < %d dB) (holddown)',
+                        name, mac, ap_name, rssi, args.minrssi)
+
+    time.sleep(args.checkintv)
+

unifi-ls-clients

@@ -0,0 +1,34 @@
+#!/usr/bin/env python
+
+import argparse
+
+from unifi.controller import Controller
+
+parser = argparse.ArgumentParser()
+parser.add_argument('-c', '--controller', default='unifi', help='the controller address (default "unifi")')
+parser.add_argument('-u', '--username', default='admin', help='the controller usernane (default("admin")')
+parser.add_argument('-p', '--password', default='', help='the controller password')
+args = parser.parse_args()
+
+c = Controller(args.controller, args.username, args.password)
+c.login()
+
+aps = c.get_aps()
+
+FORMAT = '%-16s  %18s  %-12s  %4s  %4s  %3s  %3s'
+print(FORMAT % ('NAME', 'MAC', 'AP', 'CHAN', 'RSSI', 'RX', 'TX'))
+for ap in aps:
+    ap_name = ap['name']
+
+    for vap in ap['vap_table']:
+        channel = vap['channel']
+
+        for sta in vap['sta_table']:
+            name = sta['hostname'] or sta['ip'] or sta['mac']
+            rssi = sta['rssi']
+            mac = sta['mac']
+            rx = int(sta['rx_rate'] / 1000)
+            tx = int(sta['tx_rate'] / 1000)
+
+            print(FORMAT % (name, mac, ap_name, channel, rssi, rx, tx))
+

unifi/controller.py

@@ -0,0 +1,77 @@
+import cookielib
+import json
+import logging
+import urllib
+import urllib2
+
+log = logging.getLogger(__name__)
+
+class APIError(Exception):
+    pass
+
+class Controller:
+    def __init__(self, host, username, password):
+        self.host = host
+        self.username = username
+        self.password = password
+        self.url = 'https://' + host + ':8443/'
+        log.debug('Controller for %s', self.url)
+
+        cj = cookielib.CookieJar()
+        self.opener = urllib2.build_opener(urllib2.HTTPCookieProcessor(cj))
+
+    def _jsondec(self, data):
+        obj = json.loads(data)
+        if 'meta' in obj:
+            if obj['meta']['rc'] != 'ok':
+                raise APIError(obj['meta']['msg'])
+        if 'data' in obj:
+            return obj['data']
+        return obj
+
+    def _read(self, url, params=None):
+        res = self.opener.open(url, params)
+        return self._jsondec(res.read())
+
+    def login(self):
+        log.debug('login() as %s', self.username)
+        params = urllib.urlencode({'login': 'login',
+            'username': self.username, 'password': self.password})
+        self.opener.open(self.url + 'login', params).read()
+
+    def get_aps(self):
+        js = json.dumps({'_depth': 2, 'test': None})
+        params = urllib.urlencode({'json': js})
+        return self._read(self.url + 'api/stat/device', params)
+
+    def get_clients(self):
+        return self._read(self.url + 'api/stat/sta')
+
+    def get_wlan_conf(self):
+        return self._read(self.url + 'api/list/wlanconf')
+
+    def _mac_cmd(self, target_mac, command, mgr='stamgr'):
+        log.debug('_mac_cmd(%s, %s)', target_mac, command)
+        params = urllib.urlencode({'json':
+            {'mac': target_mac, 'cmd': command}})
+        self._read(self.url + 'api/cmd/' + mgr, params)
+
+    def block_client(self, mac):
+        self._mac_cmd(mac, 'block-sta')
+
+    def unblock_client(self, mac):
+        self._mac_cmd(mac, 'unblock-sta')
+
+    def reconnect_client(self, mac):
+        self._mac_cmd(mac, 'kick-sta')
+
+    def restart_ap(self, mac):
+        self._mac_cmd(mac, 'restart', 'devmgr')
+
+    def restart_ap_name(self, apnamefilter=''):
+        aplist = self.get_aps()
+        for ap in aplist:
+            if ap.get('state', 0) == 1:
+                if 'name' in ap and ap['name'].startswith(apnamefilter):
+                    self.reboot_ap(ap['mac'])
+