core/homeassistant/components/device_tracker/snmp.py

"""
homeassistant.components.device_tracker.demo
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

Fetch wifi associations through snmp

device_tracker:
  platform: snmp
  host: YOUR_WAP_IP
  community: SNMP_COMMUNITY
  baseoid: BASE_OID


Little help with base oids:
    Microtik: 1.3.6.1.4.1.14988.1.1.1.2.1.1 (confirmed)
    Aruba: 1.3.6.1.4.1.14823.2.3.3.1.2.4.1.2 (untested)

"""
import logging
from datetime import timedelta
import threading
import binascii

from homeassistant.const import CONF_HOST, CONF_COMMUNITY, CONF_BASEOID
from homeassistant.helpers import validate_config
from homeassistant.util import Throttle
from homeassistant.components.device_tracker import DOMAIN

MIN_TIME_BETWEEN_SCANS = timedelta(seconds=10)

_LOGGER = logging.getLogger(__name__)
REQUIREMENTS = ['pysnmp']

def setup_scanner(hass, config, see):
    """ Setup snmp scanning """
    #if not validate_config(config, {DOMAIN: [CONF_HOST, CONF_COMMUNITY, CONF_BASEOID]}, _LOGGER):
    #    return None

    #scanner = SnmpScanner(config[DOMAIN])
    scanner = SnmpScanner(config)

    return scanner if scanner.success_init else None


class SnmpScanner(object):
    """ This class queries any SNMP capable Acces Point for connected devices. """
    def __init__(self, config):
        self.host = config[CONF_HOST]
        self.community = config[CONF_COMMUNITY]
        self.baseoid = config[CONF_BASEOID]

        self.lock = threading.Lock()

        self.last_results = {}

        # Test the router is accessible
        data = self.get_snmp_data()
        self.success_init = data is not None

    def scan_devices(self):
        """
        Scans for new devices and return a list containing found device IDs.
        """

        self._update_info()
        return [client['mac'] for client in self.last_results]

    def get_device_name(self, device):
        """ Returns the name of the given device or None if we don't know. """
        return None

    @Throttle(MIN_TIME_BETWEEN_SCANS)
    def _update_info(self):
        """
        Ensures the information from the WAP is up to date.
        Returns boolean if scanning successful.
        """
        if not self.success_init:
            return False

        with self.lock:
            data = self.get_snmp_data()
            if not data:
                return False

            self.last_results = data
            return True

    def get_snmp_data(self):
        """ Fetch mac addresses from WAP via SNMP. """
        from pysnmp.entity.rfc3413.oneliner import cmdgen

        devices = {}

        cmdGen = cmdgen.CommandGenerator()
        errorIndication, errorStatus, errorIndex, varBindTable = cmdGen.nextCmd(
            cmdgen.CommunityData( self.community ),
            cmdgen.UdpTransportTarget( ( self.host , 161) ),
            cmdgen.MibVariable( self.baseoid )
        )
        if errorIndication:
            _LOGGER.exception( "SNMPLIB error: {}".format( errorIndication ) )
            return
        if errorStatus:
            _LOGGER.exception( "SNMP error: {} at {}".format( errorStatus.prettyPrint(), errorIndex and varBindTable[-1][int(errorIndex)-1] or '?' ) )
            return
        for varBindTableRow in varBindTable:
            for key,val in varBindTableRow:
                mac = binascii.hexlify( val.asOctets() ).decode('utf-8')
                mac = ':'.join( [ mac[i:i+2] for i in range( 0, len(mac), 2 ) ] )
                devices[mac] = { 'mac' : mac }
        return devices