<s> """<STR_LIT>""" <EOL> import logging <EOL> import re <EOL> import telnetlib <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:5> ) <EOL> _LOGGER = logging . getLogger ( __name__ ) <EOL> _LEASES_REGEX = re . compile ( <EOL> r'<STR_LIT>' + <EOL> r'<STR_LIT>' + <EOL> r'<STR_LIT>' + <EOL> r'<STR_LIT>' ) <EOL> _IP_NEIGH_REGEX = re . compile ( <EOL> r'<STR_LIT>' + <EOL> r'<STR_LIT>' + <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 = AsusWrtDeviceScanner ( config [ DOMAIN ] ) <EOL> return scanner if scanner . success_init else None <EOL> class AsusWrtDeviceScanner ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , config ) : <EOL> """<STR_LIT>""" <EOL> self . host = config [ CONF_HOST ] <EOL> self . username = str ( config [ CONF_USERNAME ] ) <EOL> self . password = str ( config [ CONF_PASSWORD ] ) <EOL> self . lock = threading . Lock ( ) <EOL> self . last_results = { } <EOL> data = self . get_asuswrt_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:host>' ] <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> _LOGGER . info ( "<STR_LIT>" ) <EOL> data = self . get_asuswrt_data ( ) <EOL> if not data : <EOL> return False <EOL> active_clients = [ client for client in data . values ( ) if <EOL> client [ '<STR_LIT:status>' ] == '<STR_LIT>' or <EOL> client [ '<STR_LIT:status>' ] == '<STR_LIT>' or <EOL> client [ '<STR_LIT:status>' ] == '<STR_LIT>' ] <EOL> self . last_results = active_clients <EOL> return True <EOL> def get_asuswrt_data ( self ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> telnet = telnetlib . Telnet ( self . host ) <EOL> telnet . read_until ( b'<STR_LIT>' ) <EOL> telnet . write ( ( self . username + '<STR_LIT:\n>' ) . encode ( '<STR_LIT:ascii>' ) ) <EOL> telnet . read_until ( b'<STR_LIT>' ) <EOL> telnet . write ( ( self . password + '<STR_LIT:\n>' ) . encode ( '<STR_LIT:ascii>' ) ) <EOL> prompt_string = telnet . read_until ( b'<STR_LIT:#>' ) . split ( b'<STR_LIT:\n>' ) [ - <NUM_LIT:1> ] <EOL> telnet . write ( '<STR_LIT>' . encode ( '<STR_LIT:ascii>' ) ) <EOL> neighbors = telnet . read_until ( prompt_string ) . split ( b'<STR_LIT:\n>' ) [ <NUM_LIT:1> : - <NUM_LIT:1> ] <EOL> telnet . write ( '<STR_LIT>' . encode ( '<STR_LIT:ascii>' ) ) <EOL> leases_result = telnet . read_until ( prompt_string ) . split ( b'<STR_LIT:\n>' ) [ <NUM_LIT:1> : - <NUM_LIT:1> ] <EOL> telnet . write ( '<STR_LIT>' . encode ( '<STR_LIT:ascii>' ) ) <EOL> except EOFError : <EOL> _LOGGER . exception ( "<STR_LIT>" ) <EOL> return <EOL> except ConnectionRefusedError : <EOL> _LOGGER . exception ( "<STR_LIT>" + <EOL> "<STR_LIT>" ) <EOL> return <EOL> devices = { } <EOL> for lease in leases_result : <EOL> match = _LEASES_REGEX . search ( lease . decode ( '<STR_LIT:utf-8>' ) ) <EOL> if not match : <EOL> _LOGGER . warning ( "<STR_LIT>" , lease ) <EOL> continue <EOL> host = match . group ( '<STR_LIT:host>' ) <EOL> if host == '<STR_LIT:*>' : <EOL> host = '<STR_LIT>' <EOL> devices [ match . group ( '<STR_LIT>' ) ] = { <EOL> '<STR_LIT:host>' : host , <EOL> '<STR_LIT:status>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : match . group ( '<STR_LIT>' ) , <EOL> '<STR_LIT>' : match . group ( '<STR_LIT>' ) . upper ( ) , <EOL> } <EOL> for neighbor in neighbors : <EOL> match = _IP_NEIGH_REGEX . search ( neighbor . decode ( '<STR_LIT:utf-8>' ) ) <EOL> if not match : <EOL> _LOGGER . warning ( "<STR_LIT>" , neighbor ) <EOL> continue <EOL> if match . group ( '<STR_LIT>' ) in devices : <EOL> devices [ match . group ( '<STR_LIT>' ) ] [ '<STR_LIT:status>' ] = match . group ( '<STR_LIT:status>' ) <EOL> return devices </s>
