signature:
def ToJsonString ( self ) :
    """Converts Duration to string format .
 Returns :
 A string converted from self . The string format will contains
 3 , 6 , or 9 fractional digits depending on the precision required to
 represent the exact Duration value . For example : " 1s " , " 1.010s " ,
 "1.00000100s " , " - 3.100s " """

implementation:
if self . seconds < 0 or self . nanos < 0 :
    result = '-'
    seconds = - self . seconds + int ( ( 0 - self . nanos ) // 1e9 )
    nanos = ( 0 - self . nanos ) % 1e9
else :
    result = ''
    seconds = self . seconds + int ( self . nanos // 1e9 )
    nanos = self . nanos % 1e9
result += '%d' % seconds
if ( nanos % 1e9 ) == 0 : # If there are 0 fractional digits , the fractional
# point ' . ' should be omitted when serializing .
    return result + 's'
if ( nanos % 1e6 ) == 0 : # Serialize 3 fractional digits .
    return result + '.%03ds' % ( nanos / 1e6 )
if ( nanos % 1e3 ) == 0 : # Serialize 6 fractional digits .
    return result + '.%06ds' % ( nanos / 1e3 )
# Serialize 9 fractional digits .
return result + '.%09ds' % nanos


signature:
def FromJsonString ( self , value ) :
    """Converts a string to Duration .
 Args :
 value : A string to be converted . The string must end with ' s ' . Any
 fractional digits ( or none ) are accepted as long as they fit into
 precision . For example : " 1s " , " 1.01s " , " 1.000001s " , " - 3.100s
 Raises :
 ParseError : On parsing problems ."""

implementation:
if len ( value ) < 1 or value [ - 1 ] != 's' :
    raise ParseError ( 'Duration must end with letter "s": {0}.' . format ( value ) )
try :
    pos = value . find ( '.' )
    if pos == - 1 :
        self . seconds = int ( value [ : - 1 ] )
        self . nanos = 0
    else :
        self . seconds = int ( value [ : pos ] )
        if value [ 0 ] == '-' :
            self . nanos = int ( round ( float ( '-0{0}' . format ( value [ pos : - 1 ] ) ) * 1e9 ) )
        else :
            self . nanos = int ( round ( float ( '0{0}' . format ( value [ pos : - 1 ] ) ) * 1e9 ) )
except ValueError :
    raise ParseError ( 'Couldn\'t parse duration: {0}.' . format ( value ) )


signature:
def FromJsonString ( self , value ) :
    """Converts string to FieldMask according to proto3 JSON spec ."""

implementation:
self . Clear ( )
for path in value . split ( ',' ) :
    self . paths . append ( path )


signature:
def get_doc ( doc_id , db_name , server_url = 'http://127.0.0.1:5984/' , rev = None ) :
    """Return a CouchDB document , given its ID , revision and database name ."""

implementation:
db = get_server ( server_url ) [ db_name ]
if rev :
    headers , response = db . resource . get ( doc_id , rev = rev )
    return couchdb . client . Document ( response )
return db [ doc_id ]


signature:
def get_or_create_db ( db_name , server_url = 'http://127.0.0.1:5984/' ) :
    """Return an ( optionally existing ) CouchDB database instance ."""

implementation:
server = get_server ( server_url )
if db_name in server :
    return server [ db_name ]
return server . create ( db_name )


signature:
def read ( readme ) :
    """Give reST format README for pypi ."""

implementation:
extend = os . path . splitext ( readme ) [ 1 ]
if ( extend == '.rst' ) :
    import codecs
    return codecs . open ( readme , 'r' , 'utf-8' ) . read ( )
elif ( extend == '.md' ) :
    import pypandoc
    return pypandoc . convert ( readme , 'rst' )


signature:
def main ( ) :
    """Register your own mode and handle method here ."""

implementation:
plugin = Register ( )
if plugin . args . option == 'sql' :
    plugin . sql_handle ( )
elif plugin . args . option == 'database-used' :
    plugin . database_used_handle ( )
elif plugin . args . option == 'databaselog-used' :
    plugin . database_log_used_handle ( )
else :
    plugin . unknown ( "Unknown actions." )


signature:
def parse ( self , args ) :
    """: param args : arguments
 : type args : None or string or list of string
 : return : formatted arguments if specified else ` ` self . default _ args ` `
 : rtype : list of string"""

implementation:
if args is None :
    args = self . _default_args
if isinstance ( args , six . string_types ) :
    args = shlex . split ( args )
return args


signature:
def _request ( self , method , * relative_path_parts , ** kwargs ) :
    """Sends an HTTP request to the REST API and receives the requested data .
 : param str method : HTTP method name
 : param relative _ path _ parts : the relative paths for the request URI
 : param kwargs : argument keywords
 : returns : requested data
 : raises APIError : for non - 2xx responses"""

implementation:
uri = self . _create_api_uri ( * relative_path_parts )
if method == 'get' :
    response = get ( uri , auth = self . auth , params = kwargs . get ( 'params' , None ) )
elif method == 'post' :
    response = post ( uri , auth = self . auth , json = kwargs . get ( 'data' , None ) )
else :
    response = delete ( uri , auth = self . auth , json = kwargs . get ( 'data' , None ) )
return self . _handle_response ( response ) . json ( )


signature:
def _place_order ( self , side , product_id = 'BTC-USD' , client_oid = None , type = None , stp = None , price = None , size = None , funds = None , time_in_force = None , cancel_after = None , post_only = None ) :
    """` < https : / / docs . exchange . coinbase . com / # orders > ` _"""

implementation:
data = { 'side' : side , 'product_id' : product_id , 'client_oid' : client_oid , 'type' : type , 'stp' : stp , 'price' : price , 'size' : size , 'funds' : funds , 'time_in_force' : time_in_force , 'cancel_after' : cancel_after , 'post_only' : post_only }
return self . _post ( 'orders' , data = data )


signature:
def place_limit_order ( self , side , price , size , product_id = 'BTC-USD' , client_oid = None , stp = None , time_in_force = None , cancel_after = None , post_only = None ) :
    """` < https : / / docs . exchange . coinbase . com / # orders > ` _"""

implementation: return self . _place_order ( side , product_id = product_id , client_oid = client_oid , type = 'limit' , stp = stp , price = price , size = size , time_in_force = time_in_force , cancel_after = cancel_after , post_only = post_only )


signature:
def place_market_order ( self , side , product_id = 'BTC-USD' , size = None , funds = None , client_oid = None , stp = None ) :
    """` < https : / / docs . exchange . coinbase . com / # orders > ` _"""

implementation: return self . _place_order ( type = 'market' , side = size , product_id = product_id , size = size , funds = funds , client_oid = client_oid , stp = stp )


signature:
def _deposit_withdraw ( self , type , amount , coinbase_account_id ) :
    """` < https : / / docs . exchange . coinbase . com / # depositwithdraw > ` _"""

implementation:
data = { 'type' : type , 'amount' : amount , 'coinbase_account_id' : coinbase_account_id }
return self . _post ( 'transfers' , data = data )


signature:
def _new_report ( self , type , start_date , end_date , product_id = 'BTC-USD' , account_id = None , format = None , email = None ) :
    """` < https : / / docs . exchange . coinbase . com / # create - a - new - report > ` _"""

implementation:
data = { 'type' : type , 'start_date' : self . _format_iso_time ( start_date ) , 'end_date' : self . _format_iso_time ( end_date ) , 'product_id' : product_id , 'account_id' : account_id , 'format' : format , 'email' : email }
return self . _post ( 'reports' , data = data )


signature:
def new_fills_report ( self , start_date , end_date , account_id = None , product_id = 'BTC-USD' , format = None , email = None ) :
    """` < https : / / docs . exchange . coinbase . com / # create - a - new - report > ` _"""

implementation: return self . _new_report ( start_date , 'fills' , end_date , account_id , product_id , format , email )


signature:
def _request ( self , method , * relative_path_parts , ** kwargs ) :
    """Sends an HTTP request to the REST API and receives the requested data .
 Additionally sets up pagination cursors .
 : param str method : HTTP method name
 : param relative _ path _ parts : the relative paths for the request URI
 : param kwargs : argument keywords
 : returns : requested data
 : raises APIError : for non - 2xx responses"""

implementation:
uri = self . _create_api_uri ( * relative_path_parts )
if method == 'get' :
    response = get ( uri , auth = self . auth , params = kwargs . get ( 'params' , None ) )
elif method == 'post' :
    response = post ( uri , auth = self . auth , json = kwargs . get ( 'data' , None ) )
else :
    response = delete ( uri , auth = self . auth , json = kwargs . get ( 'data' , None ) )
self . is_initial = False
self . before_cursor = response . headers . get ( 'cb-before' , None )
self . after_cursor = response . headers . get ( 'cb-after' , None )
return self . _handle_response ( response ) . json ( )


signature:
def fetch ( self , collection , ** kwargs ) :
    '''return one record from the collection whose parameters match kwargs
 kwargs should be a dictionary whose keys match column names ( in
 traditional SQL / fields in NoSQL ) and whose values are the values of
 those fields .
 e . g . kwargs = { name = ' my application name ' , client _ id = 12345}'''

implementation:
callback = kwargs . pop ( 'callback' )
data = yield Op ( self . db [ collection ] . find_one , kwargs )
callback ( data )


signature:
def remove ( self , collection , ** kwargs ) :
    '''remove records from collection whose parameters match kwargs'''

implementation:
callback = kwargs . pop ( 'callback' )
yield Op ( self . db [ collection ] . remove , kwargs )
callback ( )


signature:
def store ( self , collection , ** kwargs ) :
    '''validate the passed values in kwargs based on the collection ,
 store them in the mongodb collection'''

implementation:
callback = kwargs . pop ( 'callback' )
key = validate ( collection , ** kwargs )
data = yield Task ( self . fetch , collection , ** { key : kwargs [ key ] } )
if data is not None :
    raise Proauth2Error ( 'duplicate_key' )
yield Op ( self . db [ collection ] . insert , kwargs )
callback ( )


signature:
def generate_api ( version ) :
    """Generates a factory function to instantiate the API with the given
 version ."""

implementation:
def get_partial_api ( key , token = None ) :
    return TrelloAPI ( ENDPOINTS [ version ] , version , key , token = token )
get_partial_api . __doc__ = """Interfaz REST con Trello. Versión {}""" . format ( version )
return get_partial_api


signature:
def _url ( self ) :
    """Resolve the URL to this point .
 > > > trello = TrelloAPIV1 ( ' APIKEY ' )
 > > > trello . batch . _ url
 '1 / batch '
 > > > trello . boards ( board _ id = ' BOARD _ ID ' ) . _ url
 '1 / boards / BOARD _ ID '
 > > > trello . boards ( board _ id = ' BOARD _ ID ' ) ( field = ' FIELD ' ) . _ url
 '1 / boards / BOARD _ ID / FIELD '
 > > > trello . boards ( board _ id = ' BOARD _ ID ' ) . cards ( filter = ' FILTER ' ) . _ url
 '1 / boards / BOARD _ ID / cards / FILTER '"""

implementation:
if self . _api_arg :
    mypart = str ( self . _api_arg )
else :
    mypart = self . _name
if self . _parent :
    return '/' . join ( filter ( None , [ self . _parent . _url , mypart ] ) )
else :
    return mypart


signature:
def _api_call ( self , method_name , * args , ** kwargs ) :
    """Makes the HTTP request ."""

implementation:
params = kwargs . setdefault ( 'params' , { } )
params . update ( { 'key' : self . _apikey } )
if self . _token is not None :
    params . update ( { 'token' : self . _token } )
http_method = getattr ( requests , method_name )
return http_method ( TRELLO_URL + self . _url , * args , ** kwargs )


signature:
def Merge ( text , message , allow_unknown_extension = False , allow_field_number = False ) :
    """Parses an text representation of a protocol message into a message .
 Like Parse ( ) , but allows repeated values for a non - repeated field , and uses
 the last one .
 Args :
 text : Message text representation .
 message : A protocol buffer message to merge into .
 allow _ unknown _ extension : if True , skip over missing extensions and keep
 parsing
 allow _ field _ number : if True , both field number and field name are allowed .
 Returns :
 The same message passed as argument .
 Raises :
 ParseError : On text parsing problems ."""

implementation: return MergeLines ( text . split ( '\n' ) , message , allow_unknown_extension , allow_field_number )


signature:
def ParseLines ( lines , message , allow_unknown_extension = False , allow_field_number = False ) :
    """Parses an text representation of a protocol message into a message .
 Args :
 lines : An iterable of lines of a message ' s text representation .
 message : A protocol buffer message to merge into .
 allow _ unknown _ extension : if True , skip over missing extensions and keep
 parsing
 allow _ field _ number : if True , both field number and field name are allowed .
 Returns :
 The same message passed as argument .
 Raises :
 ParseError : On text parsing problems ."""

implementation:
parser = _Parser ( allow_unknown_extension , allow_field_number )
return parser . ParseLines ( lines , message )


signature:
def _SkipFieldValue ( tokenizer ) :
    """Skips over a field value .
 Args :
 tokenizer : A tokenizer to parse the field name and values .
 Raises :
 ParseError : In case an invalid field value is found ."""

implementation:
# String / bytes tokens can come in multiple adjacent string literals .
# If we can consume one , consume as many as we can .
if tokenizer . TryConsumeByteString ( ) :
    while tokenizer . TryConsumeByteString ( ) :
        pass
    return
if ( not tokenizer . TryConsumeIdentifier ( ) and not tokenizer . TryConsumeInt64 ( ) and not tokenizer . TryConsumeUint64 ( ) and not tokenizer . TryConsumeFloat ( ) ) :
    raise ParseError ( 'Invalid field value: ' + tokenizer . token )


signature:
def ParseInteger ( text , is_signed = False , is_long = False ) :
    """Parses an integer .
 Args :
 text : The text to parse .
 is _ signed : True if a signed integer must be parsed .
 is _ long : True if a long integer must be parsed .
 Returns :
 The integer value .
 Raises :
 ValueError : Thrown Iff the text is not a valid integer ."""

implementation:
# Do the actual parsing . Exception handling is propagated to caller .
try : # We force 32 - bit values to int and 64 - bit values to long to make
# alternate implementations where the distinction is more significant
# ( e . g . the C + + implementation ) simpler .
    if is_long :
        result = long ( text , 0 )
    else :
        result = int ( text , 0 )
except ValueError :
    raise ValueError ( 'Couldn\'t parse integer: %s' % text )
# Check if the integer is sane . Exceptions handled by callers .
checker = _INTEGER_CHECKERS [ 2 * int ( is_long ) + int ( is_signed ) ]
checker . CheckValue ( result )
return result


signature:
def PrintMessage ( self , message ) :
    """Convert protobuf message to text format .
 Args :
 message : The protocol buffers message ."""

implementation:
fields = message . ListFields ( )
if self . use_index_order :
    fields . sort ( key = lambda x : x [ 0 ] . index )
for field , value in fields :
    if _IsMapEntry ( field ) :
        for key in sorted ( value ) : # This is slow for maps with submessage entires because it copies the
        # entire tree . Unfortunately this would take significant refactoring
        # of this file to work around .
        # TODO ( haberman ) : refactor and optimize if this becomes an issue .
            entry_submsg = field . message_type . _concrete_class ( key = key , value = value [ key ] )
            self . PrintField ( field , entry_submsg )
    elif field . label == descriptor . FieldDescriptor . LABEL_REPEATED :
        for element in value :
            self . PrintField ( field , element )
    else :
        self . PrintField ( field , value )


signature:
def PrintFieldValue ( self , field , value ) :
    """Print a single field value ( not including name ) .
 For repeated fields , the value should be a single element .
 Args :
 field : The descriptor of the field to be printed .
 value : The value of the field ."""

implementation:
out = self . out
if self . pointy_brackets :
    openb = '<'
    closeb = '>'
else :
    openb = '{'
    closeb = '}'
if field . cpp_type == descriptor . FieldDescriptor . CPPTYPE_MESSAGE :
    if self . as_one_line :
        out . write ( ' %s ' % openb )
        self . PrintMessage ( value )
        out . write ( closeb )
    else :
        out . write ( ' %s\n' % openb )
        self . indent += 2
        self . PrintMessage ( value )
        self . indent -= 2
        out . write ( ' ' * self . indent + closeb )
elif field . cpp_type == descriptor . FieldDescriptor . CPPTYPE_ENUM :
    enum_value = field . enum_type . values_by_number . get ( value , None )
    if enum_value is not None :
        out . write ( enum_value . name )
    else :
        out . write ( str ( value ) )
elif field . cpp_type == descriptor . FieldDescriptor . CPPTYPE_STRING :
    out . write ( '\"' )
    if isinstance ( value , six . text_type ) :
        out_value = value . encode ( 'utf-8' )
    else :
        out_value = value
    if field . type == descriptor . FieldDescriptor . TYPE_BYTES : # We need to escape non - UTF8 chars in TYPE _ BYTES field .
        out_as_utf8 = False
    else :
        out_as_utf8 = self . as_utf8
    out . write ( text_encoding . CEscape ( out_value , out_as_utf8 ) )
    out . write ( '\"' )
elif field . cpp_type == descriptor . FieldDescriptor . CPPTYPE_BOOL :
    if value :
        out . write ( 'true' )
    else :
        out . write ( 'false' )
elif field . cpp_type in _FLOAT_TYPES and self . float_format is not None :
    out . write ( '{1:{0}}' . format ( self . float_format , value ) )
else :
    out . write ( str ( value ) )


signature:
def _ParseOrMerge ( self , lines , message ) :
    """Converts an text representation of a protocol message into a message .
 Args :
 lines : Lines of a message ' s text representation .
 message : A protocol buffer message to merge into .
 Raises :
 ParseError : On text parsing problems ."""

implementation:
tokenizer = _Tokenizer ( lines )
while not tokenizer . AtEnd ( ) :
    self . _MergeField ( tokenizer , message )


signature:
def _MergeMessageField ( self , tokenizer , message , field ) :
    """Merges a single scalar field into a message .
 Args :
 tokenizer : A tokenizer to parse the field value .
 message : The message of which field is a member .
 field : The descriptor of the field to be merged .
 Raises :
 ParseError : In case of text parsing problems ."""

implementation:
is_map_entry = _IsMapEntry ( field )
if tokenizer . TryConsume ( '<' ) :
    end_token = '>'
else :
    tokenizer . Consume ( '{' )
    end_token = '}'
if field . label == descriptor . FieldDescriptor . LABEL_REPEATED :
    if field . is_extension :
        sub_message = message . Extensions [ field ] . add ( )
    elif is_map_entry : # pylint : disable = protected - access
        sub_message = field . message_type . _concrete_class ( )
    else :
        sub_message = getattr ( message , field . name ) . add ( )
else :
    if field . is_extension :
        sub_message = message . Extensions [ field ]
    else :
        sub_message = getattr ( message , field . name )
    sub_message . SetInParent ( )
while not tokenizer . TryConsume ( end_token ) :
    if tokenizer . AtEnd ( ) :
        raise tokenizer . ParseErrorPreviousToken ( 'Expected "%s".' % ( end_token , ) )
    self . _MergeField ( tokenizer , sub_message )
if is_map_entry :
    value_cpptype = field . message_type . fields_by_name [ 'value' ] . cpp_type
    if value_cpptype == descriptor . FieldDescriptor . CPPTYPE_MESSAGE :
        value = getattr ( message , field . name ) [ sub_message . key ]
        value . MergeFrom ( sub_message . value )
    else :
        getattr ( message , field . name ) [ sub_message . key ] = sub_message . value


signature:
def ConsumeIdentifier ( self ) :
    """Consumes protocol message field identifier .
 Returns :
 Identifier string .
 Raises :
 ParseError : If an identifier couldn ' t be consumed ."""

implementation:
result = self . token
if not self . _IDENTIFIER . match ( result ) :
    raise self . _ParseError ( 'Expected identifier.' )
self . NextToken ( )
return result


signature:
def ConsumeInt32 ( self ) :
    """Consumes a signed 32bit integer number .
 Returns :
 The integer parsed .
 Raises :
 ParseError : If a signed 32bit integer couldn ' t be consumed ."""

implementation:
try :
    result = ParseInteger ( self . token , is_signed = True , is_long = False )
except ValueError as e :
    raise self . _ParseError ( str ( e ) )
self . NextToken ( )
return result


signature:
def ConsumeFloat ( self ) :
    """Consumes an floating point number .
 Returns :
 The number parsed .
 Raises :
 ParseError : If a floating point number couldn ' t be consumed ."""

implementation:
try :
    result = ParseFloat ( self . token )
except ValueError as e :
    raise self . _ParseError ( str ( e ) )
self . NextToken ( )
return result


signature:
def ConsumeBool ( self ) :
    """Consumes a boolean value .
 Returns :
 The bool parsed .
 Raises :
 ParseError : If a boolean value couldn ' t be consumed ."""

implementation:
try :
    result = ParseBool ( self . token )
except ValueError as e :
    raise self . _ParseError ( str ( e ) )
self . NextToken ( )
return result


signature:
def _ConsumeSingleByteString ( self ) :
    """Consume one token of a string literal .
 String literals ( whether bytes or text ) can come in multiple adjacent
 tokens which are automatically concatenated , like in C or Python . This
 method only consumes one token .
 Returns :
 The token parsed .
 Raises :
 ParseError : When the wrong format data is found ."""

implementation:
text = self . token
if len ( text ) < 1 or text [ 0 ] not in _QUOTES :
    raise self . _ParseError ( 'Expected string but found: %r' % ( text , ) )
if len ( text ) < 2 or text [ - 1 ] != text [ 0 ] :
    raise self . _ParseError ( 'String missing ending quote: %r' % ( text , ) )
try :
    result = text_encoding . CUnescape ( text [ 1 : - 1 ] )
except ValueError as e :
    raise self . _ParseError ( str ( e ) )
self . NextToken ( )
return result


signature:
def timestamp ( t = None , forfilename = False ) :
    """Returns a human - readable timestamp given a Unix timestamp ' t ' or
 for the current time . The Unix timestamp is the number of seconds since
 start of epoch ( 1970-01-01 00:00:00 ) .
 When forfilename is True , then spaces and semicolons are replace with
 hyphens . The returned string is usable as a ( part of a ) filename ."""

implementation:
datetimesep = ' '
timesep = ':'
if forfilename :
    datetimesep = '-'
    timesep = '-'
return time . strftime ( '%Y-%m-%d' + datetimesep + '%H' + timesep + '%M' + timesep + '%S' , time . localtime ( t ) )


signature:
def arktimestamp ( arkt , forfilename = False ) :
    """Returns a human - readable timestamp given an Ark timestamp ' arct ' .
 An Ark timestamp is the number of seconds since Genesis block ,
 2017:03:21 15:55:44."""


signature:
def arkt_to_unixt ( ark_timestamp ) :
    """convert ark timestamp to unix timestamp"""

implementation:
res = datetime . datetime ( 2017 , 3 , 21 , 15 , 55 , 44 ) + datetime . timedelta ( seconds = ark_timestamp )
return res . timestamp ( )


signature:
def close ( self ) :
    """Close the connection ."""

implementation:
try :
    self . conn . close ( )
    self . logger . debug ( "Close connect succeed." )
except pymssql . Error as e :
    self . unknown ( "Close connect error: %s" % e )


signature:
def get_version ( ) :
    """Extract package _ _ version _ _"""

implementation:
with open ( VERSION_FILE , encoding = 'utf-8' ) as fp :
    content = fp . read ( )
match = re . search ( r'^__version__ = [\'"]([^\'"]*)[\'"]' , content , re . M )
if match :
    return match . group ( 1 )
raise RuntimeError ( "Could not extract package __version__" )


signature:
def process_macros ( self , content : str ) -> str :
    '''Replace macros with content defined in the config .
 : param content : Markdown content
 : returns : Markdown content without macros'''

implementation:
def _sub ( macro ) :
    name = macro . group ( 'body' )
    params = self . get_options ( macro . group ( 'options' ) )
    return self . options [ 'macros' ] . get ( name , '' ) . format_map ( params )
return self . pattern . sub ( _sub , content )


signature:
def _request ( self , method , * relative_path_parts , ** kwargs ) :
    """Sends an HTTP request to the REST API and receives the requested data .
 : param str method : HTTP method name
 : param relative _ path _ parts : the relative paths for the request URI
 : param kwargs : argument keywords
 : returns : requested data
 : raises APIError : for non - 2xx responses"""

implementation:
uri = self . _create_api_uri ( * relative_path_parts )
response = get ( uri , params = kwargs . get ( 'params' , None ) )
return self . _handle_response ( response ) . json ( )


signature:
def get_historic_trades ( self , start , end , granularity , product_id = 'BTC-USD' ) :
    """` < https : / / docs . exchange . coinbase . com / # get - historic - rates > ` _
 : param start : either datetime . datetime or str in ISO 8601
 : param end : either datetime . datetime or str in ISO 8601
 : pram int granularity : desired timeslice in seconds
 : returns : desired data"""

implementation:
params = { 'start' : self . _format_iso_time ( start ) , 'end' : self . _format_iso_time ( end ) , 'granularity' : granularity }
return self . _get ( 'products' , product_id , 'candles' , params = params )


signature:
def _request ( self , method , * relative_path_parts , ** kwargs ) :
    """Sends an HTTP request to the REST API and receives the requested data .
 Additionally sets up pagination cursors .
 : param str method : HTTP method name
 : param relative _ path _ parts : the relative paths for the request URI
 : param kwargs : argument keywords
 : returns : requested data
 : raises APIError : for non - 2xx responses"""

implementation:
uri = self . _create_api_uri ( * relative_path_parts )
response = get ( uri , params = self . _get_params ( ** kwargs ) )
self . is_initial = False
self . before_cursor = response . headers . get ( 'cb-before' , None )
self . after_cursor = response . headers . get ( 'cb-after' , None )
return self . _handle_response ( response ) . json ( )


signature:
def get_unique_pathname ( path , root = '' ) :
    """Return a pathname possibly with a number appended to it so that it is
 unique in the directory ."""

implementation:
path = os . path . join ( root , path )
# consider the path supplied , then the paths with numbers appended
potentialPaths = itertools . chain ( ( path , ) , __get_numbered_paths ( path ) )
potentialPaths = six . moves . filterfalse ( os . path . exists , potentialPaths )
return next ( potentialPaths )


signature:
def __get_numbered_paths ( filepath ) :
    """Append numbers in sequential order to the filename or folder name
 Numbers should be appended before the extension on a filename ."""

implementation:
format = '%s (%%d)%s' % splitext_files_only ( filepath )
return map ( lambda n : format % n , itertools . count ( 1 ) )


signature:
def splitext_files_only ( filepath ) :
    "Custom version of splitext that doesn ' t perform splitext on directories"

implementation: return ( ( filepath , '' ) if os . path . isdir ( filepath ) else os . path . splitext ( filepath ) )


signature:
def set_time ( filename , mod_time ) :
    """Set the modified time of a file"""

implementation:
log . debug ( 'Setting modified time to %s' , mod_time )
mtime = calendar . timegm ( mod_time . utctimetuple ( ) )
# utctimetuple discards microseconds , so restore it ( for consistency )
mtime += mod_time . microsecond / 1000000
atime = os . stat ( filename ) . st_atime
os . utime ( filename , ( atime , mtime ) )


signature:
def get_time ( filename ) :
    """Get the modified time for a file as a datetime instance"""

implementation:
ts = os . stat ( filename ) . st_mtime
return datetime . datetime . utcfromtimestamp ( ts )


signature:
def insert_before_extension ( filename , content ) :
    """Given a filename and some content , insert the content just before
 the extension .
 > > > insert _ before _ extension ( ' pages . pdf ' , ' - old ' )
 ' pages - old . pdf '"""

implementation:
parts = list ( os . path . splitext ( filename ) )
parts [ 1 : 1 ] = [ content ]
return '' . join ( parts )


signature:
def recursive_glob ( root , spec ) :
    """Like iglob , but recurse directories
 > > > any ( ' path . py ' in result for result in recursive _ glob ( ' . ' , ' * . py ' ) )
 True
 > > > all ( result . startswith ( ' . ' ) for result in recursive _ glob ( ' . ' , ' * . py ' ) )
 True
 > > > len ( list ( recursive _ glob ( ' . ' , ' * . foo ' ) ) )"""

implementation:
specs = ( os . path . join ( dirpath , dirname , spec ) for dirpath , dirnames , filenames in os . walk ( root ) for dirname in dirnames )
return itertools . chain . from_iterable ( glob . iglob ( spec ) for spec in specs )


signature:
def encode ( name , system = 'NTFS' ) :
    """Encode the name for a suitable name in the given filesystem
 > > > encode ( ' Test : 1 ' )
 ' Test _ 1'"""

implementation:
assert system == 'NTFS' , 'unsupported filesystem'
special_characters = r'<>:"/\|?*' + '' . join ( map ( chr , range ( 32 ) ) )
pattern = '|' . join ( map ( re . escape , special_characters ) )
pattern = re . compile ( pattern )
return pattern . sub ( '_' , name )


signature:
def ensure_dir_exists ( func ) :
    "wrap a function that returns a dir , making sure it exists"

implementation:
@ functools . wraps ( func )
def make_if_not_present ( ) :
    dir = func ( )
    if not os . path . isdir ( dir ) :
        os . makedirs ( dir )
    return dir
return make_if_not_present


signature:
def read_chunks ( file , chunk_size = 2048 , update_func = lambda x : None ) :
    """Read file in chunks of size chunk _ size ( or smaller ) .
 If update _ func is specified , call it on every chunk with the amount
 read ."""

implementation:
while ( True ) :
    res = file . read ( chunk_size )
    if not res :
        break
    update_func ( len ( res ) )
    yield res


signature:
def is_hidden ( path ) :
    """Check whether a file is presumed hidden , either because
 the pathname starts with dot or because the platform
 indicates such ."""

implementation:
full_path = os . path . abspath ( path )
name = os . path . basename ( full_path )
def no ( path ) :
    return False
platform_hidden = globals ( ) . get ( 'is_hidden_' + platform . system ( ) , no )
return name . startswith ( '.' ) or platform_hidden ( full_path )


signature:
def age ( self ) :
    """Get closer to your EOL"""

implementation:
# 0 means this composer will never decompose
if self . rounds == 1 :
    self . do_run = False
elif self . rounds > 1 :
    self . rounds -= 1


signature:
def run ( self ) :
    """Open a connection over the serial line and receive data lines"""

implementation:
if not self . device :
    return
try :
    data = ""
    while ( self . do_run ) :
        try :
            if ( self . device . inWaiting ( ) > 1 ) :
                l = self . device . readline ( ) [ : - 2 ]
                l = l . decode ( "UTF-8" )
                if ( l == "[" ) : # start recording
                    data = "["
                elif ( l == "]" ) and ( len ( data ) > 4 ) and ( data [ 0 ] == "[" ) : # now parse the input
                    data = data + "]"
                    self . store . register_json ( data )
                    self . age ( )
                elif ( l [ 0 : 3 ] == "  {" ) : # this is a data line
                    data = data + " " + l
            else : # this is a slow interface - give it some time
                sleep ( 1 )
                # then count down . .
                self . age ( )
        except ( UnicodeDecodeError , ValueError ) : # only accepting unicode : throw away the whole bunch
            data = ""
            # and count down the exit condition
            self . age ( )
except serial . serialutil . SerialException :
    print ( "Could not connect to the serial line at " + self . device_name )


signature:
def append_main_thread ( self ) :
    """create & start main thread
 : return : None"""

implementation:
thread = MainThread ( main_queue = self . main_queue , main_spider = self . main_spider , branch_spider = self . branch_spider )
thread . daemon = True
thread . start ( )


signature:
def getTextFromNode ( node ) :
    """Scans through all children of node and gathers the
 text . If node has non - text child - nodes then
 NotTextNodeError is raised ."""

implementation:
t = ""
for n in node . childNodes :
    if n . nodeType == n . TEXT_NODE :
        t += n . nodeValue
    else :
        raise NotTextNodeError
return t


signature:
def getbalance ( self , url = 'http://services.ambientmobile.co.za/credits' ) :
    """Get the number of credits remaining at AmbientSMS"""

implementation:
postXMLList = [ ]
postXMLList . append ( "<api-key>%s</api-key>" % self . api_key )
postXMLList . append ( "<password>%s</password>" % self . password )
postXML = '<sms>%s</sms>' % "" . join ( postXMLList )
result = self . curl ( url , postXML )
if result . get ( "credits" , None ) :
    return result [ "credits" ]
else :
    raise AmbientSMSError ( result [ "status" ] )


signature:
def sendmsg ( self , message , recipient_mobiles = [ ] , url = 'http://services.ambientmobile.co.za/sms' , concatenate_message = True , message_id = str ( time ( ) ) . replace ( "." , "" ) , reply_path = None , allow_duplicates = True , allow_invalid_numbers = True , ) :
    """Send a mesage via the AmbientSMS API server"""

implementation:
if not recipient_mobiles or not ( isinstance ( recipient_mobiles , list ) or isinstance ( recipient_mobiles , tuple ) ) :
    raise AmbientSMSError ( "Missing recipients" )
if not message or not len ( message ) :
    raise AmbientSMSError ( "Missing message" )
postXMLList = [ ]
postXMLList . append ( "<api-key>%s</api-key>" % self . api_key )
postXMLList . append ( "<password>%s</password>" % self . password )
postXMLList . append ( "<recipients>%s</recipients>" % "" . join ( [ "<mobile>%s</mobile>" % m for m in recipient_mobiles ] ) )
postXMLList . append ( "<msg>%s</msg>" % message )
postXMLList . append ( "<concat>%s</concat>" % ( 1 if concatenate_message else 0 ) )
postXMLList . append ( "<message_id>%s</message_id>" % message_id )
postXMLList . append ( "<allow_duplicates>%s</allow_duplicates>" % ( 1 if allow_duplicates else 0 ) )
postXMLList . append ( "<allow_invalid_numbers>%s</allow_invalid_numbers>" % ( 1 if allow_invalid_numbers else 0 ) )
if reply_path :
    postXMLList . append ( "<reply_path>%s</reply_path>" % reply_path )
postXML = '<sms>%s</sms>' % "" . join ( postXMLList )
result = self . curl ( url , postXML )
status = result . get ( "status" , None )
if status and int ( status ) in [ 0 , 1 , 2 ] :
    return result
else :
    raise AmbientSMSError ( int ( status ) )


signature:
def curl ( self , url , post ) :
    """Inteface for sending web requests to the AmbientSMS API Server"""

implementation:
try :
    req = urllib2 . Request ( url )
    req . add_header ( "Content-type" , "application/xml" )
    data = urllib2 . urlopen ( req , post . encode ( 'utf-8' ) ) . read ( )
except urllib2 . URLError , v :
    raise AmbientSMSError ( v )
return dictFromXml ( data )


signature:
def execute ( self , correlation_id , args ) :
    """Executes the command given specific arguments as an input .
 Args :
 correlation _ id : a unique correlation / transaction id
 args : command arguments
 Returns : an execution result .
 Raises :
 MicroserviceError : when execution fails for whatever reason ."""

implementation: return self . _intercepter . execute ( _next , correlation_id , args )


signature:
def contents ( self , f , text ) :
    """Called for each file
 Must return file content
 Can be wrapped
 : type f : static _ bundle . files . StaticFileResult
 : type text : str | unicode
 : rtype : str | unicode"""

implementation:
text += self . _read ( f . abs_path ) + "\r\n"
return text


signature:
def is_date_type ( cls ) :
    """Return True if the class is a date type ."""

implementation:
if not isinstance ( cls , type ) :
    return False
return issubclass ( cls , date ) and not issubclass ( cls , datetime )


signature:
def to_datetime ( when ) :
    """Convert a date or time to a datetime . If when is a date then it sets the time to midnight . If
 when is a time it sets the date to the epoch . If when is None or a datetime it returns when .
 Otherwise a TypeError is raised . Returned datetimes have tzinfo set to None unless when is a
 datetime with tzinfo set in which case it remains the same ."""

implementation:
if when is None or is_datetime ( when ) :
    return when
if is_time ( when ) :
    return datetime . combine ( epoch . date ( ) , when )
if is_date ( when ) :
    return datetime . combine ( when , time ( 0 ) )
raise TypeError ( "unable to convert {} to datetime" . format ( when . __class__ . __name__ ) )


signature:
def totz ( when , tz = None ) :
    """Return a date , time , or datetime converted to a datetime in the given timezone . If when is a
 datetime and has no timezone it is assumed to be local time . Date and time objects are also
 assumed to be UTC . The tz value defaults to UTC . Raise TypeError if when cannot be converted to
 a datetime ."""

implementation:
if when is None :
    return None
when = to_datetime ( when )
if when . tzinfo is None :
    when = when . replace ( tzinfo = localtz )
return when . astimezone ( tz or utc )


signature:
def timeago ( tz = None , * args , ** kwargs ) :
    """Return a datetime so much time ago . Takes the same arguments as timedelta ( ) ."""

implementation: return totz ( datetime . now ( ) , tz ) - timedelta ( * args , ** kwargs )


signature:
def ts ( when , tz = None ) :
    """Return a Unix timestamp in seconds for the provided datetime . The ` totz ` function is called
 on the datetime to convert it to the provided timezone . It will be converted to UTC if no
 timezone is provided ."""

implementation:
if not when :
    return None
when = totz ( when , tz )
return calendar . timegm ( when . timetuple ( ) )


signature:
def tsms ( when , tz = None ) :
    """Return a Unix timestamp in milliseconds for the provided datetime . The ` totz ` function is
 called on the datetime to convert it to the provided timezone . It will be converted to UTC if
 no timezone is provided ."""

implementation:
if not when :
    return None
when = totz ( when , tz )
return calendar . timegm ( when . timetuple ( ) ) * 1000 + int ( round ( when . microsecond / 1000.0 ) )


signature:
def fromts ( ts , tzin = None , tzout = None ) :
    """Return the datetime representation of the provided Unix timestamp . By defaults the timestamp is
 interpreted as UTC . If tzin is set it will be interpreted as this timestamp instead . By default
 the output datetime will have UTC time . If tzout is set it will be converted in this timezone
 instead ."""

implementation:
if ts is None :
    return None
when = datetime . utcfromtimestamp ( ts ) . replace ( tzinfo = tzin or utc )
return totz ( when , tzout )


signature:
def fromtsms ( ts , tzin = None , tzout = None ) :
    """Return the Unix timestamp in milliseconds as a datetime object . If tz is set it will be
 converted to the requested timezone otherwise it defaults to UTC ."""

implementation:
if ts is None :
    return None
when = datetime . utcfromtimestamp ( ts / 1000 ) . replace ( microsecond = ts % 1000 * 1000 )
when = when . replace ( tzinfo = tzin or utc )
return totz ( when , tzout )


signature:
def truncate ( when , unit , week_start = mon ) :
    """Return the datetime truncated to the precision of the provided unit ."""

implementation:
if is_datetime ( when ) :
    if unit == millisecond :
        return when . replace ( microsecond = int ( round ( when . microsecond / 1000.0 ) ) * 1000 )
    elif unit == second :
        return when . replace ( microsecond = 0 )
    elif unit == minute :
        return when . replace ( second = 0 , microsecond = 0 )
    elif unit == hour :
        return when . replace ( minute = 0 , second = 0 , microsecond = 0 )
    elif unit == day :
        return when . replace ( hour = 0 , minute = 0 , second = 0 , microsecond = 0 )
    elif unit == week :
        weekday = prevweekday ( when , week_start )
        return when . replace ( year = weekday . year , month = weekday . month , day = weekday . day , hour = 0 , minute = 0 , second = 0 , microsecond = 0 )
    elif unit == month :
        return when . replace ( day = 1 , hour = 0 , minute = 0 , second = 0 , microsecond = 0 )
    elif unit == year :
        return when . replace ( month = 1 , day = 1 , hour = 0 , minute = 0 , second = 0 , microsecond = 0 )
elif is_date ( when ) :
    if unit == week :
        return prevweekday ( when , week_start )
    elif unit == month :
        return when . replace ( day = 1 )
    elif unit == year :
        return when . replace ( month = 1 , day = 1 )
elif is_time ( when ) :
    if unit == millisecond :
        return when . replace ( microsecond = int ( when . microsecond / 1000.0 ) * 1000 )
    elif unit == second :
        return when . replace ( microsecond = 0 )
    elif unit == minute :
        return when . replace ( second = 0 , microsecond = 0 )
return when


signature:
def weekday ( when , weekday , start = mon ) :
    """Return the date for the day of this week ."""

implementation:
if isinstance ( when , datetime ) :
    when = when . date ( )
today = when . weekday ( )
delta = weekday - today
if weekday < start and today >= start :
    delta += 7
elif weekday >= start and today < start :
    delta -= 7
return when + timedelta ( days = delta )


signature:
def prevweekday ( when , weekday , inclusive = True ) :
    """Return the date for the most recent day of the week . If inclusive is True ( the default ) today
 may count as the weekday we ' re looking for ."""

implementation:
if isinstance ( when , datetime ) :
    when = when . date ( )
delta = weekday - when . weekday ( )
if ( inclusive and delta > 0 ) or ( not inclusive and delta >= 0 ) :
    delta -= 7
return when + timedelta ( days = delta )


signature:
def opt_normalizations ( params , func , limits , abandon_threshold = 100 , noimprovement_threshold = 1e-3 , disp = 0 ) :
    """* * optimization algorithm for scale variables ( positive value of unknown magnitude ) * *
 Each parameter is a normalization of a feature , and its value is sought .
 The parameters are handled in order ( assumed to be independent ) ,
 but a second round can be run .
 Various magnitudes of the normalization are tried . If the normalization converges
 to zero , the largest value yielding a comparable value is used .
 Optimizes each normalization parameter in rough steps
 using multiples of 3 of start point
 to find reasonable starting values for another algorithm .
 parameters , minimization function , parameter space definition [ ( lo , hi ) for i in params ]
 : param abandon _ threshold :
 if in one direction the function increases by this much over the best value ,
 abort search in this direction
 : param noimprovement _ threshold :
 when decreasing the normalization , if the function increases by less than
 this amount , abort search in this direction
 : param disp :
 verbosity"""

implementation:
newparams = numpy . copy ( params )
lower = [ lo for lo , hi in limits ]
upper = [ hi for lo , hi in limits ]
for i , p in enumerate ( params ) :
    startval = p
    beststat = func ( newparams )
    bestval = startval
    if disp > 0 :
        print '\t\tstart val = %e: %e' % ( startval , beststat )
    go_up = True
    go_down = True
    # go up and down in multiples of 3
    # once that is done , refine in multiples of 1.1
    for n in list ( 3. ** numpy . arange ( 1 , 20 ) ) + [ None ] + list ( 1.1 ** numpy . arange ( 1 , 13 ) ) :
        if n is None :
            startval = bestval
            if disp > 0 :
                print '\t\trefining from %e' % ( startval )
            go_up = True
            go_down = True
            continue
        if go_up and startval * n > upper [ i ] :
            if disp > 0 :
                print '\t\thit upper border (%e * %e > %e)' % ( startval , n , upper [ i ] )
            go_up = False
        if go_down and startval / n < lower [ i ] :
            if disp > 0 :
                print '\t\thit lower border (%e / %e > %e)' % ( startval , n , lower [ i ] )
            go_down = False
        if go_up :
            if disp > 1 :
                print '\t\ttrying %e ^' % ( startval * n )
            newparams [ i ] = startval * n
            newstat = func ( newparams )
            if disp > 1 :
                print '\t\tval = %e: %e' % ( newparams [ i ] , newstat )
            if newstat <= beststat :
                bestval = newparams [ i ]
                beststat = newstat
                if disp > 0 :
                    print '\t\t\timprovement: %e' % newparams [ i ]
            if newstat > beststat + abandon_threshold :
                go_up = False
        if go_down :
            if disp > 1 :
                print '\t\ttrying %e v' % ( startval / n )
            newparams [ i ] = startval / n
            newstat = func ( newparams )
            if disp > 1 :
                print '\t\tval = %e: %e' % ( newparams [ i ] , newstat )
            if newstat + noimprovement_threshold < beststat : # avoid zeros in normalizations
                bestval = newparams [ i ]
                beststat = newstat
                if disp > 0 :
                    print '\t\t\timprovement: %e' % newparams [ i ]
            if newstat > beststat + abandon_threshold :
                go_down = False
    newparams [ i ] = bestval
    print '\tnew normalization of %d: %e' % ( i , newparams [ i ] )
print 'optimization done, reached %.3f' % ( beststat )
return newparams


signature:
def opt_grid ( params , func , limits , ftol = 0.01 , disp = 0 , compute_errors = True ) :
    """see : func : ` optimize1d . optimize ` , considers each parameter in order
 : param ftol :
 difference in values at which the function can be considered flat
 : param compute _ errors :
 compute standard deviation of gaussian around optimum"""

implementation:
caches = [ [ ] for p in params ]
newparams = numpy . copy ( params )
errors = [ [ ] for p in params ]
for i , p in enumerate ( params ) :
    cache = [ ]
    def func1 ( x0 ) :
        newparams [ i ] = x0
        v = func ( newparams )
        cache . append ( [ x0 , v ] )
        return v
    lo , hi = limits [ i ]
    bestval = optimize ( func1 , x0 = p , cons = [ lambda x : x - lo , lambda x : hi - x ] , ftol = ftol , disp = disp - 1 )
    beststat = func1 ( bestval )
    if compute_errors :
        errors [ i ] = cache2errors ( func1 , cache , disp = disp - 1 )
    newparams [ i ] = bestval
    caches [ i ] = cache
    if disp > 0 :
        if compute_errors :
            print '\tnew value of %d: %e [%e .. %e] yielded %e' % ( i , bestval , errors [ i ] [ 0 ] , errors [ i ] [ 1 ] , beststat )
        else :
            print '\tnew value of %d: %e yielded %e' % ( i , bestval , beststat )
beststat = func ( newparams )
if disp > 0 :
    print 'optimization done, reached %.3f' % ( beststat )
if compute_errors :
    return newparams , errors
else :
    return newparams


signature:
def opt_grid_parallel ( params , func , limits , ftol = 0.01 , disp = 0 , compute_errors = True ) :
    """parallelized version of : func : ` opt _ grid `"""

implementation:
import multiprocessing
def spawn ( f ) :
    def fun ( q_in , q_out ) :
        while True :
            i , x = q_in . get ( )
            if i == None :
                break
            q_out . put ( ( i , f ( x ) ) )
    return fun
def parmap ( f , X , nprocs = multiprocessing . cpu_count ( ) ) :
    q_in = multiprocessing . Queue ( 1 )
    q_out = multiprocessing . Queue ( )
    proc = [ multiprocessing . Process ( target = spawn ( f ) , args = ( q_in , q_out ) ) for _ in range ( nprocs ) ]
    for p in proc :
        p . daemon = True
        p . start ( )
    sent = [ q_in . put ( ( i , x ) ) for i , x in enumerate ( X ) ]
    [ q_in . put ( ( None , None ) ) for _ in range ( nprocs ) ]
    res = [ q_out . get ( ) for _ in range ( len ( sent ) ) ]
    [ p . join ( ) for p in proc ]
    return [ x for i , x in sorted ( res ) ]
nthreads = multiprocessing . cpu_count ( )
caches = [ [ ] for p in params ]
newparams = numpy . copy ( params )
errors = [ [ ] for p in params ]
indices = range ( 0 , len ( params ) , nthreads )
k = 0
while k < len ( params ) :
    j = min ( len ( params ) , k + nthreads * 2 )
    def run1d ( ( i , curparams , curlimits ) ) :
        cache = [ ]
        def func1 ( x0 ) :
            curparams [ i ] = x0
            v = func ( curparams )
            cache . append ( [ x0 , v ] )
            return v
        lo , hi = curlimits
        bestval = optimize ( func1 , x0 = p , cons = [ lambda x : x - lo , lambda x : hi - x ] , ftol = ftol , disp = disp - 1 )
        beststat = func1 ( bestval )
        if compute_errors :
            errors = cache2errors ( func1 , cache , disp = disp - 1 )
            return bestval , beststat , errors , cache
        return bestval , beststat , cache
    results = parmap ( run1d , [ ( i , numpy . copy ( newparams ) , limits [ i ] ) for i in range ( k , j ) ] )
    for i , r in enumerate ( results ) :
        if compute_errors :
            v , s , e , c = r
            if disp > 0 :
                print '\tnew value of %d: %e [%e .. %e] yielded %e' % ( i + k , v , e [ 0 ] , e [ 1 ] , s )
        else :
            v , s , c = r
            e = [ ]
            if disp > 0 :
                print '\tnew value of %d: %e yielded %e' % ( i + k , v , s )
        newparams [ i + k ] = v
        caches [ i + k ] = c
        errors [ i + k ] = e
    k = j
beststat = func ( newparams )
if disp > 0 :
    print 'optimization done, reached %e' % ( beststat )
if compute_errors :
    return newparams , errors
else :
    return newparams


signature:
def _GetNativeEolStyle ( platform = sys . platform ) :
    '''Internal function that determines EOL _ STYLE _ NATIVE constant with the proper value for the
 current platform .'''

implementation:
_NATIVE_EOL_STYLE_MAP = { 'win32' : EOL_STYLE_WINDOWS , 'linux2' : EOL_STYLE_UNIX , 'linux' : EOL_STYLE_UNIX , 'darwin' : EOL_STYLE_MAC , }
result = _NATIVE_EOL_STYLE_MAP . get ( platform )
if result is None :
    from . _exceptions import UnknownPlatformError
    raise UnknownPlatformError ( platform )
return result


signature:
def Cwd ( directory ) :
    '''Context manager for current directory ( uses with _ statement )
 e . g . :
 # working on some directory
 with Cwd ( ' / home / new _ dir ' ) :
 # working on new _ dir
 # working on some directory again
 : param unicode directory :
 Target directory to enter'''

implementation:
old_directory = six . moves . getcwd ( )
if directory is not None :
    os . chdir ( directory )
try :
    yield directory
finally :
    os . chdir ( old_directory )


signature:
def NormalizePath ( path ) :
    '''Normalizes a path maintaining the final slashes .
 Some environment variables need the final slash in order to work .
 Ex . The SOURCES _ DIR set by subversion must end with a slash because of the way it is used
 in the Visual Studio projects .
 : param unicode path :
 The path to normalize .
 : rtype : unicode
 : returns :
 Normalized path'''

implementation:
if path . endswith ( '/' ) or path . endswith ( '\\' ) :
    slash = os . path . sep
else :
    slash = ''
return os . path . normpath ( path ) + slash


signature:
def CanonicalPath ( path ) :
    '''Returns a version of a path that is unique .
 Given two paths path1 and path2:
 CanonicalPath ( path1 ) = = CanonicalPath ( path2 ) if and only if they represent the same file on
 the host OS . Takes account of case , slashes and relative paths .
 : param unicode path :
 The original path .
 : rtype : unicode
 : returns :
 The unique path .'''

implementation:
path = os . path . normpath ( path )
path = os . path . abspath ( path )
path = os . path . normcase ( path )
return path


signature:
def StandardizePath ( path , strip = False ) :
    '''Replaces all slashes and backslashes with the target separator
 StandardPath :
 We are defining that the standard - path is the one with only back - slashes in it , either
 on Windows or any other platform .
 : param bool strip :
 If True , removes additional slashes from the end of the path .'''

implementation:
path = path . replace ( SEPARATOR_WINDOWS , SEPARATOR_UNIX )
if strip :
    path = path . rstrip ( SEPARATOR_UNIX )
return path


signature:
def NormStandardPath ( path ) :
    '''Normalizes a standard path ( posixpath . normpath ) maintaining any slashes at the end of the path .
 Normalize :
 Removes any local references in the path " / . . / "
 StandardPath :
 We are defining that the standard - path is the one with only back - slashes in it , either
 on Windows or any other platform .'''

implementation:
import posixpath
if path . endswith ( '/' ) :
    slash = '/'
else :
    slash = ''
return posixpath . normpath ( path ) + slash


signature:
def CreateMD5 ( source_filename , target_filename = None ) :
    '''Creates a md5 file from a source file ( contents are the md5 hash of source file )
 : param unicode source _ filename :
 Path to source file
 : type target _ filename : unicode or None
 : param target _ filename :
 Name of the target file with the md5 contents
 If None , defaults to source _ filename + ' . md5' '''

implementation:
if target_filename is None :
    target_filename = source_filename + '.md5'
from six . moves . urllib . parse import urlparse
source_url = urlparse ( source_filename )
# Obtain MD5 hex
if _UrlIsLocal ( source_url ) : # If using a local file , we can give Md5Hex the filename
    md5_contents = Md5Hex ( filename = source_filename )
else : # Md5Hex can ' t handle remote files , we open it and pray we won ' t run out of memory .
    md5_contents = Md5Hex ( contents = GetFileContents ( source_filename , binary = True ) )
# Write MD5 hash to a file
CreateFile ( target_filename , md5_contents )


signature:
def CopyFile ( source_filename , target_filename , override = True , md5_check = False , copy_symlink = True ) :
    '''Copy a file from source to target .
 : param source _ filename :
 @ see _ DoCopyFile
 : param target _ filename :
 @ see _ DoCopyFile
 : param bool md5 _ check :
 If True , checks md5 files ( of both source and target files ) , if they match , skip this copy
 and return MD5 _ SKIP
 Md5 files are assumed to be { source , target } + ' . md5'
 If any file is missing ( source , target or md5 ) , the copy will always be made .
 : param copy _ symlink :
 @ see _ DoCopyFile
 : raises FileAlreadyExistsError :
 If target _ filename already exists , and override is False
 : raises NotImplementedProtocol :
 If file protocol is not accepted
 Protocols allowed are :
 source _ filename : local , ftp , http
 target _ filename : local , ftp
 : rtype : None | MD5 _ SKIP
 : returns :
 MD5 _ SKIP if the file was not copied because there was a matching . md5 file
 . . seealso : : FTP LIMITATIONS at this module ' s doc for performance issues information'''

implementation:
from . _exceptions import FileNotFoundError
# Check override
if not override and Exists ( target_filename ) :
    from . _exceptions import FileAlreadyExistsError
    raise FileAlreadyExistsError ( target_filename )
# Don ' t do md5 check for md5 files themselves .
md5_check = md5_check and not target_filename . endswith ( '.md5' )
# If we enabled md5 checks , ignore copy of files that haven ' t changed their md5 contents .
if md5_check :
    source_md5_filename = source_filename + '.md5'
    target_md5_filename = target_filename + '.md5'
    try :
        source_md5_contents = GetFileContents ( source_md5_filename )
    except FileNotFoundError :
        source_md5_contents = None
    try :
        target_md5_contents = GetFileContents ( target_md5_filename )
    except FileNotFoundError :
        target_md5_contents = None
    if source_md5_contents is not None and source_md5_contents == target_md5_contents and Exists ( target_filename ) :
        return MD5_SKIP
# Copy source file
_DoCopyFile ( source_filename , target_filename , copy_symlink = copy_symlink )
# If we have a source _ md5 , but no target _ md5 , create the target _ md5 file
if md5_check and source_md5_contents is not None and source_md5_contents != target_md5_contents :
    CreateFile ( target_md5_filename , source_md5_contents )


signature:
def _DoCopyFile ( source_filename , target_filename , copy_symlink = True ) :
    ''': param unicode source _ filename :
 The source filename .
 Schemas : local , ftp , http
 : param unicode target _ filename :
 Target filename .
 Schemas : local , ftp
 : param copy _ symlink :
 @ see _ CopyFileLocal
 : raises FileNotFoundError :
 If source _ filename does not exist'''

implementation:
from six . moves . urllib . parse import urlparse
source_url = urlparse ( source_filename )
target_url = urlparse ( target_filename )
if _UrlIsLocal ( source_url ) :
    if not Exists ( source_filename ) :
        from . _exceptions import FileNotFoundError
        raise FileNotFoundError ( source_filename )
    if _UrlIsLocal ( target_url ) : # local to local
        _CopyFileLocal ( source_filename , target_filename , copy_symlink = copy_symlink )
    elif target_url . scheme in [ 'ftp' ] :
        from . _exceptions import NotImplementedProtocol
        raise NotImplementedProtocol ( target_url . scheme )
    else :
        from . _exceptions import NotImplementedProtocol
        raise NotImplementedProtocol ( target_url . scheme )
elif source_url . scheme in [ 'http' , 'https' , 'ftp' ] :
    if _UrlIsLocal ( target_url ) : # HTTP / FTP to local
        from . _exceptions import NotImplementedProtocol
        raise NotImplementedProtocol ( target_url . scheme )
    else : # HTTP / FTP to other = = > NotImplemented
        from . _exceptions import NotImplementedProtocol
        raise NotImplementedProtocol ( target_url . scheme )
else :
    from . _exceptions import NotImplementedProtocol
    # @ Reimport
    raise NotImplementedProtocol ( source_url . scheme )


signature:
def _CopyFileLocal ( source_filename , target_filename , copy_symlink = True ) :
    '''Copy a file locally to a directory .
 : param unicode source _ filename :
 The filename to copy from .
 : param unicode target _ filename :
 The filename to copy to .
 : param bool copy _ symlink :
 If True and source _ filename is a symlink , target _ filename will also be created as
 a symlink .
 If False , the file being linked will be copied instead .'''

implementation:
import shutil
try : # > > > Create the target _ filename directory if necessary
    dir_name = os . path . dirname ( target_filename )
    if dir_name and not os . path . isdir ( dir_name ) :
        os . makedirs ( dir_name )
    if copy_symlink and IsLink ( source_filename ) : # > > > Delete the target _ filename if it already exists
        if os . path . isfile ( target_filename ) or IsLink ( target_filename ) :
            DeleteFile ( target_filename )
        # > > > Obtain the relative path from link to source _ filename ( linkto )
        source_filename = ReadLink ( source_filename )
        CreateLink ( source_filename , target_filename )
    else : # shutil can ' t copy links in Windows , so we must find the real file manually
        if sys . platform == 'win32' :
            while IsLink ( source_filename ) :
                link = ReadLink ( source_filename )
                if os . path . isabs ( link ) :
                    source_filename = link
                else :
                    source_filename = os . path . join ( os . path . dirname ( source_filename ) , link )
        shutil . copyfile ( source_filename , target_filename )
        shutil . copymode ( source_filename , target_filename )
except Exception as e :
    reraise ( e , 'While executiong _filesystem._CopyFileLocal(%s, %s)' % ( source_filename , target_filename ) )


signature:
def CopyFiles ( source_dir , target_dir , create_target_dir = False , md5_check = False ) :
    '''Copy files from the given source to the target .
 : param unicode source _ dir :
 A filename , URL or a file mask .
 Ex .
 x : \ coilib50
 x : \ coilib50 \ *
 http : / / server / directory / file
 ftp : / / server / directory / file
 : param unicode target _ dir :
 A directory or an URL
 Ex .
 d : \ Temp
 ftp : / / server / directory
 : param bool create _ target _ dir :
 If True , creates the target path if it doesn ' t exists .
 : param bool md5 _ check :
 . . seealso : : CopyFile
 : raises DirectoryNotFoundError :
 If target _ dir does not exist , and create _ target _ dir is False
 . . seealso : : CopyFile for documentation on accepted protocols
 . . seealso : : FTP LIMITATIONS at this module ' s doc for performance issues information'''

implementation:
import fnmatch
# Check if we were given a directory or a directory with mask
if IsDir ( source_dir ) : # Yes , it ' s a directory , copy everything from it
    source_mask = '*'
else : # Split directory and mask
    source_dir , source_mask = os . path . split ( source_dir )
# Create directory if necessary
if not IsDir ( target_dir ) :
    if create_target_dir :
        CreateDirectory ( target_dir )
    else :
        from . _exceptions import DirectoryNotFoundError
        raise DirectoryNotFoundError ( target_dir )
# List and match files
filenames = ListFiles ( source_dir )
# Check if we have a source directory
if filenames is None :
    return
# Copy files
for i_filename in filenames :
    if md5_check and i_filename . endswith ( '.md5' ) :
        continue
    # md5 files will be copied by CopyFile when copying their associated files
    if fnmatch . fnmatch ( i_filename , source_mask ) :
        source_path = source_dir + '/' + i_filename
        target_path = target_dir + '/' + i_filename
        if IsDir ( source_path ) : # If we found a directory , copy it recursively
            CopyFiles ( source_path , target_path , create_target_dir = True , md5_check = md5_check )
        else :
            CopyFile ( source_path , target_path , md5_check = md5_check )


signature:
def CopyFilesX ( file_mapping ) :
    '''Copies files into directories , according to a file mapping
 : param list ( tuple ( unicode , unicode ) ) file _ mapping :
 A list of mappings between the directory in the target and the source .
 For syntax , @ see : ExtendedPathMask
 : rtype : list ( tuple ( unicode , unicode ) )
 : returns :
 List of files copied . ( source _ filename , target _ filename )
 . . seealso : : FTP LIMITATIONS at this module ' s doc for performance issues information'''

implementation:
# List files that match the mapping
files = [ ]
for i_target_path , i_source_path_mask in file_mapping :
    tree_recurse , flat_recurse , dirname , in_filters , out_filters = ExtendedPathMask . Split ( i_source_path_mask )
    _AssertIsLocal ( dirname )
    filenames = FindFiles ( dirname , in_filters , out_filters , tree_recurse )
    for i_source_filename in filenames :
        if os . path . isdir ( i_source_filename ) :
            continue
        # Do not copy dirs
        i_target_filename = i_source_filename [ len ( dirname ) + 1 : ]
        if flat_recurse :
            i_target_filename = os . path . basename ( i_target_filename )
        i_target_filename = os . path . join ( i_target_path , i_target_filename )
        files . append ( ( StandardizePath ( i_source_filename ) , StandardizePath ( i_target_filename ) ) )
# Copy files
for i_source_filename , i_target_filename in files : # Create target dir if necessary
    target_dir = os . path . dirname ( i_target_filename )
    CreateDirectory ( target_dir )
    CopyFile ( i_source_filename , i_target_filename )
return files


signature:
def IsFile ( path ) :
    ''': param unicode path :
 Path to a file ( local or ftp )
 : raises NotImplementedProtocol :
 If checking for a non - local , non - ftp file
 : rtype : bool
 : returns :
 True if the file exists
 . . seealso : : FTP LIMITATIONS at this module ' s doc for performance issues information'''

implementation:
from six . moves . urllib . parse import urlparse
url = urlparse ( path )
if _UrlIsLocal ( url ) :
    if IsLink ( path ) :
        return IsFile ( ReadLink ( path ) )
    return os . path . isfile ( path )
elif url . scheme == 'ftp' :
    from . _exceptions import NotImplementedProtocol
    raise NotImplementedProtocol ( url . scheme )
else :
    from . _exceptions import NotImplementedProtocol
    raise NotImplementedProtocol ( url . scheme )


signature:
def GetDriveType ( path ) :
    '''Determine the type of drive , which can be one of the following values :
 DRIVE _ UNKNOWN = 0
 The drive type cannot be determined .
 DRIVE _ NO _ ROOT _ DIR = 1
 The root path is invalid ; for example , there is no volume mounted at the specified path .
 DRIVE _ REMOVABLE = 2
 The drive has removable media ; for example , a floppy drive , thumb drive , or flash card reader .
 DRIVE _ FIXED = 3
 The drive has fixed media ; for example , a hard disk drive or flash drive .
 DRIVE _ REMOTE = 4
 The drive is a remote ( network ) drive .
 DRIVE _ CDROM = 5
 The drive is a CD - ROM drive .
 DRIVE _ RAMDISK = 6
 The drive is a RAM disk
 : note :
 The implementation is valid only for Windows OS
 Linux will always return DRIVE _ UNKNOWN
 : param path :
 Path to a file or directory'''

implementation:
if sys . platform == 'win32' :
    import ctypes
    kdll = ctypes . windll . LoadLibrary ( "kernel32.dll" )
    return kdll . GetDriveType ( path + '\\' )
    import win32file
    if IsFile ( path ) :
        path = os . path . dirname ( path )
    # A trailing backslash is required .
    return win32file . GetDriveType ( path + '\\' )
else :
    return DRIVE_UNKNOWN


signature:
def IsDir ( directory ) :
    ''': param unicode directory :
 A path
 : rtype : bool
 : returns :
 Returns whether the given path points to an existent directory .
 : raises NotImplementedProtocol :
 If the path protocol is not local or ftp
 . . seealso : : FTP LIMITATIONS at this module ' s doc for performance issues information'''

implementation:
from six . moves . urllib . parse import urlparse
directory_url = urlparse ( directory )
if _UrlIsLocal ( directory_url ) :
    return os . path . isdir ( directory )
elif directory_url . scheme == 'ftp' :
    from . _exceptions import NotImplementedProtocol
    raise NotImplementedProtocol ( target_url . scheme )
else :
    from . _exceptions import NotImplementedProtocol
    raise NotImplementedProtocol ( directory_url . scheme )


signature:
def Exists ( path ) :
    ''': rtype : bool
 : returns :
 True if the path already exists ( either a file or a directory )
 . . seealso : : FTP LIMITATIONS at this module ' s doc for performance issues information'''

implementation:
from six . moves . urllib . parse import urlparse
path_url = urlparse ( path )
# Handle local
if _UrlIsLocal ( path_url ) :
    return IsFile ( path ) or IsDir ( path ) or IsLink ( path )
return IsFile ( path ) or IsDir ( path )


signature:
def CopyDirectory ( source_dir , target_dir , override = False ) :
    '''Recursively copy a directory tree .
 : param unicode source _ dir :
 Where files will come from
 : param unicode target _ dir :
 Where files will go to
 : param bool override :
 If True and target _ dir already exists , it will be deleted before copying .
 : raises NotImplementedForRemotePathError :
 If trying to copy to / from remote directories'''

implementation:
_AssertIsLocal ( source_dir )
_AssertIsLocal ( target_dir )
if override and IsDir ( target_dir ) :
    DeleteDirectory ( target_dir , skip_on_error = False )
import shutil
shutil . copytree ( source_dir , target_dir )


signature:
def DeleteFile ( target_filename ) :
    '''Deletes the given local filename .
 . . note : : If file doesn ' t exist this method has no effect .
 : param unicode target _ filename :
 A local filename
 : raises NotImplementedForRemotePathError :
 If trying to delete a non - local path
 : raises FileOnlyActionError :
 Raised when filename refers to a directory .'''

implementation:
_AssertIsLocal ( target_filename )
try :
    if IsLink ( target_filename ) :
        DeleteLink ( target_filename )
    elif IsFile ( target_filename ) :
        os . remove ( target_filename )
    elif IsDir ( target_filename ) :
        from . _exceptions import FileOnlyActionError
        raise FileOnlyActionError ( target_filename )
except Exception as e :
    reraise ( e , 'While executing filesystem.DeleteFile(%s)' % ( target_filename ) )


signature:
def AppendToFile ( filename , contents , eol_style = EOL_STYLE_NATIVE , encoding = None , binary = False ) :
    '''Appends content to a local file .
 : param unicode filename :
 : param unicode contents :
 : type eol _ style : EOL _ STYLE _ XXX constant
 : param eol _ style :
 Replaces the EOL by the appropriate EOL depending on the eol _ style value .
 Considers that all content is using only " \n " as EOL .
 : param unicode encoding :
 Target file ' s content encoding .
 Defaults to sys . getfilesystemencoding ( )
 : param bool binary :
 If True , content is appended in binary mode . In this case , ` contents ` must be ` bytes ` and not
 ` unicode `
 : raises NotImplementedForRemotePathError :
 If trying to modify a non - local path
 : raises ValueError :
 If trying to mix unicode ` contents ` without ` encoding ` , or ` encoding ` without
 unicode ` contents `'''

implementation:
_AssertIsLocal ( filename )
assert isinstance ( contents , six . text_type ) ^ binary , 'Must always receive unicode contents, unless binary=True'
if not binary : # Replaces eol on each line by the given eol _ style .
    contents = _HandleContentsEol ( contents , eol_style )
    # Handle encoding here , and always write in binary mode . We can ' t use io . open because it
    # tries to do its own line ending handling .
    contents = contents . encode ( encoding or sys . getfilesystemencoding ( ) )
oss = open ( filename , 'ab' )
try :
    oss . write ( contents )
finally :
    oss . close ( )


signature:
def MoveFile ( source_filename , target_filename ) :
    '''Moves a file .
 : param unicode source _ filename :
 : param unicode target _ filename :
 : raises NotImplementedForRemotePathError :
 If trying to operate with non - local files .'''

implementation:
_AssertIsLocal ( source_filename )
_AssertIsLocal ( target_filename )
import shutil
shutil . move ( source_filename , target_filename )


signature:
def MoveDirectory ( source_dir , target_dir ) :
    '''Moves a directory .
 : param unicode source _ dir :
 : param unicode target _ dir :
 : raises NotImplementedError :
 If trying to move anything other than :
 Local dir - > local dir
 FTP dir - > FTP dir ( same host )'''

implementation:
if not IsDir ( source_dir ) :
    from . _exceptions import DirectoryNotFoundError
    raise DirectoryNotFoundError ( source_dir )
if Exists ( target_dir ) :
    from . _exceptions import DirectoryAlreadyExistsError
    raise DirectoryAlreadyExistsError ( target_dir )
from six . moves . urllib . parse import urlparse
source_url = urlparse ( source_dir )
target_url = urlparse ( target_dir )
# Local to local
if _UrlIsLocal ( source_url ) and _UrlIsLocal ( target_url ) :
    import shutil
    shutil . move ( source_dir , target_dir )
# FTP to FTP
elif source_url . scheme == 'ftp' and target_url . scheme == 'ftp' :
    from . _exceptions import NotImplementedProtocol
    raise NotImplementedProtocol ( target_url . scheme )
else :
    raise NotImplementedError ( 'Can only move directories local->local or ftp->ftp' )


signature:
def GetFileContents ( filename , binary = False , encoding = None , newline = None ) :
    '''Reads a file and returns its contents . Works for both local and remote files .
 : param unicode filename :
 : param bool binary :
 If True returns the file as is , ignore any EOL conversion .
 : param unicode encoding :
 File ' s encoding . If not None , contents obtained from file will be decoded using this
 ` encoding ` .
 : param None | ' ' | ' \n ' | ' \r ' | ' \r \n ' newline :
 Controls universal newlines .
 See ' io . open ' newline parameter documentation for more details .
 : returns str | unicode :
 The file ' s contents .
 Returns unicode string when ` encoding ` is not None .
 . . seealso : : FTP LIMITATIONS at this module ' s doc for performance issues information'''

implementation:
source_file = OpenFile ( filename , binary = binary , encoding = encoding , newline = newline )
try :
    contents = source_file . read ( )
finally :
    source_file . close ( )
return contents


signature:
def GetFileLines ( filename , newline = None , encoding = None ) :
    '''Reads a file and returns its contents as a list of lines . Works for both local and remote files .
 : param unicode filename :
 : param None | ' ' | ' \n ' | ' \r ' | ' \r \n ' newline :
 Controls universal newlines .
 See ' io . open ' newline parameter documentation for more details .
 : param unicode encoding :
 File ' s encoding . If not None , contents obtained from file will be decoded using this
 ` encoding ` .
 : returns list ( unicode ) :
 The file ' s lines
 . . seealso : : FTP LIMITATIONS at this module ' s doc for performance issues information'''

implementation: return GetFileContents ( filename , binary = False , encoding = encoding , newline = newline , ) . split ( '\n' )


signature:
def OpenFile ( filename , binary = False , newline = None , encoding = None ) :
    '''Open a file and returns it .
 Consider the possibility of a remote file ( HTTP , HTTPS , FTP )
 : param unicode filename :
 Local or remote filename .
 : param bool binary :
 If True returns the file as is , ignore any EOL conversion .
 If set ignores univeral _ newlines parameter .
 : param None | ' ' | ' \n ' | ' \r ' | ' \r \n ' newline :
 Controls universal newlines .
 See ' io . open ' newline parameter documentation for more details .
 : param unicode encoding :
 File ' s encoding . If not None , contents obtained from file will be decoded using this
 ` encoding ` .
 : returns file :
 The open file , it must be closed by the caller
 @ raise : FileNotFoundError
 When the given filename cannot be found
 . . seealso : : FTP LIMITATIONS at this module ' s doc for performance issues information'''

implementation:
from six . moves . urllib . parse import urlparse
filename_url = urlparse ( filename )
# Check if file is local
if _UrlIsLocal ( filename_url ) :
    if not os . path . isfile ( filename ) :
        from . _exceptions import FileNotFoundError
        raise FileNotFoundError ( filename )
    mode = 'rb' if binary else 'r'
    return io . open ( filename , mode , encoding = encoding , newline = newline )
# Not local
from . _exceptions import NotImplementedProtocol
raise NotImplementedProtocol ( target_url . scheme )


signature:
def ListFiles ( directory ) :
    '''Lists the files in the given directory
 : type directory : unicode | unicode
 : param directory :
 A directory or URL
 : rtype : list ( unicode ) | list ( unicode )
 : returns :
 List of filenames / directories found in the given directory .
 Returns None if the given directory does not exists .
 If ` directory ` is a unicode string , all files returned will also be unicode
 : raises NotImplementedProtocol :
 If file protocol is not local or FTP
 . . seealso : : FTP LIMITATIONS at this module ' s doc for performance issues information'''

implementation:
from six . moves . urllib . parse import urlparse
directory_url = urlparse ( directory )
# Handle local
if _UrlIsLocal ( directory_url ) :
    if not os . path . isdir ( directory ) :
        return None
    return os . listdir ( directory )
# Handle FTP
elif directory_url . scheme == 'ftp' :
    from . _exceptions import NotImplementedProtocol
    raise NotImplementedProtocol ( directory_url . scheme )
else :
    from . _exceptions import NotImplementedProtocol
    raise NotImplementedProtocol ( directory_url . scheme )


signature:
def CreateFile ( filename , contents , eol_style = EOL_STYLE_NATIVE , create_dir = True , encoding = None , binary = False ) :
    '''Create a file with the given contents .
 : param unicode filename :
 Filename and path to be created .
 : param unicode contents :
 The file contents as a string .
 : type eol _ style : EOL _ STYLE _ XXX constant
 : param eol _ style :
 Replaces the EOL by the appropriate EOL depending on the eol _ style value .
 Considers that all content is using only " \n " as EOL .
 : param bool create _ dir :
 If True , also creates directories needed in filename ' s path
 : param unicode encoding :
 Target file ' s content encoding . Defaults to sys . getfilesystemencoding ( )
 Ignored if ` binary ` = True
 : param bool binary :
 If True , file is created in binary mode . In this case , ` contents ` must be ` bytes ` and not
 ` unicode `
 : return unicode :
 Returns the name of the file created .
 : raises NotImplementedProtocol :
 If file protocol is not local or FTP
 : raises ValueError :
 If trying to mix unicode ` contents ` without ` encoding ` , or ` encoding ` without
 unicode ` contents `
 . . seealso : : FTP LIMITATIONS at this module ' s doc for performance issues information'''

implementation:
# Lots of checks when writing binary files
if binary :
    if isinstance ( contents , six . text_type ) :
        raise TypeError ( 'contents must be str (bytes) when binary=True' )
else :
    if not isinstance ( contents , six . text_type ) :
        raise TypeError ( 'contents must be unicode when binary=False' )
    # Replaces eol on each line by the given eol _ style .
    contents = _HandleContentsEol ( contents , eol_style )
    # Encode string and pretend we are using binary to prevent ' open ' from automatically
    # changing Eols
    encoding = encoding or sys . getfilesystemencoding ( )
    contents = contents . encode ( encoding )
    binary = True
# If asked , creates directory containing file
if create_dir :
    dirname = os . path . dirname ( filename )
    if dirname :
        CreateDirectory ( dirname )
from six . moves . urllib . parse import urlparse
filename_url = urlparse ( filename )
# Handle local
if _UrlIsLocal ( filename_url ) : # Always writing as binary ( see handling above )
    with open ( filename , 'wb' ) as oss :
        oss . write ( contents )
# Handle FTP
elif filename_url . scheme == 'ftp' : # Always writing as binary ( see handling above )
    from . _exceptions import NotImplementedProtocol
    raise NotImplementedProtocol ( directory_url . scheme )
else :
    from . _exceptions import NotImplementedProtocol
    raise NotImplementedProtocol ( filename_url . scheme )
return filename


signature:
def ReplaceInFile ( filename , old , new , encoding = None ) :
    '''Replaces all occurrences of " old " by " new " in the given file .
 : param unicode filename :
 The name of the file .
 : param unicode old :
 The string to search for .
 : param unicode new :
 Replacement string .
 : return unicode :
 The new contents of the file .'''

implementation:
contents = GetFileContents ( filename , encoding = encoding )
contents = contents . replace ( old , new )
CreateFile ( filename , contents , encoding = encoding )
return contents


signature:
def CreateDirectory ( directory ) :
    '''Create directory including any missing intermediate directory .
 : param unicode directory :
 : return unicode | urlparse . ParseResult :
 Returns the created directory or url ( see urlparse ) .
 : raises NotImplementedProtocol :
 If protocol is not local or FTP .
 . . seealso : : FTP LIMITATIONS at this module ' s doc for performance issues information'''

implementation:
from six . moves . urllib . parse import urlparse
directory_url = urlparse ( directory )
# Handle local
if _UrlIsLocal ( directory_url ) :
    if not os . path . exists ( directory ) :
        os . makedirs ( directory )
    return directory
# Handle FTP
elif directory_url . scheme == 'ftp' :
    from . _exceptions import NotImplementedProtocol
    raise NotImplementedProtocol ( directory_url . scheme )
else :
    from . _exceptions import NotImplementedProtocol
    raise NotImplementedProtocol ( directory_url . scheme )


signature:
def GetMTime ( path ) :
    ''': param unicode path :
 Path to file or directory
 : rtype : float
 : returns :
 Modification time for path .
 If this is a directory , the highest mtime from files inside it will be returned .
 @ note :
 In some Linux distros ( such as CentOs , or anything with ext3 ) , mtime will not return a value
 with resolutions higher than a second .
 http : / / stackoverflow . com / questions / 2428556 / os - path - getmtime - doesnt - return - fraction - of - a - second'''

implementation:
_AssertIsLocal ( path )
if os . path . isdir ( path ) :
    files = FindFiles ( path )
    if len ( files ) > 0 :
        return max ( map ( os . path . getmtime , files ) )
return os . path . getmtime ( path )


signature:
def ListMappedNetworkDrives ( ) :
    '''On Windows , returns a list of mapped network drives
 : return : tuple ( string , string , bool )
 For each mapped netword drive , return 3 values tuple :
 - the local drive
 - the remote path -
 - True if the mapping is enabled ( warning : not reliable )'''

implementation:
if sys . platform != 'win32' :
    raise NotImplementedError
drives_list = [ ]
netuse = _CallWindowsNetCommand ( [ 'use' ] )
for line in netuse . split ( EOL_STYLE_WINDOWS ) :
    match = re . match ( "(\w*)\s+(\w:)\s+(.+)" , line . rstrip ( ) )
    if match :
        drives_list . append ( ( match . group ( 2 ) , match . group ( 3 ) , match . group ( 1 ) == 'OK' ) )
return drives_list


signature:
def CreateLink ( target_path , link_path , override = True ) :
    '''Create a symbolic link at ` link _ path ` pointing to ` target _ path ` .
 : param unicode target _ path :
 Link target
 : param unicode link _ path :
 Fullpath to link name
 : param bool override :
 If True and ` link _ path ` already exists as a link , that link is overridden .'''

implementation:
_AssertIsLocal ( target_path )
_AssertIsLocal ( link_path )
if override and IsLink ( link_path ) :
    DeleteLink ( link_path )
# Create directories leading up to link
dirname = os . path . dirname ( link_path )
if dirname :
    CreateDirectory ( dirname )
if sys . platform != 'win32' :
    return os . symlink ( target_path , link_path )
    # @ UndefinedVariable
else : # import ntfsutils . junction
# return ntfsutils . junction . create ( target _ path , link _ path )
    import jaraco . windows . filesystem
    return jaraco . windows . filesystem . symlink ( target_path , link_path )
    from . _easyfs_win32 import CreateSymbolicLink
    try :
        dw_flags = 0
        if target_path and os . path . isdir ( target_path ) :
            dw_flags = 1
        return CreateSymbolicLink ( target_path , link_path , dw_flags )
    except Exception as e :
        reraise ( e , 'Creating link "%(link_path)s" pointing to "%(target_path)s"' % locals ( ) )


signature:
def IsLink ( path ) :
    ''': param unicode path :
 Path being tested
 : returns bool :
 True if ` path ` is a link'''

implementation:
_AssertIsLocal ( path )
if sys . platform != 'win32' :
    return os . path . islink ( path )
import jaraco . windows . filesystem
return jaraco . windows . filesystem . islink ( path )


signature:
def ReadLink ( path ) :
    '''Read the target of the symbolic link at ` path ` .
 : param unicode path :
 Path to a symbolic link
 : returns unicode :
 Target of a symbolic link'''

implementation:
_AssertIsLocal ( path )
if sys . platform != 'win32' :
    return os . readlink ( path )
# @ UndefinedVariable
if not IsLink ( path ) :
    from . _exceptions import FileNotFoundError
    raise FileNotFoundError ( path )
import jaraco . windows . filesystem
result = jaraco . windows . filesystem . readlink ( path )
if '\\??\\' in result :
    result = result . split ( '\\??\\' ) [ 1 ]
return result


signature:
def _AssertIsLocal ( path ) :
    '''Checks if a given path is local , raise an exception if not .
 This is used in filesystem functions that do not support remote operations yet .
 : param unicode path :
 : raises NotImplementedForRemotePathError :
 If the given path is not local'''

implementation:
from six . moves . urllib . parse import urlparse
if not _UrlIsLocal ( urlparse ( path ) ) :
    from . _exceptions import NotImplementedForRemotePathError
    raise NotImplementedForRemotePathError


signature:
def _HandleContentsEol ( contents , eol_style ) :
    '''Replaces eol on each line by the given eol _ style .
 : param unicode contents :
 : type eol _ style : EOL _ STYLE _ XXX constant
 : param eol _ style :'''

implementation:
if eol_style == EOL_STYLE_NONE :
    return contents
if eol_style == EOL_STYLE_UNIX :
    return contents . replace ( '\r\n' , eol_style ) . replace ( '\r' , eol_style )
if eol_style == EOL_STYLE_MAC :
    return contents . replace ( '\r\n' , eol_style ) . replace ( '\n' , eol_style )
if eol_style == EOL_STYLE_WINDOWS :
    return contents . replace ( '\r\n' , '\n' ) . replace ( '\r' , '\n' ) . replace ( '\n' , EOL_STYLE_WINDOWS )
raise ValueError ( 'Unexpected eol style: %r' % ( eol_style , ) )


signature:
def _CallWindowsNetCommand ( parameters ) :
    '''Call Windows NET command , used to acquire / configure network services settings .
 : param parameters : list of command line parameters
 : return : command output'''

implementation:
import subprocess
popen = subprocess . Popen ( [ "net" ] + parameters , stdout = subprocess . PIPE , stderr = subprocess . STDOUT )
stdoutdata , stderrdata = popen . communicate ( )
if stderrdata :
    raise OSError ( "Failed on call net.exe: %s" % stderrdata )
return stdoutdata


signature:
def CheckForUpdate ( source , target ) :
    '''Checks if the given target filename should be re - generated because the source has changed .
 : param source : the source filename .
 : param target : the target filename .
 : return bool :
 True if the target is out - dated , False otherwise .'''

implementation: return not os . path . isfile ( target ) or os . path . getmtime ( source ) > os . path . getmtime ( target )


signature:
def MatchMasks ( filename , masks ) :
    '''Verifies if a filename match with given patterns .
 : param str filename : The filename to match .
 : param list ( str ) masks : The patterns to search in the filename .
 : return bool :
 True if the filename has matched with one pattern , False otherwise .'''

implementation:
import fnmatch
if not isinstance ( masks , ( list , tuple ) ) :
    masks = [ masks ]
for i_mask in masks :
    if fnmatch . fnmatch ( filename , i_mask ) :
        return True
return False


signature:
def FindFiles ( dir_ , in_filters = None , out_filters = None , recursive = True , include_root_dir = True , standard_paths = False ) :
    '''Searches for files in a given directory that match with the given patterns .
 : param str dir _ : the directory root , to search the files .
 : param list ( str ) in _ filters : a list with patterns to match ( default = all ) . E . g . : [ ' * . py ' ]
 : param list ( str ) out _ filters : a list with patterns to ignore ( default = none ) . E . g . : [ ' * . py ' ]
 : param bool recursive : if True search in subdirectories , otherwise , just in the root .
 : param bool include _ root _ dir : if True , includes the directory being searched in the returned paths
 : param bool standard _ paths : if True , always uses unix path separators " / "
 : return list ( str ) :
 A list of strings with the files that matched ( with the full path in the filesystem ) .'''

implementation:
# all files
if in_filters is None :
    in_filters = [ '*' ]
if out_filters is None :
    out_filters = [ ]
result = [ ]
# maintain just files that don ' t have a pattern that match with out _ filters
# walk through all directories based on dir
for dir_root , directories , filenames in os . walk ( dir_ ) :
    for i_directory in directories [ : ] :
        if MatchMasks ( i_directory , out_filters ) :
            directories . remove ( i_directory )
    for filename in directories + filenames :
        if MatchMasks ( filename , in_filters ) and not MatchMasks ( filename , out_filters ) :
            result . append ( os . path . join ( dir_root , filename ) )
    if not recursive :
        break
if not include_root_dir : # Remove root dir from all paths
    dir_prefix = len ( dir_ ) + 1
    result = [ file [ dir_prefix : ] for file in result ]
if standard_paths :
    result = map ( StandardizePath , result )
return result


signature:
def ExpandUser ( path ) :
    '''os . path . expanduser wrapper , necessary because it cannot handle unicode strings properly .
 This is not necessary in Python 3.
 : param path :
 . . seealso : : os . path . expanduser'''

implementation:
if six . PY2 :
    encoding = sys . getfilesystemencoding ( )
    path = path . encode ( encoding )
result = os . path . expanduser ( path )
if six . PY2 :
    result = result . decode ( encoding )
return result


signature:
def DumpDirHashToStringIO ( directory , stringio , base = '' , exclude = None , include = None ) :
    '''Helper to iterate over the files in a directory putting those in the passed StringIO in ini
 format .
 : param unicode directory :
 The directory for which the hash should be done .
 : param StringIO stringio :
 The string to which the dump should be put .
 : param unicode base :
 If provided should be added ( along with a ' / ' ) before the name = hash of file .
 : param unicode exclude :
 Pattern to match files to exclude from the hashing . E . g . : * . gz
 : param unicode include :
 Pattern to match files to include in the hashing . E . g . : * . zip'''

implementation:
import fnmatch
import os
files = [ ( os . path . join ( directory , i ) , i ) for i in os . listdir ( directory ) ]
files = [ i for i in files if os . path . isfile ( i [ 0 ] ) ]
for fullname , filename in files :
    if include is not None :
        if not fnmatch . fnmatch ( fullname , include ) :
            continue
    if exclude is not None :
        if fnmatch . fnmatch ( fullname , exclude ) :
            continue
    md5 = Md5Hex ( fullname )
    if base :
        stringio . write ( '%s/%s=%s\n' % ( base , filename , md5 ) )
    else :
        stringio . write ( '%s=%s\n' % ( filename , md5 ) )


signature:
def Md5Hex ( filename = None , contents = None ) :
    ''': param unicode filename :
 The file from which the md5 should be calculated . If the filename is given , the contents
 should NOT be given .
 : param unicode contents :
 The contents for which the md5 should be calculated . If the contents are given , the filename
 should NOT be given .
 : rtype : unicode
 : returns :
 Returns a string with the hex digest of the stream .'''

implementation:
import io
import hashlib
md5 = hashlib . md5 ( )
if filename :
    stream = io . open ( filename , 'rb' )
    try :
        while True :
            data = stream . read ( md5 . block_size * 128 )
            if not data :
                break
            md5 . update ( data )
    finally :
        stream . close ( )
else :
    md5 . update ( contents )
return six . text_type ( md5 . hexdigest ( ) )


signature:
def IterHashes ( iterator_size , hash_length = 7 ) :
    '''Iterator for random hexadecimal hashes
 : param iterator _ size :
 Amount of hashes return before this iterator stops .
 Goes on forever if ` iterator _ size ` is negative .
 : param int hash _ length :
 Size of each hash returned .
 : return generator ( unicode ) :'''

implementation:
if not isinstance ( iterator_size , int ) :
    raise TypeError ( 'iterator_size must be integer.' )
count = 0
while count != iterator_size :
    count += 1
    yield GetRandomHash ( hash_length )


signature:
def PushPopItem ( obj , key , value ) :
    '''A context manager to replace and restore a value using a getter and setter .
 : param object obj : The object to replace / restore .
 : param object key : The key to replace / restore in the object .
 : param object value : The value to replace .
 Example : :
 with PushPop2 ( sys . modules , ' alpha ' , None ) :
 pytest . raises ( ImportError ) :
 import alpha'''

implementation:
if key in obj :
    old_value = obj [ key ]
    obj [ key ] = value
    yield value
    obj [ key ] = old_value
else :
    obj [ key ] = value
    yield value
    del obj [ key ]


signature:
def Split ( cls , extended_path_mask ) :
    '''Splits the given path into their components : recursive , dirname , in _ filters and out _ filters
 : param str : extended _ path _ mask :
 The " extended path mask " to split
 : rtype : tuple ( bool , bool , str , list ( str ) , list ( str ) )
 : returns :
 Returns the extended path 5 components :
 - The tree - recurse flag
 - The flat - recurse flag
 - The actual path
 - A list of masks to include
 - A list of masks to exclude'''

implementation:
import os . path
r_tree_recurse = extended_path_mask [ 0 ] in '+-'
r_flat_recurse = extended_path_mask [ 0 ] in '-'
r_dirname , r_filters = os . path . split ( extended_path_mask )
if r_tree_recurse :
    r_dirname = r_dirname [ 1 : ]
filters = r_filters . split ( ';' )
r_in_filters = [ i for i in filters if not i . startswith ( '!' ) ]
r_out_filters = [ i [ 1 : ] for i in filters if i . startswith ( '!' ) ]
return r_tree_recurse , r_flat_recurse , r_dirname , r_in_filters , r_out_filters


signature:
def GetMessages ( self , files ) :
    """Gets all the messages from a specified file .
 This will find and resolve dependencies , failing if the descriptor
 pool cannot satisfy them .
 Args :
 files : The file names to extract messages from .
 Returns :
 A dictionary mapping proto names to the message classes . This will include
 any dependent messages as well as any messages defined in the same file as
 a specified message ."""

implementation:
result = { }
for file_name in files :
    file_desc = self . pool . FindFileByName ( file_name )
    for name , msg in file_desc . message_types_by_name . items ( ) :
        if file_desc . package :
            full_name = '.' . join ( [ file_desc . package , name ] )
        else :
            full_name = msg . name
        result [ full_name ] = self . GetPrototype ( self . pool . FindMessageTypeByName ( full_name ) )
    # While the extension FieldDescriptors are created by the descriptor pool ,
    # the python classes created in the factory need them to be registered
    # explicitly , which is done below .
    # The call to RegisterExtension will specifically check if the
    # extension was already registered on the object and either
    # ignore the registration if the original was the same , or raise
    # an error if they were different .
    for name , extension in file_desc . extensions_by_name . items ( ) :
        if extension . containing_type . full_name not in self . _classes :
            self . GetPrototype ( extension . containing_type )
        extended_class = self . _classes [ extension . containing_type . full_name ]
        extended_class . RegisterExtension ( extension )
return result


signature:
def multinest ( parameter_names , transform , loglikelihood , output_basename , ** problem ) :
    """* * MultiNest Nested Sampling * *
 via ` PyMultiNest < http : / / johannesbuchner . github . com / PyMultiNest / index . html > ` _ .
 : param parameter _ names : name of parameters ; not directly used here ,
 but for multinest _ marginal . py plotting tool ."""

implementation:
import numpy
from numpy import log , exp
import pymultinest
# n observations
# number of dimensions our problem has
parameters = parameter_names
n_params = len ( parameters )
def myprior ( cube , ndim , nparams ) :
    params = transform ( [ cube [ i ] for i in range ( ndim ) ] )
    for i in range ( ndim ) :
        cube [ i ] = params [ i ]
def myloglike ( cube , ndim , nparams ) :
    l = loglikelihood ( [ cube [ i ] for i in range ( ndim ) ] )
    return l
# run MultiNest
mn_args = dict ( importance_nested_sampling = False , outputfiles_basename = output_basename , resume = problem . get ( 'resume' , False ) , verbose = True , n_live_points = problem . get ( 'n_live_points' , 400 ) , const_efficiency_mode = False )
if 'seed' in problem :
    mn_args [ 'seed' ] = problem [ 'seed' ]
pymultinest . run ( myloglike , myprior , n_params , ** mn_args )
import json
# store name of parameters , always useful
with file ( '%sparams.json' % output_basename , 'w' ) as f :
    json . dump ( parameters , f , indent = 2 )
# analyse
a = pymultinest . Analyzer ( n_params = n_params , outputfiles_basename = output_basename )
s = a . get_stats ( )
with open ( '%sstats.json' % a . outputfiles_basename , mode = 'w' ) as f :
    json . dump ( s , f , indent = 2 )
chain = a . get_equal_weighted_posterior ( ) [ : , : - 1 ]
lower = [ m [ '1sigma' ] [ 0 ] for m in s [ 'marginals' ] ]
upper = [ m [ '1sigma' ] [ 1 ] for m in s [ 'marginals' ] ]
stdev = ( numpy . array ( upper ) - numpy . array ( lower ) ) / 2
center = [ m [ 'median' ] for m in s [ 'marginals' ] ]
# final = a . get _ best _ fit ( ) [ ' parameters ' ] # is already transformed
data = numpy . loadtxt ( '%slive.points' % output_basename )
i = data [ : , - 1 ] . argmax ( )
final = data [ i , : - 1 ]
# untransformed
return dict ( start = final , chain = chain , stdev = stdev , upper = upper , lower = lower , method = 'MultiNest' )


signature:
def specifier_to_db ( db_spec ) :
    """Return the database string for a database specifier .
 The database specifier takes a custom format for specifying local and remote
 databases . A local database is specified by the following format :
 local : < db _ name >
 For example , a database called ' sessions ' would be specified by the string
 ` ` ' local : sessions ' ` ` . Remote databases are specified like this :
 remote : < host > : < port _ num > : < db _ name >
 For example , a database called ' log ' on the server ' dev . example . com ' at port
 number 5984 would be specified by ` ` ' remote : dev . example . com : 5984 : log ' ` ` .
 These specifiers are translated into strings acceptable to CouchDB ; local
 specs are turned into the database name alone , and remote specs are turned
 into ` ` ' http : / / host : port / db _ name ' ` ` URLs ."""

implementation:
local_match = LOCAL_RE . match ( db_spec )
remote_match = REMOTE_RE . match ( db_spec )
plain_match = PLAIN_RE . match ( db_spec )
# If this looks like a local specifier :
if local_match :
    return local_match . groupdict ( ) [ 'database' ]
# If this looks like a remote specifier :
elif remote_match : # A fancy ' unpacking ' . . .
    hostname , portnum , database = map ( remote_match . groupdict ( ) . get , ( 'hostname' , 'portnum' , 'database' ) )
    local_url = settings . _ ( 'COUCHDB_SERVER' , 'http://127.0.0.1:5984/' )
    localhost , localport = urlparse . urlparse ( local_url ) [ 1 ] . split ( ':' )
    # If it ' s local , return a local DB string .
    if ( localhost == hostname ) and ( localport == portnum ) :
        return database
    # Otherwise , get a remote URL .
    return 'http://%s:%s/%s' % ( hostname , portnum , database )
# If this looks like a plain database name , return it .
elif plain_match :
    return plain_match . groupdict ( ) [ 'database' ]
# Throw a wobbly .
raise ValueError ( 'Invalid database spec: %r' % ( db_spec , ) )