<s> import collections <EOL> import functools <EOL> import json <EOL> import os <EOL> import re <EOL> import subprocess <EOL> import sys <EOL> from glob import glob <EOL> from os . path import ( abspath , basename , dirname , expanduser , isdir , isfile , <EOL> join , splitdrive ) <EOL> from platform import system , uname <EOL> from threading import Thread <EOL> from platformio import __apiip__ , __apiurl__ , __version__ , exception <EOL> try : <EOL> from configparser import ConfigParser <EOL> except ImportError : <EOL> from ConfigParser import ConfigParser <EOL> class AsyncPipe ( Thread ) : <EOL> def __init__ ( self , outcallback = None ) : <EOL> Thread . __init__ ( self ) <EOL> self . outcallback = outcallback <EOL> self . _fd_read , self . _fd_write = os . pipe ( ) <EOL> self . _pipe_reader = os . fdopen ( self . _fd_read ) <EOL> self . _buffer = [ ] <EOL> self . start ( ) <EOL> def get_buffer ( self ) : <EOL> return self . _buffer <EOL> def fileno ( self ) : <EOL> return self . _fd_write <EOL> def run ( self ) : <EOL> for line in iter ( self . _pipe_reader . readline , "<STR_LIT>" ) : <EOL> line = line . strip ( ) <EOL> self . _buffer . append ( line ) <EOL> if self . outcallback : <EOL> self . outcallback ( line ) <EOL> else : <EOL> print line <EOL> self . _pipe_reader . close ( ) <EOL> def close ( self ) : <EOL> os . close ( self . _fd_write ) <EOL> self . join ( ) <EOL> class cd ( object ) : <EOL> def __init__ ( self , new_path ) : <EOL> self . new_path = new_path <EOL> self . prev_path = os . getcwd ( ) <EOL> def __enter__ ( self ) : <EOL> os . chdir ( self . new_path ) <EOL> def __exit__ ( self , etype , value , traceback ) : <EOL> os . chdir ( self . prev_path ) <EOL> class memoized ( object ) : <EOL> '''<STR_LIT>''' <EOL> def __init__ ( self , func ) : <EOL> self . func = func <EOL> self . cache = { } <EOL> def __call__ ( self , * args ) : <EOL> if not isinstance ( args , collections . Hashable ) : <EOL> return self . func ( * args ) <EOL> if args in self . cache : <EOL> return self . cache [ args ] <EOL> else : <EOL> value = self . func ( * args ) <EOL> self . cache [ args ] = value <EOL> return value <EOL> def __repr__ ( self ) : <EOL> '''<STR_LIT>''' <EOL> return self . func . __doc__ <EOL> def __get__ ( self , obj , objtype ) : <EOL> '''<STR_LIT>''' <EOL> return functools . partial ( self . __call__ , obj ) <EOL> def singleton ( cls ) : <EOL> """<STR_LIT>""" <EOL> _instances = { } <EOL> def get_instance ( * args , ** kwargs ) : <EOL> if cls not in _instances : <EOL> _instances [ cls ] = cls ( * args , ** kwargs ) <EOL> return _instances [ cls ] <EOL> return get_instance <EOL> def load_json ( file_path ) : <EOL> with open ( file_path , "<STR_LIT:r>" ) as f : <EOL> return json . load ( f ) <EOL> def get_systype ( ) : <EOL> data = uname ( ) <EOL> type_ = data [ <NUM_LIT:0> ] . lower ( ) <EOL> arch = data [ <NUM_LIT:4> ] . lower ( ) if data [ <NUM_LIT:4> ] else "<STR_LIT>" <EOL> return "<STR_LIT>" % ( type_ , arch ) if arch else type_ <EOL> def pioversion_to_intstr ( ) : <EOL> vermatch = re . match ( r"<STR_LIT>" , __version__ ) <EOL> assert vermatch <EOL> return [ int ( i ) for i in vermatch . group ( <NUM_LIT:1> ) . split ( "<STR_LIT:.>" ) [ : <NUM_LIT:3> ] ] <EOL> def _get_projconf_option_dir ( name , default = None ) : <EOL> _env_name = "<STR_LIT>" % name . upper ( ) <EOL> if _env_name in os . environ : <EOL> return os . getenv ( _env_name ) <EOL> try : <EOL> config = get_project_config ( ) <EOL> if ( config . has_section ( "<STR_LIT>" ) and <EOL> config . has_option ( "<STR_LIT>" , name ) ) : <EOL> option_dir = config . get ( "<STR_LIT>" , name ) <EOL> if option_dir . startswith ( "<STR_LIT>" ) : <EOL> option_dir = expanduser ( option_dir ) <EOL> return abspath ( option_dir ) <EOL> except exception . NotPlatformProject : <EOL> pass <EOL> return default <EOL> def get_home_dir ( ) : <EOL> home_dir = _get_projconf_option_dir ( <EOL> "<STR_LIT>" , <EOL> join ( expanduser ( "<STR_LIT>" ) , "<STR_LIT>" ) <EOL> ) <EOL> if "<STR_LIT>" in get_systype ( ) : <EOL> try : <EOL> home_dir . encode ( "<STR_LIT:utf8>" ) <EOL> except UnicodeDecodeError : <EOL> home_dir = splitdrive ( home_dir ) [ <NUM_LIT:0> ] + "<STR_LIT>" <EOL> if not isdir ( home_dir ) : <EOL> os . makedirs ( home_dir ) <EOL> assert isdir ( home_dir ) <EOL> return home_dir <EOL> def get_lib_dir ( ) : <EOL> return _get_projconf_option_dir ( <EOL> "<STR_LIT>" , <EOL> join ( get_home_dir ( ) , "<STR_LIT>" ) <EOL> ) <EOL> def get_source_dir ( ) : <EOL> curpath = abspath ( __file__ ) <EOL> if not isfile ( curpath ) : <EOL> for p in sys . path : <EOL> if isfile ( join ( p , __file__ ) ) : <EOL> curpath = join ( p , __file__ ) <EOL> break <EOL> return dirname ( curpath ) <EOL> def get_project_dir ( ) : <EOL> return os . getcwd ( ) <EOL> def get_projectsrc_dir ( ) : <EOL> return _get_projconf_option_dir ( <EOL> "<STR_LIT>" , <EOL> join ( get_project_dir ( ) , "<STR_LIT:src>" ) <EOL> ) <EOL> def get_projectlib_dir ( ) : <EOL> return join ( get_project_dir ( ) , "<STR_LIT>" ) <EOL> def get_pioenvs_dir ( ) : <EOL> return _get_projconf_option_dir ( <EOL> "<STR_LIT>" , <EOL> join ( get_project_dir ( ) , "<STR_LIT>" ) <EOL> ) <EOL> def get_projectdata_dir ( ) : <EOL> return _get_projconf_option_dir ( <EOL> "<STR_LIT>" , <EOL> join ( get_project_dir ( ) , "<STR_LIT:data>" ) <EOL> ) <EOL> def get_project_config ( ini_path = None ) : <EOL> if not ini_path : <EOL> ini_path = join ( get_project_dir ( ) , "<STR_LIT>" ) <EOL> if not isfile ( ini_path ) : <EOL> raise exception . NotPlatformProject ( get_project_dir ( ) ) <EOL> cp = ConfigParser ( ) <EOL> cp . read ( ini_path ) <EOL> return cp <EOL> def change_filemtime ( path , time ) : <EOL> os . utime ( path , ( time , time ) ) <EOL> def is_ci ( ) : <EOL> return os . getenv ( "<STR_LIT>" , "<STR_LIT>" ) . lower ( ) == "<STR_LIT:true>" <EOL> def exec_command ( * args , ** kwargs ) : <EOL> result = { <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : None <EOL> } <EOL> default = dict ( <EOL> stdout = subprocess . PIPE , <EOL> stderr = subprocess . PIPE , <EOL> shell = system ( ) == "<STR_LIT>" <EOL> ) <EOL> default . update ( kwargs ) <EOL> kwargs = default <EOL> p = subprocess . Popen ( * args , ** kwargs ) <EOL> try : <EOL> result [ '<STR_LIT>' ] , result [ '<STR_LIT>' ] = p . communicate ( ) <EOL> result [ '<STR_LIT>' ] = p . returncode <EOL> except KeyboardInterrupt : <EOL> raise exception . AbortedByUser ( ) <EOL> finally : <EOL> for s in ( "<STR_LIT>" , "<STR_LIT>" ) : <EOL> if isinstance ( kwargs [ s ] , AsyncPipe ) : <EOL> kwargs [ s ] . close ( ) <EOL> for s in ( "<STR_LIT>" , "<STR_LIT>" ) : <EOL> if isinstance ( kwargs [ s ] , AsyncPipe ) : <EOL> result [ s [ <NUM_LIT:3> : ] ] = "<STR_LIT:\n>" . join ( kwargs [ s ] . get_buffer ( ) ) <EOL> for k , v in result . iteritems ( ) : <EOL> if v and isinstance ( v , basestring ) : <EOL> result [ k ] . strip ( ) <EOL> return result <EOL> def get_serialports ( ) : <EOL> try : <EOL> from serial . tools . list_ports import comports <EOL> except ImportError : <EOL> raise exception . GetSerialPortsError ( os . name ) <EOL> result = [ ] <EOL> for p , d , h in comports ( ) : <EOL> if not p : <EOL> continue <EOL> if "<STR_LIT>" in get_systype ( ) : <EOL> try : <EOL> d = unicode ( d , errors = "<STR_LIT:ignore>" ) <EOL> except TypeError : <EOL> pass <EOL> result . append ( { "<STR_LIT:port>" : p , "<STR_LIT:description>" : d , "<STR_LIT>" : h } ) <EOL> if not result and system ( ) == "<STR_LIT>" : <EOL> for p in glob ( "<STR_LIT>" ) : <EOL> result . append ( { "<STR_LIT:port>" : p , "<STR_LIT:description>" : "<STR_LIT>" , "<STR_LIT>" : "<STR_LIT>" } ) <EOL> return result <EOL> def get_logicaldisks ( ) : <EOL> disks = [ ] <EOL> if system ( ) == "<STR_LIT>" : <EOL> result = exec_command ( <EOL> [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] ) . get ( "<STR_LIT>" ) <EOL> disknamere = re . compile ( r"<STR_LIT>" ) <EOL> for line in result . split ( "<STR_LIT:\n>" ) : <EOL> match = disknamere . match ( line . strip ( ) ) <EOL> if not match : <EOL> continue <EOL> disks . append ( { "<STR_LIT>" : match . group ( <NUM_LIT:1> ) , "<STR_LIT:name>" : match . group ( <NUM_LIT:2> ) } ) <EOL> else : <EOL> result = exec_command ( [ "<STR_LIT>" ] ) . get ( "<STR_LIT>" ) <EOL> disknamere = re . compile ( r"<STR_LIT>" , flags = re . I ) <EOL> for line in result . split ( "<STR_LIT:\n>" ) : <EOL> match = disknamere . search ( line . strip ( ) ) <EOL> if not match : <EOL> continue <EOL> disks . append ( { "<STR_LIT>" : match . group ( <NUM_LIT:1> ) , <EOL> "<STR_LIT:name>" : basename ( match . group ( <NUM_LIT:1> ) ) } ) <EOL> return disks <EOL> def get_request_defheaders ( ) : <EOL> import requests <EOL> return { "<STR_LIT>" : "<STR_LIT>" % ( <EOL> __version__ , int ( is_ci ( ) ) , requests . utils . default_user_agent ( ) <EOL> ) } <EOL> def get_api_result ( path , params = None , data = None , skipdns = False ) : <EOL> import requests <EOL> result = None <EOL> r = None <EOL> headers = get_request_defheaders ( ) <EOL> url = __apiurl__ <EOL> if skipdns : <EOL> url = "<STR_LIT>" % __apiip__ <EOL> headers [ '<STR_LIT:host>' ] = __apiurl__ [ __apiurl__ . index ( "<STR_LIT>" ) + <NUM_LIT:3> : ] <EOL> try : <EOL> if data : <EOL> r = requests . post ( <EOL> url + path , params = params , data = data , headers = headers ) <EOL> else : <EOL> r = requests . get ( url + path , params = params , headers = headers ) <EOL> r . raise_for_status ( ) <EOL> result = r . json ( ) <EOL> except requests . exceptions . HTTPError as e : <EOL> if result and "<STR_LIT>" in result : <EOL> raise exception . APIRequestError ( result [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT:title>' ] ) <EOL> else : <EOL> raise exception . APIRequestError ( e ) <EOL> except ( requests . exceptions . ConnectionError , <EOL> requests . exceptions . ConnectTimeout , <EOL> requests . exceptions . ReadTimeout ) : <EOL> if not skipdns : <EOL> return get_api_result ( path , params , data , skipdns = True ) <EOL> raise exception . APIRequestError ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> except ValueError : <EOL> raise exception . APIRequestError ( <EOL> "<STR_LIT>" % r . text . encode ( "<STR_LIT:utf-8>" ) ) <EOL> finally : <EOL> if r : <EOL> r . close ( ) <EOL> return result <EOL> @ memoized <EOL> def _lookup_boards ( ) : <EOL> boards = { } <EOL> bdirs = [ join ( get_source_dir ( ) , "<STR_LIT>" ) ] <EOL> if isdir ( join ( get_home_dir ( ) , "<STR_LIT>" ) ) : <EOL> bdirs . append ( join ( get_home_dir ( ) , "<STR_LIT>" ) ) <EOL> for bdir in bdirs : <EOL> for json_file in sorted ( os . listdir ( bdir ) ) : <EOL> if not json_file . endswith ( "<STR_LIT>" ) : <EOL> continue <EOL> boards . update ( load_json ( join ( bdir , json_file ) ) ) <EOL> return boards <EOL> def get_boards ( type_ = None ) : <EOL> boards = _lookup_boards ( ) <EOL> if type_ is None : <EOL> return boards <EOL> else : <EOL> if type_ not in boards : <EOL> raise exception . UnknownBoard ( type_ ) <EOL> return boards [ type_ ] <EOL> @ memoized <EOL> def _lookup_frameworks ( ) : <EOL> frameworks = { } <EOL> frameworks_path = join ( <EOL> get_source_dir ( ) , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ) <EOL> frameworks_list = [ f [ : - <NUM_LIT:3> ] for f in os . listdir ( frameworks_path ) <EOL> if not f . startswith ( "<STR_LIT>" ) and f . endswith ( "<STR_LIT>" ) ] <EOL> for _type in frameworks_list : <EOL> script_path = join ( frameworks_path , "<STR_LIT>" % _type ) <EOL> with open ( script_path ) as f : <EOL> fcontent = f . read ( ) <EOL> assert '<STR_LIT>' in fcontent <EOL> _doc_start = fcontent . index ( '<STR_LIT>' ) + <NUM_LIT:3> <EOL> fdoc = fcontent [ <EOL> _doc_start : fcontent . index ( '<STR_LIT>' , _doc_start ) ] . strip ( ) <EOL> doclines = [ l . strip ( ) for l in fdoc . splitlines ( ) if l . strip ( ) ] <EOL> frameworks [ _type ] = { <EOL> "<STR_LIT:name>" : doclines [ <NUM_LIT:0> ] , <EOL> "<STR_LIT:description>" : "<STR_LIT:U+0020>" . join ( doclines [ <NUM_LIT:1> : - <NUM_LIT:1> ] ) , <EOL> "<STR_LIT:url>" : doclines [ - <NUM_LIT:1> ] , <EOL> "<STR_LIT>" : script_path <EOL> } <EOL> return frameworks <EOL> def get_frameworks ( type_ = None ) : <EOL> frameworks = _lookup_frameworks ( ) <EOL> if type_ is None : <EOL> return frameworks <EOL> else : <EOL> if type_ not in frameworks : <EOL> raise exception . UnknownFramework ( type_ ) <EOL> return frameworks [ type_ ] <EOL> return frameworks <EOL> def where_is_program ( program , envpath = None ) : <EOL> env = os . environ <EOL> if envpath : <EOL> env [ '<STR_LIT>' ] = envpath <EOL> try : <EOL> result = exec_command ( <EOL> [ "<STR_LIT>" if "<STR_LIT>" in get_systype ( ) else "<STR_LIT>" , program ] , <EOL> env = env <EOL> ) <EOL> if result [ '<STR_LIT>' ] == <NUM_LIT:0> and isfile ( result [ '<STR_LIT>' ] . strip ( ) ) : <EOL> return result [ '<STR_LIT>' ] . strip ( ) <EOL> except OSError : <EOL> pass <EOL> for bin_dir in env . get ( "<STR_LIT>" , "<STR_LIT>" ) . split ( os . pathsep ) : <EOL> if isfile ( join ( bin_dir , program ) ) : <EOL> return join ( bin_dir , program ) <EOL> elif isfile ( join ( bin_dir , "<STR_LIT>" % program ) ) : <EOL> return join ( bin_dir , "<STR_LIT>" % program ) <EOL> return program </s>
