<s> """<STR_LIT>""" <EOL> import logging <EOL> import re <EOL> import threading <EOL> from datetime import timedelta <EOL> from homeassistant . components . device_tracker import DOMAIN <EOL> from homeassistant . const import CONF_HOST , CONF_PASSWORD , CONF_USERNAME <EOL> from homeassistant . helpers import validate_config <EOL> from homeassistant . util import Throttle <EOL> MIN_TIME_BETWEEN_SCANS = timedelta ( seconds = <NUM_LIT:10> ) <EOL> REQUIREMENTS = [ '<STR_LIT>' ] <EOL> _LOGGER = logging . getLogger ( __name__ ) <EOL> _DEVICES_REGEX = re . compile ( <EOL> r'<STR_LIT>' + <EOL> r'<STR_LIT>' + <EOL> r'<STR_LIT>' ) <EOL> def get_scanner ( hass , config ) : <EOL> """<STR_LIT>""" <EOL> if not validate_config ( config , <EOL> { DOMAIN : [ CONF_HOST , CONF_USERNAME , CONF_PASSWORD ] } , <EOL> _LOGGER ) : <EOL> return None <EOL> scanner = ArubaDeviceScanner ( config [ DOMAIN ] ) <EOL> return scanner if scanner . success_init else None <EOL> class ArubaDeviceScanner ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , config ) : <EOL> """<STR_LIT>""" <EOL> self . host = config [ CONF_HOST ] <EOL> self . username = config [ CONF_USERNAME ] <EOL> self . password = config [ CONF_PASSWORD ] <EOL> self . lock = threading . Lock ( ) <EOL> self . last_results = { } <EOL> data = self . get_aruba_data ( ) <EOL> self . success_init = data is not None <EOL> def scan_devices ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _update_info ( ) <EOL> return [ client [ '<STR_LIT>' ] for client in self . last_results ] <EOL> def get_device_name ( self , device ) : <EOL> """<STR_LIT>""" <EOL> if not self . last_results : <EOL> return None <EOL> for client in self . last_results : <EOL> if client [ '<STR_LIT>' ] == device : <EOL> return client [ '<STR_LIT:name>' ] <EOL> return None <EOL> @ Throttle ( MIN_TIME_BETWEEN_SCANS ) <EOL> def _update_info ( self ) : <EOL> """<STR_LIT>""" <EOL> if not self . success_init : <EOL> return False <EOL> with self . lock : <EOL> data = self . get_aruba_data ( ) <EOL> if not data : <EOL> return False <EOL> self . last_results = data . values ( ) <EOL> return True <EOL> def get_aruba_data ( self ) : <EOL> """<STR_LIT>""" <EOL> import pexpect <EOL> connect = "<STR_LIT>" <EOL> ssh = pexpect . spawn ( connect . format ( self . username , self . host ) ) <EOL> query = ssh . expect ( [ '<STR_LIT>' , pexpect . TIMEOUT , pexpect . EOF , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ] , timeout = <NUM_LIT> ) <EOL> if query == <NUM_LIT:1> : <EOL> _LOGGER . error ( "<STR_LIT>" ) <EOL> return <EOL> elif query == <NUM_LIT:2> : <EOL> _LOGGER . error ( "<STR_LIT>" ) <EOL> return <EOL> elif query == <NUM_LIT:3> : <EOL> ssh . sendline ( '<STR_LIT:yes>' ) <EOL> ssh . expect ( '<STR_LIT>' ) <EOL> elif query == <NUM_LIT:4> : <EOL> _LOGGER . error ( "<STR_LIT>" ) <EOL> return <EOL> elif query == <NUM_LIT:5> : <EOL> _LOGGER . error ( "<STR_LIT>" ) <EOL> return <EOL> elif query == <NUM_LIT:6> : <EOL> _LOGGER . error ( "<STR_LIT>" ) <EOL> return <EOL> ssh . sendline ( self . password ) <EOL> ssh . expect ( '<STR_LIT:#>' ) <EOL> ssh . sendline ( '<STR_LIT>' ) <EOL> ssh . expect ( '<STR_LIT:#>' ) <EOL> devices_result = ssh . before . split ( b'<STR_LIT:\r\n>' ) <EOL> ssh . sendline ( '<STR_LIT>' ) <EOL> devices = { } <EOL> for device in devices_result : <EOL> match = _DEVICES_REGEX . search ( device . decode ( '<STR_LIT:utf-8>' ) ) <EOL> if match : <EOL> devices [ match . group ( '<STR_LIT>' ) ] = { <EOL> '<STR_LIT>' : match . group ( '<STR_LIT>' ) , <EOL> '<STR_LIT>' : match . group ( '<STR_LIT>' ) . upper ( ) , <EOL> '<STR_LIT:name>' : match . group ( '<STR_LIT:name>' ) <EOL> } <EOL> return devices </s>
