Cisco IOS device tracker support (#4193)

.coveragerc

@@ -144,6 +144,7 @@ omit =
     homeassistant/components/device_tracker/bluetooth_le_tracker.py
     homeassistant/components/device_tracker/bluetooth_tracker.py
     homeassistant/components/device_tracker/bt_home_hub_5.py
+    homeassistant/components/device_tracker/cisco_ios.py
     homeassistant/components/device_tracker/fritz.py
     homeassistant/components/device_tracker/icloud.py
     homeassistant/components/device_tracker/luci.py

homeassistant/components/device_tracker/cisco_ios.py

+"""
+Support for Cisco IOS Routers.
+
+For more details about this platform, please refer to the documentation at
+https://home-assistant.io/components/device_tracker.cisco_ios/
+"""
+import logging
+from datetime import timedelta
+
+import voluptuous as vol
+
+import homeassistant.helpers.config_validation as cv
+from homeassistant.components.device_tracker import DOMAIN, PLATFORM_SCHEMA
+from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_USERNAME, \
+    CONF_PORT
+from homeassistant.util import Throttle
+
+# Return cached results if last scan was less then this time ago.
+MIN_TIME_BETWEEN_SCANS = timedelta(seconds=5)
+
+_LOGGER = logging.getLogger(__name__)
+
+REQUIREMENTS = ['pexpect==4.0.1']
+
+PLATFORM_SCHEMA = vol.All(
+    PLATFORM_SCHEMA.extend({
+        vol.Required(CONF_HOST): cv.string,
+        vol.Required(CONF_USERNAME): cv.string,
+        vol.Optional(CONF_PASSWORD, default=''): cv.string,
+        vol.Optional(CONF_PORT): cv.port,
+    })
+)
+
+
+def get_scanner(hass, config):
+    """Validate the configuration and return a Cisco scanner."""
+    scanner = CiscoDeviceScanner(config[DOMAIN])
+
+    return scanner if scanner.success_init else None
+
+
+class CiscoDeviceScanner(object):
+    """This class queries a wireless router running Cisco IOS firmware."""
+
+    def __init__(self, config):
+        """Initialize the scanner."""
+        self.host = config[CONF_HOST]
+        self.username = config[CONF_USERNAME]
+        self.port = config.get(CONF_PORT)
+        self.password = config.get(CONF_PASSWORD)
+
+        self.last_results = {}
+
+        self.success_init = self._update_info()
+        _LOGGER.info('cisco_ios scanner initialized')
+
+    # pylint: disable=no-self-use
+    def get_device_name(self, device):
+        """The firmware doesn't save the name of the wireless device."""
+        return None
+
+    def scan_devices(self):
+        """Scan for new devices and return a list with found device IDs."""
+        self._update_info()
+
+        return self.last_results
+
+    @Throttle(MIN_TIME_BETWEEN_SCANS)
+    def _update_info(self):
+        """
+        Ensure the information from the Cisco router is up to date.
+
+        Returns boolean if scanning successful.
+        """
+        string_result = self._get_arp_data()
+
+        if string_result:
+            self.last_results = []
+            last_results = []
+
+            lines_result = string_result.splitlines()
+
+            # Remove the first two lines, as they contains the arp command
+            # and the arp table titles e.g.
+            # show ip arp
+            # Protocol  Address | Age (min) | Hardware Addr | Type | Interface
+            lines_result = lines_result[2:]
+
+            for line in lines_result:
+                if len(line.split()) is 6:
+                    parts = line.split()
+                    if len(parts) != 6:
+                        continue
+
+                    # ['Internet', '10.10.11.1', '-', '0027.d32d.0123', 'ARPA',
+                    # 'GigabitEthernet0']
+                    age = parts[2]
+                    hw_addr = parts[3]
+
+                    if age != "-":
+                        mac = _parse_cisco_mac_address(hw_addr)
+                        age = int(age)
+                        if age < 1:
+                            last_results.append(mac)
+
+            self.last_results = last_results
+            return True
+
+        return False
+
+    def _get_arp_data(self):
+        """Open connection to the router and get arp entries."""
+        from pexpect import pxssh
+        import re
+
+        try:
+            cisco_ssh = pxssh.pxssh()
+            cisco_ssh.login(self.host, self.username, self.password,
+                            port=self.port, auto_prompt_reset=False)
+
+            # Find the hostname
+            initial_line = cisco_ssh.before.decode('utf-8').splitlines()
+            router_hostname = initial_line[len(initial_line) - 1]
+            router_hostname += "#"
+            # Set the discovered hostname as prompt
+            regex_expression = ('(?i)^%s' % router_hostname).encode()
+            cisco_ssh.PROMPT = re.compile(regex_expression, re.MULTILINE)
+            # Allow full arp table to print at once
+            cisco_ssh.sendline("terminal length 0")
+            cisco_ssh.prompt(1)
+
+            cisco_ssh.sendline("show ip arp")
+            cisco_ssh.prompt(1)
+
+            devices_result = cisco_ssh.before
+
+            return devices_result.decode("utf-8")
+        except pxssh.ExceptionPxssh as px_e:
+            _LOGGER.error("pxssh failed on login.")
+            _LOGGER.error(px_e)
+
+        return None
+
+
+def _parse_cisco_mac_address(cisco_hardware_addr):
+    """
+    Parse a Cisco formatted HW address to normal MAC.
+
+    e.g. convert
+    001d.ec02.07ab
+
+    to:
+    00:1D:EC:02:07:AB
+
+    Takes in cisco_hwaddr: HWAddr String from Cisco ARP table
+    Returns a regular standard MAC address
+    """
+    cisco_hardware_addr = cisco_hardware_addr.replace('.', '')
+    blocks = [cisco_hardware_addr[x:x + 2]
+              for x in range(0, len(cisco_hardware_addr), 2)]
+
+    return ':'.join(blocks).upper()

requirements_all.txt

@@ -294,6 +294,7 @@ panasonic_viera==0.2
 
 # homeassistant.components.device_tracker.aruba
 # homeassistant.components.device_tracker.asuswrt
+# homeassistant.components.device_tracker.cisco_ios
 # homeassistant.components.media_player.pandora
 pexpect==4.0.1