-
-
Notifications
You must be signed in to change notification settings - Fork 37.7k
Add device tracker for Linksys Smart Wifi devices #8144
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Changes from 1 commit
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,107 @@ | ||
| """ | ||
| Support for Linksys Smart Wifi routers. | ||
| """ | ||
| import logging | ||
| import threading | ||
| from datetime import timedelta | ||
|
|
||
| import homeassistant.helpers.config_validation as cv | ||
| import requests | ||
| import voluptuous as vol | ||
| from homeassistant.components.device_tracker import ( | ||
| DOMAIN, PLATFORM_SCHEMA, DeviceScanner) | ||
| from homeassistant.const import CONF_HOST | ||
| from homeassistant.util import Throttle | ||
|
|
||
| MIN_TIME_BETWEEN_SCANS = timedelta(seconds=5) | ||
| DEFAULT_TIMEOUT = 10 | ||
|
|
||
| _LOGGER = logging.getLogger(__name__) | ||
|
|
||
| PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({ | ||
| vol.Required(CONF_HOST): cv.string, | ||
| }) | ||
|
|
||
|
|
||
| def get_scanner(hass, config): | ||
| """Validate the configuration and return a Linksys AP scanner.""" | ||
| try: | ||
| return LinksysSmartWifiDeviceScanner(config[DOMAIN]) | ||
| except ConnectionError: | ||
| return None | ||
|
|
||
|
|
||
| class LinksysSmartWifiDeviceScanner(DeviceScanner): | ||
| """This class queries a Linksys Access Point.""" | ||
|
|
||
| def __init__(self, config): | ||
| """Initialize the scanner.""" | ||
| self.host = config[CONF_HOST] | ||
|
|
||
| self.lock = threading.Lock() | ||
| self.last_results = {} | ||
|
|
||
| # Check if the access point is accessible | ||
| response = self._make_request() | ||
| if not response.status_code == 200: | ||
| raise ConnectionError("Cannot connect to Linksys Access Point") | ||
|
|
||
| def scan_devices(self): | ||
| """Scan for new devices and return a list with found device IDs (MACs).""" | ||
| self._update_info() | ||
|
|
||
| return self.last_results.keys() | ||
|
|
||
| def get_device_name(self, mac): | ||
| """Return the name (if known) of the device.""" | ||
| return self.last_results.get(mac) | ||
|
|
||
| @Throttle(MIN_TIME_BETWEEN_SCANS) | ||
| def _update_info(self): | ||
| """Check for connected devices.""" | ||
| with self.lock: | ||
| _LOGGER.info("Checking Linksys Smart Wifi") | ||
|
|
||
| self.last_results = {} | ||
| response = self._make_request() | ||
| if response.status_code != 200: | ||
| _LOGGER.error("Unable to get device list from router, got HTTP status code %d", response.status_code) | ||
|
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. line too long (117 > 79 characters) |
||
| return False | ||
| try: | ||
| data = response.json() | ||
| result = data["responses"][0] | ||
| devices = result["output"]["devices"] | ||
| for device in devices: | ||
| macs = device["knownMACAddresses"] | ||
| if not macs: | ||
| _LOGGER.warn("Found device without known MAC address, skipping") | ||
|
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. line too long (88 > 79 characters) |
||
| continue | ||
| mac = macs[-1] | ||
| connections = device["connections"] | ||
| if not connections: | ||
| _LOGGER.debug("Device %s is not connected", mac) | ||
| continue | ||
| name = device["friendlyName"] | ||
| properties = device["properties"] | ||
| for prop in properties: | ||
| if prop["name"] == "userDeviceName": | ||
| name = prop["value"] | ||
| _LOGGER.debug("Device %s is connected", mac) | ||
| self.last_results[mac] = name | ||
| except (KeyError, IndexError): | ||
| _LOGGER.exception("Router returned unexpected response") | ||
| return False | ||
| return True | ||
|
|
||
| def _make_request(self): | ||
| # Weirdly enough, this doesn't seem to require authentication | ||
| data = [{ | ||
| "request": { | ||
| "sinceRevision": 0 | ||
| }, | ||
| "action": "http://linksys.com/jnap/devicelist/GetDevices" | ||
| }] | ||
| return requests.post('http://{}/JNAP/'.format(self.host), | ||
| timeout=DEFAULT_TIMEOUT, | ||
| headers={"X-JNAP-Action": "http://linksys.com/jnap/core/Transaction"}, | ||
|
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. line too long (99 > 79 characters) |
||
| json=data) | ||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
line too long (82 > 79 characters)