<s> from contextlib import contextmanager <EOL> from glob import glob <EOL> from math import ceil <EOL> from os . path import join <EOL> from socket import inet_aton , inet_ntoa <EOL> from struct import pack , unpack <EOL> from subprocess import check_call <EOL> import requests <EOL> from platformio_api import __version__ , config <EOL> from platformio_api . exception import ( DLFileError , DLFileSizeError , <EOL> InvalidLibConf ) <EOL> def ip2int ( ip_string ) : <EOL> return unpack ( "<STR_LIT>" , inet_aton ( ip_string ) ) [ <NUM_LIT:0> ] <EOL> def int2ip ( ip_int ) : <EOL> return inet_ntoa ( pack ( "<STR_LIT>" , ip_int ) ) <EOL> def download_file ( source_url , destination_path ) : <EOL> CHUNK_SIZE = <NUM_LIT> <EOL> downloaded = <NUM_LIT:0> <EOL> f = None <EOL> r = None <EOL> try : <EOL> headers = { "<STR_LIT>" : "<STR_LIT>" % ( <EOL> __version__ , requests . utils . default_user_agent ( ) ) } <EOL> r = requests . get ( source_url , headers = headers , stream = True ) <EOL> if r . status_code != <NUM_LIT:200> : <EOL> raise DLFileError ( "<STR_LIT>" % ( <EOL> r . status_code , source_url ) ) <EOL> if int ( r . headers . get ( "<STR_LIT>" , <NUM_LIT:0> ) ) > config [ '<STR_LIT>' ] : <EOL> raise DLFileSizeError ( config [ '<STR_LIT>' ] , <EOL> int ( r . headers [ '<STR_LIT>' ] ) ) <EOL> f = open ( destination_path , "<STR_LIT:wb>" ) <EOL> for data in r . iter_content ( chunk_size = CHUNK_SIZE ) : <EOL> if downloaded > config [ '<STR_LIT>' ] : <EOL> raise DLFileSizeError ( config [ '<STR_LIT>' ] , downloaded ) <EOL> f . write ( data ) <EOL> downloaded += CHUNK_SIZE <EOL> finally : <EOL> if f : <EOL> f . close ( ) <EOL> if r : <EOL> r . close ( ) <EOL> def create_archive ( archive_path , source_dir ) : <EOL> if archive_path . endswith ( "<STR_LIT>" ) : <EOL> check_call ( [ "<STR_LIT>" , "<STR_LIT>" , archive_path , "<STR_LIT>" , source_dir , "<STR_LIT:.>" ] ) <EOL> else : <EOL> raise NotImplementedError ( ) <EOL> def extract_archive ( archive_path , destination_dir ) : <EOL> if archive_path . endswith ( "<STR_LIT>" ) : <EOL> check_call ( [ "<STR_LIT>" , "<STR_LIT>" , archive_path , "<STR_LIT>" , destination_dir ] ) <EOL> elif archive_path . endswith ( "<STR_LIT>" ) : <EOL> check_call ( [ "<STR_LIT>" , "<STR_LIT>" , archive_path , "<STR_LIT>" , destination_dir ] ) <EOL> else : <EOL> raise NotImplementedError ( ) <EOL> def get_packages_dir ( ) : <EOL> return join ( config [ '<STR_LIT>' ] , "<STR_LIT>" ) <EOL> def get_package_url ( package_name ) : <EOL> return "<STR_LIT>" % ( config [ '<STR_LIT>' ] , package_name ) <EOL> def get_libarch_relpath ( lib_id , version_id ) : <EOL> lib_id = int ( lib_id ) <EOL> version_id = int ( version_id ) <EOL> assert lib_id > <NUM_LIT:0> and version_id > <NUM_LIT:0> <EOL> return join ( "<STR_LIT>" , "<STR_LIT>" , str ( int ( ceil ( lib_id / <NUM_LIT:100> ) ) ) , <EOL> "<STR_LIT>" % version_id ) <EOL> def get_libarch_path ( lib_id , version_id ) : <EOL> return join ( config [ '<STR_LIT>' ] , get_libarch_relpath ( lib_id , version_id ) ) <EOL> def get_libarch_url ( lib_id , version_id ) : <EOL> return "<STR_LIT>" % ( config [ '<STR_LIT>' ] , <EOL> get_libarch_relpath ( lib_id , version_id ) ) <EOL> def get_libexample_relpath ( lib_id ) : <EOL> lib_id = int ( lib_id ) <EOL> assert lib_id > <NUM_LIT:0> <EOL> return join ( "<STR_LIT>" , "<STR_LIT>" , <EOL> str ( int ( ceil ( lib_id / <NUM_LIT:100> ) ) ) , str ( lib_id ) ) <EOL> def get_libexample_dir ( lib_id ) : <EOL> return join ( config [ '<STR_LIT>' ] , get_libexample_relpath ( lib_id ) ) <EOL> def get_libexample_url ( lib_id , name ) : <EOL> return "<STR_LIT>" % ( config [ '<STR_LIT>' ] , get_libexample_relpath ( lib_id ) , <EOL> name ) <EOL> def validate_libconf ( data ) : <EOL> fields = set ( data . keys ( ) ) <EOL> if not fields . issuperset ( set ( [ "<STR_LIT:name>" , "<STR_LIT>" , "<STR_LIT:description>" ] ) ) : <EOL> raise InvalidLibConf ( <EOL> "<STR_LIT>" ) <EOL> if ( "<STR_LIT>" in data and not <EOL> ( isinstance ( data [ '<STR_LIT>' ] , list ) or <EOL> isinstance ( data [ '<STR_LIT>' ] , dict ) ) ) : <EOL> raise InvalidLibConf ( "<STR_LIT>" ) <EOL> if "<STR_LIT>" in data : <EOL> type = data [ '<STR_LIT>' ] . get ( "<STR_LIT:type>" , None ) <EOL> url = data [ '<STR_LIT>' ] . get ( "<STR_LIT:url>" , "<STR_LIT>" ) <EOL> if ( type == "<STR_LIT>" and "<STR_LIT>" in url ) or ( type == "<STR_LIT>" and "<STR_LIT>" in url ) or ( type in [ "<STR_LIT>" , "<STR_LIT>" ] and "<STR_LIT>" in url ) : <EOL> return data <EOL> authors = data . get ( "<STR_LIT>" , None ) <EOL> if authors and not isinstance ( authors , list ) : <EOL> authors = [ authors ] <EOL> if not authors : <EOL> raise InvalidLibConf ( "<STR_LIT>" ) <EOL> elif not all ( [ "<STR_LIT:name>" in item for item in authors ] ) : <EOL> raise InvalidLibConf ( "<STR_LIT>" ) <EOL> elif ( "<STR_LIT>" in data and <EOL> data [ '<STR_LIT>' ] . get ( "<STR_LIT:type>" , None ) in ( "<STR_LIT>" , "<STR_LIT>" ) ) : <EOL> return data <EOL> if "<STR_LIT:version>" not in data : <EOL> raise InvalidLibConf ( "<STR_LIT>" ) <EOL> elif "<STR_LIT>" not in data : <EOL> raise InvalidLibConf ( "<STR_LIT>" ) <EOL> return data <EOL> def get_c_sources ( in_dir ) : <EOL> return glob ( join ( in_dir , '<STR_LIT>' ) ) + glob ( join ( in_dir , '<STR_LIT>' ) ) + glob ( join ( in_dir , '<STR_LIT>' ) ) <EOL> @ contextmanager <EOL> def rollback_on_exception ( session , logger = None ) : <EOL> try : <EOL> yield <EOL> except Exception as e : <EOL> session . rollback ( ) <EOL> if logger is not None : <EOL> logger . exception ( e ) <EOL> def rollback_on_exception_decorator ( session , logger = None ) : <EOL> def actual_decorator ( f ) : <EOL> def wrapped ( * args , ** kwargs ) : <EOL> with rollback_on_exception ( session , logger ) : <EOL> f ( * args , ** kwargs ) <EOL> return wrapped <EOL> return actual_decorator <EOL> def parse_namedtitled_list ( ntlist ) : <EOL> items = [ ] <EOL> for item in ntlist . split ( "<STR_LIT:U+002C>" ) : <EOL> if "<STR_LIT::>" in item : <EOL> items . append ( item . split ( "<STR_LIT::>" ) [ <NUM_LIT:0> ] ) <EOL> return items </s>
