|
| 1 | +"""Support for Linksys Velop routers.""" |
| 2 | +import logging |
| 3 | + |
| 4 | +import requests |
| 5 | +import voluptuous as vol |
| 6 | + |
| 7 | +import homeassistant.helpers.config_validation as cv |
| 8 | +from homeassistant.components.device_tracker import ( |
| 9 | + DOMAIN, |
| 10 | + PLATFORM_SCHEMA, |
| 11 | + DeviceScanner, |
| 12 | +) |
| 13 | + |
| 14 | +from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_USERNAME |
| 15 | +from base64 import b64encode as b64 |
| 16 | + |
| 17 | +DEFAULT_TIMEOUT = 10 |
| 18 | + |
| 19 | +_LOGGER = logging.getLogger(__name__) |
| 20 | + |
| 21 | +PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend( |
| 22 | + { |
| 23 | + vol.Required(CONF_HOST): cv.string, |
| 24 | + vol.Required(CONF_PASSWORD): cv.string, |
| 25 | + vol.Optional(CONF_USERNAME, default="admin"): cv.string, |
| 26 | + } |
| 27 | +) |
| 28 | + |
| 29 | + |
| 30 | +def get_scanner(hass, config): |
| 31 | + """Validate the configuration and return a Linksys AP scanner.""" |
| 32 | + try: |
| 33 | + return LinksysSmartWifiDeviceScanner(config[DOMAIN]) |
| 34 | + except ConnectionError: |
| 35 | + return None |
| 36 | + |
| 37 | + |
| 38 | +class LinksysSmartWifiDeviceScanner(DeviceScanner): |
| 39 | + """This class queries a Linksys Access Point.""" |
| 40 | + |
| 41 | + def __init__(self, config): |
| 42 | + """Initialize the scanner.""" |
| 43 | + self.host = config[CONF_HOST] |
| 44 | + self.username = config[CONF_USERNAME] |
| 45 | + self.password = config[CONF_PASSWORD] |
| 46 | + self.last_results = {} |
| 47 | + |
| 48 | + # Check if the access point is accessible |
| 49 | + response = self._make_request() |
| 50 | + if not response.status_code == 200: |
| 51 | + raise ConnectionError("Cannot connect to Linksys Access Point") |
| 52 | + |
| 53 | + def scan_devices(self): |
| 54 | + """Scan for new devices and return a list with device IDs (MACs).""" |
| 55 | + self._update_info() |
| 56 | + |
| 57 | + return self.last_results.keys() |
| 58 | + |
| 59 | + def get_device_name(self, device): |
| 60 | + """Return the name (if known) of the device.""" |
| 61 | + return self.last_results.get(device) |
| 62 | + |
| 63 | + def _update_info(self): |
| 64 | + """Check for connected devices.""" |
| 65 | + _LOGGER.info("Checking Linksys Smart Wifi") |
| 66 | + |
| 67 | + self.last_results = {} |
| 68 | + response = self._make_request() |
| 69 | + if response.status_code != 200: |
| 70 | + _LOGGER.error( |
| 71 | + "Got HTTP status code %d when getting device list", response.status_code |
| 72 | + ) |
| 73 | + return False |
| 74 | + try: |
| 75 | + data = response.json() |
| 76 | + result = data["responses"][0] |
| 77 | + devices = result["output"]["devices"] |
| 78 | + for device in devices: |
| 79 | + macs = device["knownMACAddresses"] |
| 80 | + if not macs: |
| 81 | + _LOGGER.warning("Skipping device without known MAC address") |
| 82 | + continue |
| 83 | + mac = macs[-1] |
| 84 | + connections = device["connections"] |
| 85 | + if not connections: |
| 86 | + _LOGGER.debug("Device %s is not connected", mac) |
| 87 | + continue |
| 88 | + |
| 89 | + name = None |
| 90 | + for prop in device["properties"]: |
| 91 | + if prop["name"] == "userDeviceName": |
| 92 | + name = prop["value"] |
| 93 | + if not name: |
| 94 | + name = device.get("friendlyName", device["deviceID"]) |
| 95 | + |
| 96 | + _LOGGER.debug("Device %s is connected", mac) |
| 97 | + self.last_results[mac] = name |
| 98 | + except (KeyError, IndexError): |
| 99 | + _LOGGER.exception("Router returned unexpected response") |
| 100 | + return False |
| 101 | + return True |
| 102 | + |
| 103 | + def _make_request(self): |
| 104 | + # Weirdly enough, this doesn't seem to require authentication |
| 105 | + data = [ |
| 106 | + { |
| 107 | + "request": {"sinceRevision": 0}, |
| 108 | + "action": "http://linksys.com/jnap/devicelist/GetDevices", |
| 109 | + } |
| 110 | + ] |
| 111 | + |
| 112 | + token = b64(bytes(self.username + ":" + self.password, "utf8")).decode("ascii"); |
| 113 | + |
| 114 | + headers = { |
| 115 | + "X-JNAP-Action": "http://linksys.com/jnap/core/Transaction", |
| 116 | + "X-JNAP-Authorization": "Basic " + token |
| 117 | + } |
| 118 | + |
| 119 | + |
| 120 | + return requests.post( |
| 121 | + f"http://{self.host}/JNAP/", |
| 122 | + timeout=DEFAULT_TIMEOUT, |
| 123 | + headers=headers, |
| 124 | + json=data, |
| 125 | + ) |
0 commit comments