signature:
def validate ( self , data , expect_header_row = True , ignore_lines = 0 , summarize = False , limit = 0 , context = None , report_unexpected_exceptions = True ) :
    """Validate ` data ` and return a list of validation problems found .
 Arguments
 ` data ` - any source of row - oriented data , e . g . , as provided by a
 ` csv . reader ` , or a list of lists of strings , or . . .
 ` expect _ header _ row ` - does the data contain a header row ( i . e . , the
 first record is a list of field names ) ? Defaults to True .
 ` ignore _ lines ` - ignore n lines ( rows ) at the beginning of the data
 ` summarize ` - only report problem codes , no other details
 ` limit ` - report at most n problems
 ` context ` - a dictionary of any additional information to be added to
 any problems found - useful if problems are being aggregated from
 multiple validators
 ` report _ unexpected _ exceptions ` - value check function , value predicates ,
 record check functions , record predicates , and other user - supplied
 validation functions may raise unexpected exceptions . If this argument
 is true , any unexpected exceptions will be reported as validation
 problems ; if False , unexpected exceptions will be handled silently ."""

implementation:
problems = list ( )
problem_generator = self . ivalidate ( data , expect_header_row , ignore_lines , summarize , context , report_unexpected_exceptions )
for i , p in enumerate ( problem_generator ) :
    if not limit or i < limit :
        problems . append ( p )
return problems


signature:
def ivalidate ( self , data , expect_header_row = True , ignore_lines = 0 , summarize = False , context = None , report_unexpected_exceptions = True ) :
    """Validate ` data ` and return a iterator over problems found .
 Use this function rather than validate ( ) if you expect a large number
 of problems .
 Arguments
 ` data ` - any source of row - oriented data , e . g . , as provided by a
 ` csv . reader ` , or a list of lists of strings , or . . .
 ` expect _ header _ row ` - does the data contain a header row ( i . e . , the
 first record is a list of field names ) ? Defaults to True .
 ` ignore _ lines ` - ignore n lines ( rows ) at the beginning of the data
 ` summarize ` - only report problem codes , no other details
 ` context ` - a dictionary of any additional information to be added to
 any problems found - useful if problems are being aggregated from
 multiple validators
 ` report _ unexpected _ exceptions ` - value check function , value predicates ,
 record check functions , record predicates , and other user - supplied
 validation functions may raise unexpected exceptions . If this argument
 is true , any unexpected exceptions will be reported as validation
 problems ; if False , unexpected exceptions will be handled silently ."""

implementation:
unique_sets = self . _init_unique_sets ( )
# used for unique checks
for i , r in enumerate ( data ) :
    if expect_header_row and i == ignore_lines : # r is the header row
        for p in self . _apply_header_checks ( i , r , summarize , context ) :
            yield p
    elif i >= ignore_lines : # r is a data row
        skip = False
        for p in self . _apply_skips ( i , r , summarize , report_unexpected_exceptions , context ) :
            if p is True :
                skip = True
            else :
                yield p
        if not skip :
            for p in self . _apply_each_methods ( i , r , summarize , report_unexpected_exceptions , context ) :
                yield p
                # may yield a problem if an exception is raised
            for p in self . _apply_value_checks ( i , r , summarize , report_unexpected_exceptions , context ) :
                yield p
            for p in self . _apply_record_length_checks ( i , r , summarize , context ) :
                yield p
            for p in self . _apply_value_predicates ( i , r , summarize , report_unexpected_exceptions , context ) :
                yield p
            for p in self . _apply_record_checks ( i , r , summarize , report_unexpected_exceptions , context ) :
                yield p
            for p in self . _apply_record_predicates ( i , r , summarize , report_unexpected_exceptions , context ) :
                yield p
            for p in self . _apply_unique_checks ( i , r , unique_sets , summarize ) :
                yield p
            for p in self . _apply_check_methods ( i , r , summarize , report_unexpected_exceptions , context ) :
                yield p
            for p in self . _apply_assert_methods ( i , r , summarize , report_unexpected_exceptions , context ) :
                yield p
for p in self . _apply_finally_assert_methods ( summarize , report_unexpected_exceptions , context ) :
    yield p


signature:
def _init_unique_sets ( self ) :
    """Initialise sets used for uniqueness checking ."""

implementation:
ks = dict ( )
for t in self . _unique_checks :
    key = t [ 0 ]
    ks [ key ] = set ( )
    # empty set
return ks


signature:
def _apply_value_checks ( self , i , r , summarize = False , report_unexpected_exceptions = True , context = None ) :
    """Apply value check functions on the given record ` r ` ."""

implementation:
for field_name , check , code , message , modulus in self . _value_checks :
    if i % modulus == 0 : # support sampling
        fi = self . _field_names . index ( field_name )
        if fi < len ( r ) : # only apply checks if there is a value
            value = r [ fi ]
            try :
                check ( value )
            except ValueError :
                p = { 'code' : code }
                if not summarize :
                    p [ 'message' ] = message
                    p [ 'row' ] = i + 1
                    p [ 'column' ] = fi + 1
                    p [ 'field' ] = field_name
                    p [ 'value' ] = value
                    p [ 'record' ] = r
                    if context is not None :
                        p [ 'context' ] = context
                yield p
            except Exception as e :
                if report_unexpected_exceptions :
                    p = { 'code' : UNEXPECTED_EXCEPTION }
                    if not summarize :
                        p [ 'message' ] = MESSAGES [ UNEXPECTED_EXCEPTION ] % ( e . __class__ . __name__ , e )
                        p [ 'row' ] = i + 1
                        p [ 'column' ] = fi + 1
                        p [ 'field' ] = field_name
                        p [ 'value' ] = value
                        p [ 'record' ] = r
                        p [ 'exception' ] = e
                        p [ 'function' ] = '%s: %s' % ( check . __name__ , check . __doc__ )
                        if context is not None :
                            p [ 'context' ] = context
                    yield p


signature:
def _apply_header_checks ( self , i , r , summarize = False , context = None ) :
    """Apply header checks on the given record ` r ` ."""

implementation:
for code , message in self . _header_checks :
    if tuple ( r ) != self . _field_names :
        p = { 'code' : code }
        if not summarize :
            p [ 'message' ] = message
            p [ 'row' ] = i + 1
            p [ 'record' ] = tuple ( r )
            p [ 'missing' ] = set ( self . _field_names ) - set ( r )
            p [ 'unexpected' ] = set ( r ) - set ( self . _field_names )
            if context is not None :
                p [ 'context' ] = context
        yield p


signature:
def _apply_record_length_checks ( self , i , r , summarize = False , context = None ) :
    """Apply record length checks on the given record ` r ` ."""

implementation:
for code , message , modulus in self . _record_length_checks :
    if i % modulus == 0 : # support sampling
        if len ( r ) != len ( self . _field_names ) :
            p = { 'code' : code }
            if not summarize :
                p [ 'message' ] = message
                p [ 'row' ] = i + 1
                p [ 'record' ] = r
                p [ 'length' ] = len ( r )
                if context is not None :
                    p [ 'context' ] = context
            yield p


signature:
def _apply_value_predicates ( self , i , r , summarize = False , report_unexpected_exceptions = True , context = None ) :
    """Apply value predicates on the given record ` r ` ."""

implementation:
for field_name , predicate , code , message , modulus in self . _value_predicates :
    if i % modulus == 0 : # support sampling
        fi = self . _field_names . index ( field_name )
        if fi < len ( r ) : # only apply predicate if there is a value
            value = r [ fi ]
            try :
                valid = predicate ( value )
                if not valid :
                    p = { 'code' : code }
                    if not summarize :
                        p [ 'message' ] = message
                        p [ 'row' ] = i + 1
                        p [ 'column' ] = fi + 1
                        p [ 'field' ] = field_name
                        p [ 'value' ] = value
                        p [ 'record' ] = r
                        if context is not None :
                            p [ 'context' ] = context
                    yield p
            except Exception as e :
                if report_unexpected_exceptions :
                    p = { 'code' : UNEXPECTED_EXCEPTION }
                    if not summarize :
                        p [ 'message' ] = MESSAGES [ UNEXPECTED_EXCEPTION ] % ( e . __class__ . __name__ , e )
                        p [ 'row' ] = i + 1
                        p [ 'column' ] = fi + 1
                        p [ 'field' ] = field_name
                        p [ 'value' ] = value
                        p [ 'record' ] = r
                        p [ 'exception' ] = e
                        p [ 'function' ] = '%s: %s' % ( predicate . __name__ , predicate . __doc__ )
                        if context is not None :
                            p [ 'context' ] = context
                    yield p


signature:
def _apply_record_checks ( self , i , r , summarize = False , report_unexpected_exceptions = True , context = None ) :
    """Apply record checks on ` r ` ."""

implementation:
for check , modulus in self . _record_checks :
    if i % modulus == 0 : # support sampling
        rdict = self . _as_dict ( r )
        try :
            check ( rdict )
        except RecordError as e :
            code = e . code if e . code is not None else RECORD_CHECK_FAILED
            p = { 'code' : code }
            if not summarize :
                message = e . message if e . message is not None else MESSAGES [ RECORD_CHECK_FAILED ]
                p [ 'message' ] = message
                p [ 'row' ] = i + 1
                p [ 'record' ] = r
                if context is not None :
                    p [ 'context' ] = context
                if e . details is not None :
                    p [ 'details' ] = e . details
            yield p
        except Exception as e :
            if report_unexpected_exceptions :
                p = { 'code' : UNEXPECTED_EXCEPTION }
                if not summarize :
                    p [ 'message' ] = MESSAGES [ UNEXPECTED_EXCEPTION ] % ( e . __class__ . __name__ , e )
                    p [ 'row' ] = i + 1
                    p [ 'record' ] = r
                    p [ 'exception' ] = e
                    p [ 'function' ] = '%s: %s' % ( check . __name__ , check . __doc__ )
                    if context is not None :
                        p [ 'context' ] = context
                yield p


signature:
def _apply_record_predicates ( self , i , r , summarize = False , report_unexpected_exceptions = True , context = None ) :
    """Apply record predicates on ` r ` ."""

implementation:
for predicate , code , message , modulus in self . _record_predicates :
    if i % modulus == 0 : # support sampling
        rdict = self . _as_dict ( r )
        try :
            valid = predicate ( rdict )
            if not valid :
                p = { 'code' : code }
                if not summarize :
                    p [ 'message' ] = message
                    p [ 'row' ] = i + 1
                    p [ 'record' ] = r
                    if context is not None :
                        p [ 'context' ] = context
                yield p
        except Exception as e :
            if report_unexpected_exceptions :
                p = { 'code' : UNEXPECTED_EXCEPTION }
                if not summarize :
                    p [ 'message' ] = MESSAGES [ UNEXPECTED_EXCEPTION ] % ( e . __class__ . __name__ , e )
                    p [ 'row' ] = i + 1
                    p [ 'record' ] = r
                    p [ 'exception' ] = e
                    p [ 'function' ] = '%s: %s' % ( predicate . __name__ , predicate . __doc__ )
                    if context is not None :
                        p [ 'context' ] = context
                yield p


signature:
def _apply_unique_checks ( self , i , r , unique_sets , summarize = False , context = None ) :
    """Apply unique checks on ` r ` ."""

implementation:
for key , code , message in self . _unique_checks :
    value = None
    values = unique_sets [ key ]
    if isinstance ( key , basestring ) : # assume key is a field name
        fi = self . _field_names . index ( key )
        if fi >= len ( r ) :
            continue
        value = r [ fi ]
    else : # assume key is a list or tuple , i . e . , compound key
        value = [ ]
        for f in key :
            fi = self . _field_names . index ( f )
            if fi >= len ( r ) :
                break
            value . append ( r [ fi ] )
        value = tuple ( value )
        # enable hashing
    if value in values :
        p = { 'code' : code }
        if not summarize :
            p [ 'message' ] = message
            p [ 'row' ] = i + 1
            p [ 'record' ] = r
            p [ 'key' ] = key
            p [ 'value' ] = value
            if context is not None :
                p [ 'context' ] = context
        yield p
    values . add ( value )


signature:
def _apply_each_methods ( self , i , r , summarize = False , report_unexpected_exceptions = True , context = None ) :
    """Invoke ' each ' methods on ` r ` ."""

implementation:
for a in dir ( self ) :
    if a . startswith ( 'each' ) :
        rdict = self . _as_dict ( r )
        f = getattr ( self , a )
        try :
            f ( rdict )
        except Exception as e :
            if report_unexpected_exceptions :
                p = { 'code' : UNEXPECTED_EXCEPTION }
                if not summarize :
                    p [ 'message' ] = MESSAGES [ UNEXPECTED_EXCEPTION ] % ( e . __class__ . __name__ , e )
                    p [ 'row' ] = i + 1
                    p [ 'record' ] = r
                    p [ 'exception' ] = e
                    p [ 'function' ] = '%s: %s' % ( f . __name__ , f . __doc__ )
                    if context is not None :
                        p [ 'context' ] = context
                yield p


signature:
def _apply_assert_methods ( self , i , r , summarize = False , report_unexpected_exceptions = True , context = None ) :
    """Apply ' assert ' methods on ` r ` ."""

implementation:
for a in dir ( self ) :
    if a . startswith ( 'assert' ) :
        rdict = self . _as_dict ( r )
        f = getattr ( self , a )
        try :
            f ( rdict )
        except AssertionError as e :
            code = ASSERT_CHECK_FAILED
            message = MESSAGES [ ASSERT_CHECK_FAILED ]
            if len ( e . args ) > 0 :
                custom = e . args [ 0 ]
                if isinstance ( custom , ( list , tuple ) ) :
                    if len ( custom ) > 0 :
                        code = custom [ 0 ]
                    if len ( custom ) > 1 :
                        message = custom [ 1 ]
                else :
                    code = custom
            p = { 'code' : code }
            if not summarize :
                p [ 'message' ] = message
                p [ 'row' ] = i + 1
                p [ 'record' ] = r
                if context is not None :
                    p [ 'context' ] = context
            yield p
        except Exception as e :
            if report_unexpected_exceptions :
                p = { 'code' : UNEXPECTED_EXCEPTION }
                if not summarize :
                    p [ 'message' ] = MESSAGES [ UNEXPECTED_EXCEPTION ] % ( e . __class__ . __name__ , e )
                    p [ 'row' ] = i + 1
                    p [ 'record' ] = r
                    p [ 'exception' ] = e
                    p [ 'function' ] = '%s: %s' % ( f . __name__ , f . __doc__ )
                    if context is not None :
                        p [ 'context' ] = context
                yield p


signature:
def _apply_check_methods ( self , i , r , summarize = False , report_unexpected_exceptions = True , context = None ) :
    """Apply ' check ' methods on ` r ` ."""

implementation:
for a in dir ( self ) :
    if a . startswith ( 'check' ) :
        rdict = self . _as_dict ( r )
        f = getattr ( self , a )
        try :
            f ( rdict )
        except RecordError as e :
            code = e . code if e . code is not None else RECORD_CHECK_FAILED
            p = { 'code' : code }
            if not summarize :
                message = e . message if e . message is not None else MESSAGES [ RECORD_CHECK_FAILED ]
                p [ 'message' ] = message
                p [ 'row' ] = i + 1
                p [ 'record' ] = r
                if context is not None :
                    p [ 'context' ] = context
                if e . details is not None :
                    p [ 'details' ] = e . details
            yield p
        except Exception as e :
            if report_unexpected_exceptions :
                p = { 'code' : UNEXPECTED_EXCEPTION }
                if not summarize :
                    p [ 'message' ] = MESSAGES [ UNEXPECTED_EXCEPTION ] % ( e . __class__ . __name__ , e )
                    p [ 'row' ] = i + 1
                    p [ 'record' ] = r
                    p [ 'exception' ] = e
                    p [ 'function' ] = '%s: %s' % ( f . __name__ , f . __doc__ )
                    if context is not None :
                        p [ 'context' ] = context
                yield p


signature:
def _apply_skips ( self , i , r , summarize = False , report_unexpected_exceptions = True , context = None ) :
    """Apply skip functions on ` r ` ."""

implementation:
for skip in self . _skips :
    try :
        result = skip ( r )
        if result is True :
            yield True
    except Exception as e :
        if report_unexpected_exceptions :
            p = { 'code' : UNEXPECTED_EXCEPTION }
            if not summarize :
                p [ 'message' ] = MESSAGES [ UNEXPECTED_EXCEPTION ] % ( e . __class__ . __name__ , e )
                p [ 'row' ] = i + 1
                p [ 'record' ] = r
                p [ 'exception' ] = e
                p [ 'function' ] = '%s: %s' % ( skip . __name__ , skip . __doc__ )
                if context is not None :
                    p [ 'context' ] = context
            yield p


signature:
def _as_dict ( self , r ) :
    """Convert the record to a dictionary using field names as keys ."""

implementation:
d = dict ( )
for i , f in enumerate ( self . _field_names ) :
    d [ f ] = r [ i ] if i < len ( r ) else None
return d


signature:
def create_validator ( ) :
    """Create an example CSV validator for patient demographic data ."""

implementation:
field_names = ( 'study_id' , 'patient_id' , 'gender' , 'age_years' , 'age_months' , 'date_inclusion' )
validator = CSVValidator ( field_names )
# basic header and record length checks
validator . add_header_check ( 'EX1' , 'bad header' )
validator . add_record_length_check ( 'EX2' , 'unexpected record length' )
# some simple value checks
validator . add_value_check ( 'study_id' , int , 'EX3' , 'study id must be an integer' )
validator . add_value_check ( 'patient_id' , int , 'EX4' , 'patient id must be an integer' )
validator . add_value_check ( 'gender' , enumeration ( 'M' , 'F' ) , 'EX5' , 'invalid gender' )
validator . add_value_check ( 'age_years' , number_range_inclusive ( 0 , 120 , int ) , 'EX6' , 'invalid age in years' )
validator . add_value_check ( 'date_inclusion' , datetime_string ( '%Y-%m-%d' ) , 'EX7' , 'invalid date' )
# a more complicated record check
def check_age_variables ( r ) :
    age_years = int ( r [ 'age_years' ] )
    age_months = int ( r [ 'age_months' ] )
    valid = ( age_months >= age_years * 12 and age_months % age_years < 12 )
    if not valid :
        raise RecordError ( 'EX8' , 'invalid age variables' )
validator . add_record_check ( check_age_variables )
return validator


signature:
def main ( ) :
    """Main function ."""

implementation:
# define a command - line argument parser
description = 'Validate a CSV data file.'
parser = argparse . ArgumentParser ( description = description )
parser . add_argument ( 'file' , metavar = 'FILE' , help = 'a file to be validated' )
parser . add_argument ( '-l' , '--limit' , dest = 'limit' , type = int , action = 'store' , default = 0 , help = 'limit the number of problems reported' )
parser . add_argument ( '-s' , '--summarize' , dest = 'summarize' , action = 'store_true' , default = False , help = 'output only a summary of the different types of problem found' )
parser . add_argument ( '-e' , '--report-unexpected-exceptions' , dest = 'report_unexpected_exceptions' , action = 'store_true' , default = False , help = 'report any unexpected exceptions as problems' )
# parse arguments
args = parser . parse_args ( )
# sanity check arguments
if not os . path . isfile ( args . file ) :
    print '%s is not a file' % args . file
    sys . exit ( 1 )
with open ( args . file , 'r' ) as f : # set up a csv reader for the data
    data = csv . reader ( f , delimiter = '\t' )
    # create a validator
    validator = create_validator ( )
    # validate the data from the csv reader
    # N . B . , validate ( ) returns a list of problems ;
    # if you expect a large number of problems , use ivalidate ( ) instead
    # of validate ( ) , but bear in mind that ivalidate ( ) returns an iterator
    # so there is no len ( )
    problems = validator . validate ( data , summarize = args . summarize , report_unexpected_exceptions = args . report_unexpected_exceptions , context = { 'file' : args . file } )
    # write problems to stdout as restructured text
    write_problems ( problems , sys . stdout , summarize = args . summarize , limit = args . limit )
    # decide how to exit
    if problems : # will not work with ivalidate ( ) because it returns an iterator
        sys . exit ( 1 )
    else :
        sys . exit ( 0 )


signature:
def process_tables ( app , docname , source ) :
    """Convert markdown tables to html , since recommonmark can ' t . This requires 3 steps :
 Snip out table sections from the markdown
 Convert them to html
 Replace the old markdown table with an html table
 This function is called by sphinx for each document . ` source ` is a 1 - item list . To update the document , replace
 element 0 in ` source ` ."""

implementation:
import markdown
md = markdown . Markdown ( extensions = [ 'markdown.extensions.tables' ] )
table_processor = markdown . extensions . tables . TableProcessor ( md . parser )
raw_markdown = source [ 0 ]
blocks = re . split ( r'\n{2,}' , raw_markdown )
for i , block in enumerate ( blocks ) :
    if table_processor . test ( None , block ) :
        html = md . convert ( block )
        styled = html . replace ( '<table>' , '<table border="1" class="docutils">' , 1 )
        # apply styling
        blocks [ i ] = styled
# re - assemble into markdown - with - tables - replaced
# must replace element 0 for changes to persist
source [ 0 ] = '\n\n' . join ( blocks )


signature:
def pack_into ( fmt , buf , offset , * args , ** kwargs ) :
    """Pack given values v1 , v2 , . . . into given bytearray ` buf ` , starting
 at given bit offset ` offset ` . Pack according to given format
 string ` fmt ` . Give ` fill _ padding ` as ` ` False ` ` to leave padding
 bits in ` buf ` unmodified ."""

implementation: return CompiledFormat ( fmt ) . pack_into ( buf , offset , * args , ** kwargs )


signature:
def pack_into_dict ( fmt , names , buf , offset , data , ** kwargs ) :
    """Same as : func : ` ~ bitstruct . pack _ into ( ) ` , but data is read from a
 dictionary .
 See : func : ` ~ bitstruct . pack _ dict ( ) ` for details on ` names ` ."""

implementation: return CompiledFormatDict ( fmt , names ) . pack_into ( buf , offset , data , ** kwargs )


signature:
def unpack_from_dict ( fmt , names , data , offset = 0 ) :
    """Same as : func : ` ~ bitstruct . unpack _ from _ dict ( ) ` , but returns a
 dictionary .
 See : func : ` ~ bitstruct . pack _ dict ( ) ` for details on ` names ` ."""

implementation: return CompiledFormatDict ( fmt , names ) . unpack_from ( data , offset )


signature:
def byteswap ( fmt , data , offset = 0 ) :
    """Swap bytes in ` data ` according to ` fmt ` , starting at byte ` offset `
 and return the result . ` fmt ` must be an iterable , iterating over
 number of bytes to swap . For example , the format string ` ` ' 24 ' ` `
 applied to the bytes ` ` b ' \\ x00 \\ x11 \\ x22 \\ x33 \\ x44 \\ x55 ' ` ` will
 produce the result ` ` b ' \\ x11 \\ x00 \\ x55 \\ x44 \\ x33 \\ x22 ' ` ` ."""

implementation:
data = BytesIO ( data )
data . seek ( offset )
data_swapped = BytesIO ( )
for f in fmt :
    swapped = data . read ( int ( f ) ) [ : : - 1 ]
    data_swapped . write ( swapped )
return data_swapped . getvalue ( )


signature:
def pack ( self , * args ) :
    """See : func : ` ~ bitstruct . pack ( ) ` ."""

implementation:
# Sanity check of the number of arguments .
if len ( args ) < self . _number_of_arguments :
    raise Error ( "pack expected {} item(s) for packing (got {})" . format ( self . _number_of_arguments , len ( args ) ) )
return self . pack_any ( args )


signature:
def pack_into ( self , buf , offset , * args , ** kwargs ) :
    """See : func : ` ~ bitstruct . pack _ into ( ) ` ."""

implementation:
# Sanity check of the number of arguments .
if len ( args ) < self . _number_of_arguments :
    raise Error ( "pack expected {} item(s) for packing (got {})" . format ( self . _number_of_arguments , len ( args ) ) )
self . pack_into_any ( buf , offset , args , ** kwargs )


signature:
def unpack_from ( self , data , offset = 0 ) :
    """See : func : ` ~ bitstruct . unpack _ from ( ) ` ."""

implementation: return tuple ( [ v [ 1 ] for v in self . unpack_from_any ( data , offset ) ] )


signature:
def pack ( self , data ) :
    """See : func : ` ~ bitstruct . pack _ dict ( ) ` ."""

implementation:
try :
    return self . pack_any ( data )
except KeyError as e :
    raise Error ( '{} not found in data dictionary' . format ( str ( e ) ) )


signature:
def pack_into ( self , buf , offset , data , ** kwargs ) :
    """See : func : ` ~ bitstruct . pack _ into _ dict ( ) ` ."""

implementation:
try :
    self . pack_into_any ( buf , offset , data , ** kwargs )
except KeyError as e :
    raise Error ( '{} not found in data dictionary' . format ( str ( e ) ) )


signature:
def unpack_from ( self , data , offset = 0 ) :
    """See : func : ` ~ bitstruct . unpack _ from _ dict ( ) ` ."""

implementation: return { info . name : v for info , v in self . unpack_from_any ( data , offset ) }


signature:
def cli ( ctx , version ) :
    """Bottery"""

implementation:
# If no subcommand was given and the version flag is true , shows
# Bottery version
if not ctx . invoked_subcommand and version :
    click . echo ( bottery . __version__ )
    ctx . exit ( )
# If no subcommand but neither the version flag , shows help message
elif not ctx . invoked_subcommand :
    click . echo ( ctx . get_help ( ) )
    ctx . exit ( )


signature:
def build_message ( self , data ) :
    '''Return a Message instance according to the data received from
 Telegram API .
 https : / / core . telegram . org / bots / api # update'''

implementation:
message_data = data . get ( 'message' ) or data . get ( 'edited_message' )
if not message_data :
    return None
edited = 'edited_message' in data
return Message ( id = message_data [ 'message_id' ] , platform = self . platform , text = message_data . get ( 'text' , '' ) , user = TelegramUser ( message_data [ 'from' ] ) , chat = TelegramChat ( message_data [ 'chat' ] ) , timestamp = message_data [ 'date' ] , raw = data , edited = edited , )


signature:
def get_chat_id ( self , message ) :
    '''Telegram chat type can be either " private " , " group " , " supergroup " or
 " channel " .
 Return user ID if it is of type " private " , chat ID otherwise .'''

implementation:
if message . chat . type == 'private' :
    return message . user . id
return message . chat . id


signature:
def build_message ( self , data ) :
    '''Return a Message instance according to the data received from
 Facebook Messenger API .'''

implementation:
if not data :
    return None
return Message ( id = data [ 'message' ] [ 'mid' ] , platform = self . platform , text = data [ 'message' ] [ 'text' ] , user = data [ 'sender' ] [ 'id' ] , timestamp = data [ 'timestamp' ] , raw = data , chat = None , # TODO : Refactor build _ messages and Message class
)


signature:
async def _get_response ( self , message ) :
    """Get response running the view with await syntax if it is a
 coroutine function , otherwise just run it the normal way ."""

implementation:
view = self . discovery_view ( message )
if not view :
    return
if inspect . iscoroutinefunction ( view ) :
    response = await view ( message )
else :
    response = view ( message )
return self . prepare_response ( response , message )


signature:
def discovery_view ( self , message ) :
    """Use the new message to search for a registered view according
 to its pattern ."""

implementation:
for handler in self . registered_handlers :
    if handler . check ( message ) :
        return handler . view
return None


signature:
async def message_handler ( self , data ) :
    """For each new message , build its platform specific message
 object and get a response ."""

implementation:
message = self . build_message ( data )
if not message :
    logger . error ( '[%s] Unable to build Message with data, data=%s, error' , self . engine_name , data )
    return
logger . info ( '[%s] New message from %s: %s' , self . engine_name , message . user , message . text )
response = await self . get_response ( message )
if response :
    await self . send_response ( response )


signature:
def diff ( iterable ) :
    """Diff elements of a sequence :
 s - > s0 - s1 , s1 - s2 , s2 - s3 , . . ."""

implementation:
a , b = tee ( iterable )
next ( b , None )
return ( i - j for i , j in izip ( a , b ) )


signature:
def unpack ( endian , fmt , data ) :
    """Unpack a byte string to the given format . If the byte string
 contains more bytes than required for the given format , the function
 returns a tuple of values ."""

implementation:
if fmt == 's' : # read data as an array of chars
    val = struct . unpack ( '' . join ( [ endian , str ( len ( data ) ) , 's' ] ) , data ) [ 0 ]
else : # read a number of values
    num = len ( data ) // struct . calcsize ( fmt )
    val = struct . unpack ( '' . join ( [ endian , str ( num ) , fmt ] ) , data )
    if len ( val ) == 1 :
        val = val [ 0 ]
return val


signature:
def read_file_header ( fd , endian ) :
    """Read mat 5 file header of the file fd .
 Returns a dict with header values ."""

implementation:
fields = [ ( 'description' , 's' , 116 ) , ( 'subsystem_offset' , 's' , 8 ) , ( 'version' , 'H' , 2 ) , ( 'endian_test' , 's' , 2 ) ]
hdict = { }
for name , fmt , num_bytes in fields :
    data = fd . read ( num_bytes )
    hdict [ name ] = unpack ( endian , fmt , data )
hdict [ 'description' ] = hdict [ 'description' ] . strip ( )
v_major = hdict [ 'version' ] >> 8
v_minor = hdict [ 'version' ] & 0xFF
hdict [ '__version__' ] = '%d.%d' % ( v_major , v_minor )
return hdict


signature:
def read_element_tag ( fd , endian ) :
    """Read data element tag : type and number of bytes .
 If tag is of the Small Data Element ( SDE ) type the element data
 is also returned ."""

implementation:
data = fd . read ( 8 )
mtpn = unpack ( endian , 'I' , data [ : 4 ] )
# The most significant two bytes of mtpn will always be 0,
# if they are not , this must be SDE format
num_bytes = mtpn >> 16
if num_bytes > 0 : # small data element format
    mtpn = mtpn & 0xFFFF
    if num_bytes > 4 :
        raise ParseError ( 'Error parsing Small Data Element (SDE) ' 'formatted data' )
    data = data [ 4 : 4 + num_bytes ]
else : # regular element
    num_bytes = unpack ( endian , 'I' , data [ 4 : ] )
    data = None
return ( mtpn , num_bytes , data )


signature:
def read_elements ( fd , endian , mtps , is_name = False ) :
    """Read elements from the file .
 If list of possible matrix data types mtps is provided , the data type
 of the elements are verified ."""

implementation:
mtpn , num_bytes , data = read_element_tag ( fd , endian )
if mtps and mtpn not in [ etypes [ mtp ] [ 'n' ] for mtp in mtps ] :
    raise ParseError ( 'Got type {}, expected {}' . format ( mtpn , ' / ' . join ( '{} ({})' . format ( etypes [ mtp ] [ 'n' ] , mtp ) for mtp in mtps ) ) )
if not data : # full format , read data
    data = fd . read ( num_bytes )
    # Seek to next 64 - bit boundary
    mod8 = num_bytes % 8
    if mod8 :
        fd . seek ( 8 - mod8 , 1 )
# parse data and return values
if is_name : # names are stored as miINT8 bytes
    fmt = 's'
    val = [ unpack ( endian , fmt , s ) for s in data . split ( b'\0' ) if s ]
    if len ( val ) == 0 :
        val = ''
    elif len ( val ) == 1 :
        val = asstr ( val [ 0 ] )
    else :
        val = [ asstr ( s ) for s in val ]
else :
    fmt = etypes [ inv_etypes [ mtpn ] ] [ 'fmt' ]
    val = unpack ( endian , fmt , data )
return val


signature:
def read_header ( fd , endian ) :
    """Read and return the matrix header ."""

implementation:
flag_class , nzmax = read_elements ( fd , endian , [ 'miUINT32' ] )
header = { 'mclass' : flag_class & 0x0FF , 'is_logical' : ( flag_class >> 9 & 1 ) == 1 , 'is_global' : ( flag_class >> 10 & 1 ) == 1 , 'is_complex' : ( flag_class >> 11 & 1 ) == 1 , 'nzmax' : nzmax }
header [ 'dims' ] = read_elements ( fd , endian , [ 'miINT32' ] )
header [ 'n_dims' ] = len ( header [ 'dims' ] )
if header [ 'n_dims' ] != 2 :
    raise ParseError ( 'Only matrices with dimension 2 are supported.' )
header [ 'name' ] = read_elements ( fd , endian , [ 'miINT8' ] , is_name = True )
return header


signature:
def read_var_header ( fd , endian ) :
    """Read full header tag .
 Return a dict with the parsed header , the file position of next tag ,
 a file like object for reading the uncompressed element data ."""

implementation:
mtpn , num_bytes = unpack ( endian , 'II' , fd . read ( 8 ) )
next_pos = fd . tell ( ) + num_bytes
if mtpn == etypes [ 'miCOMPRESSED' ] [ 'n' ] : # read compressed data
    data = fd . read ( num_bytes )
    dcor = zlib . decompressobj ( )
    # from here , read of the decompressed data
    fd_var = BytesIO ( dcor . decompress ( data ) )
    del data
    fd = fd_var
    # Check the stream is not so broken as to leave cruft behind
    if dcor . flush ( ) != b'' :
        raise ParseError ( 'Error in compressed data.' )
    # read full tag from the uncompressed data
    mtpn , num_bytes = unpack ( endian , 'II' , fd . read ( 8 ) )
if mtpn != etypes [ 'miMATRIX' ] [ 'n' ] :
    raise ParseError ( 'Expecting miMATRIX type number {}, ' 'got {}' . format ( etypes [ 'miMATRIX' ] [ 'n' ] , mtpn ) )
# read the header
header = read_header ( fd , endian )
return header , next_pos , fd


signature:
def read_numeric_array ( fd , endian , header , data_etypes ) :
    """Read a numeric matrix .
 Returns an array with rows of the numeric matrix ."""

implementation:
if header [ 'is_complex' ] :
    raise ParseError ( 'Complex arrays are not supported' )
# read array data ( stored as column - major )
data = read_elements ( fd , endian , data_etypes )
if not isinstance ( data , Sequence ) : # not an array , just a value
    return data
# transform column major data continous array to
# a row major array of nested lists
rowcount = header [ 'dims' ] [ 0 ]
colcount = header [ 'dims' ] [ 1 ]
array = [ list ( data [ c * rowcount + r ] for c in range ( colcount ) ) for r in range ( rowcount ) ]
# pack and return the array
return squeeze ( array )


signature:
def read_cell_array ( fd , endian , header ) :
    """Read a cell array .
 Returns an array with rows of the cell array ."""

implementation:
array = [ list ( ) for i in range ( header [ 'dims' ] [ 0 ] ) ]
for row in range ( header [ 'dims' ] [ 0 ] ) :
    for col in range ( header [ 'dims' ] [ 1 ] ) : # read the matrix header and array
        vheader , next_pos , fd_var = read_var_header ( fd , endian )
        varray = read_var_array ( fd_var , endian , vheader )
        array [ row ] . append ( varray )
        # move on to next field
        fd . seek ( next_pos )
# pack and return the array
if header [ 'dims' ] [ 0 ] == 1 :
    return squeeze ( array [ 0 ] )
return squeeze ( array )


signature:
def read_struct_array ( fd , endian , header ) :
    """Read a struct array .
 Returns a dict with fields of the struct array ."""

implementation:
# read field name length ( unused , as strings are null terminated )
field_name_length = read_elements ( fd , endian , [ 'miINT32' ] )
if field_name_length > 32 :
    raise ParseError ( 'Unexpected field name length: {}' . format ( field_name_length ) )
# read field names
fields = read_elements ( fd , endian , [ 'miINT8' ] , is_name = True )
if isinstance ( fields , basestring ) :
    fields = [ fields ]
# read rows and columns of each field
empty = lambda : [ list ( ) for i in range ( header [ 'dims' ] [ 0 ] ) ]
array = { }
for row in range ( header [ 'dims' ] [ 0 ] ) :
    for col in range ( header [ 'dims' ] [ 1 ] ) :
        for field in fields : # read the matrix header and array
            vheader , next_pos , fd_var = read_var_header ( fd , endian )
            data = read_var_array ( fd_var , endian , vheader )
            if field not in array :
                array [ field ] = empty ( )
            array [ field ] [ row ] . append ( data )
            # move on to next field
            fd . seek ( next_pos )
# pack the nested arrays
for field in fields :
    rows = array [ field ]
    for i in range ( header [ 'dims' ] [ 0 ] ) :
        rows [ i ] = squeeze ( rows [ i ] )
    array [ field ] = squeeze ( array [ field ] )
return array


signature:
def read_var_array ( fd , endian , header ) :
    """Read variable array ( of any supported type ) ."""

implementation:
mc = inv_mclasses [ header [ 'mclass' ] ]
if mc in numeric_class_etypes :
    return read_numeric_array ( fd , endian , header , set ( compressed_numeric ) . union ( [ numeric_class_etypes [ mc ] ] ) )
elif mc == 'mxSPARSE_CLASS' :
    raise ParseError ( 'Sparse matrices not supported' )
elif mc == 'mxCHAR_CLASS' :
    return read_char_array ( fd , endian , header )
elif mc == 'mxCELL_CLASS' :
    return read_cell_array ( fd , endian , header )
elif mc == 'mxSTRUCT_CLASS' :
    return read_struct_array ( fd , endian , header )
elif mc == 'mxOBJECT_CLASS' :
    raise ParseError ( 'Object classes not supported' )
elif mc == 'mxFUNCTION_CLASS' :
    raise ParseError ( 'Function classes not supported' )
elif mc == 'mxOPAQUE_CLASS' :
    raise ParseError ( 'Anonymous function classes not supported' )


signature:
def eof ( fd ) :
    """Determine if end - of - file is reached for file fd ."""

implementation:
b = fd . read ( 1 )
end = len ( b ) == 0
if not end :
    curpos = fd . tell ( )
    fd . seek ( curpos - 1 )
return end


signature:
def loadmat ( filename , meta = False ) :
    """Load data from MAT - file :
 data = loadmat ( filename , meta = False )
 The filename argument is either a string with the filename , or
 a file like object .
 The returned parameter ` ` data ` ` is a dict with the variables found
 in the MAT file .
 Call ` ` loadmat ` ` with parameter meta = True to include meta data , such
 as file header information and list of globals .
 A ` ` ParseError ` ` exception is raised if the MAT - file is corrupt or
 contains a data type that cannot be parsed ."""

implementation:
if isinstance ( filename , basestring ) :
    fd = open ( filename , 'rb' )
else :
    fd = filename
# Check mat file format is version 5
# For 5 format we need to read an integer in the header .
# Bytes 124 through 128 contain a version integer and an
# endian test string
fd . seek ( 124 )
tst_str = fd . read ( 4 )
little_endian = ( tst_str [ 2 : 4 ] == b'IM' )
endian = ''
if ( sys . byteorder == 'little' and little_endian ) or ( sys . byteorder == 'big' and not little_endian ) : # no byte swapping same endian
    pass
elif sys . byteorder == 'little' : # byte swapping
    endian = '>'
else : # byte swapping
    endian = '<'
maj_ind = int ( little_endian )
# major version number
maj_val = ord ( tst_str [ maj_ind ] ) if ispy2 else tst_str [ maj_ind ]
if maj_val != 1 :
    raise ParseError ( 'Can only read from Matlab level 5 MAT-files' )
# the minor version number ( unused value )
# min _ val = ord ( tst _ str [ 1 - maj _ ind ] ) if ispy2 else tst _ str [ 1 - maj _ ind ]
mdict = { }
if meta : # read the file header
    fd . seek ( 0 )
    mdict [ '__header__' ] = read_file_header ( fd , endian )
    mdict [ '__globals__' ] = [ ]
# read data elements
while not eof ( fd ) :
    hdr , next_position , fd_var = read_var_header ( fd , endian )
    name = hdr [ 'name' ]
    if name in mdict :
        raise ParseError ( 'Duplicate variable name "{}" in mat file.' . format ( name ) )
    # read the matrix
    mdict [ name ] = read_var_array ( fd_var , endian , hdr )
    if meta and hdr [ 'is_global' ] :
        mdict [ '__globals__' ] . append ( name )
    # move on to next entry in file
    fd . seek ( next_position )
fd . close ( )
return mdict


signature:
def write_elements ( fd , mtp , data , is_name = False ) :
    """Write data element tag and data .
 The tag contains the array type and the number of
 bytes the array data will occupy when written to file .
 If data occupies 4 bytes or less , it is written immediately
 as a Small Data Element ( SDE ) ."""

implementation:
fmt = etypes [ mtp ] [ 'fmt' ]
if isinstance ( data , Sequence ) :
    if fmt == 's' or is_name :
        if isinstance ( data , bytes ) :
            if is_name and len ( data ) > 31 :
                raise ValueError ( 'Name "{}" is too long (max. 31 ' 'characters allowed)' . format ( data ) )
            fmt = '{}s' . format ( len ( data ) )
            data = ( data , )
        else :
            fmt = '' . join ( '{}s' . format ( len ( s ) ) for s in data )
    else :
        l = len ( data )
        if l == 0 : # empty array
            fmt = ''
        if l > 1 : # more than one element to be written
            fmt = '{}{}' . format ( l , fmt )
else :
    data = ( data , )
num_bytes = struct . calcsize ( fmt )
if num_bytes <= 4 : # write SDE
    if num_bytes < 4 : # add pad bytes
        fmt += '{}x' . format ( 4 - num_bytes )
    fd . write ( struct . pack ( 'hh' + fmt , etypes [ mtp ] [ 'n' ] , * chain ( [ num_bytes ] , data ) ) )
    return
# write tag : element type and number of bytes
fd . write ( struct . pack ( 'b3xI' , etypes [ mtp ] [ 'n' ] , num_bytes ) )
# add pad bytes to fmt , if needed
mod8 = num_bytes % 8
if mod8 :
    fmt += '{}x' . format ( 8 - mod8 )
# write data
fd . write ( struct . pack ( fmt , * data ) )


signature:
def write_var_header ( fd , header ) :
    """Write variable header"""

implementation:
# write tag bytes ,
# and array flags + class and nzmax ( null bytes )
fd . write ( struct . pack ( 'b3xI' , etypes [ 'miUINT32' ] [ 'n' ] , 8 ) )
fd . write ( struct . pack ( 'b3x4x' , mclasses [ header [ 'mclass' ] ] ) )
# write dimensions array
write_elements ( fd , 'miINT32' , header [ 'dims' ] )
# write var name
write_elements ( fd , 'miINT8' , asbytes ( header [ 'name' ] ) , is_name = True )


signature:
def write_var_data ( fd , data ) :
    """Write variable data to file"""

implementation:
# write array data elements ( size info )
fd . write ( struct . pack ( 'b3xI' , etypes [ 'miMATRIX' ] [ 'n' ] , len ( data ) ) )
# write the data
fd . write ( data )


signature:
def write_compressed_var_array ( fd , array , name ) :
    """Write compressed variable data to file"""

implementation:
bd = BytesIO ( )
write_var_array ( bd , array , name )
data = zlib . compress ( bd . getvalue ( ) )
bd . close ( )
# write array data elements ( size info )
fd . write ( struct . pack ( 'b3xI' , etypes [ 'miCOMPRESSED' ] [ 'n' ] , len ( data ) ) )
# write the compressed data
fd . write ( data )


signature:
def write_numeric_array ( fd , header , array ) :
    """Write the numeric array"""

implementation:
# make a memory file for writing array data
bd = BytesIO ( )
# write matrix header to memory file
write_var_header ( bd , header )
if not isinstance ( array , basestring ) and header [ 'dims' ] [ 0 ] > 1 : # list array data in column major order
    array = list ( chain . from_iterable ( izip ( * array ) ) )
# write matrix data to memory file
write_elements ( bd , header [ 'mtp' ] , array )
# write the variable to disk file
data = bd . getvalue ( )
bd . close ( )
write_var_data ( fd , data )


signature:
def write_var_array ( fd , array , name = '' ) :
    """Write variable array ( of any supported type )"""

implementation:
header , array = guess_header ( array , name )
mc = header [ 'mclass' ]
if mc in numeric_class_etypes :
    return write_numeric_array ( fd , header , array )
elif mc == 'mxCHAR_CLASS' :
    return write_char_array ( fd , header , array )
elif mc == 'mxCELL_CLASS' :
    return write_cell_array ( fd , header , array )
elif mc == 'mxSTRUCT_CLASS' :
    return write_struct_array ( fd , header , array )
else :
    raise ValueError ( 'Unknown mclass {}' . format ( mc ) )


signature:
def isarray ( array , test , dim = 2 ) :
    """Returns True if test is True for all array elements .
 Otherwise , returns False ."""

implementation:
if dim > 1 :
    return all ( isarray ( array [ i ] , test , dim - 1 ) for i in range ( len ( array ) ) )
return all ( test ( i ) for i in array )


signature:
def guess_header ( array , name = '' ) :
    """Guess the array header information .
 Returns a header dict , with class , data type , and size information ."""

implementation:
header = { }
if isinstance ( array , Sequence ) and len ( array ) == 1 : # sequence with only one element , squeeze the array
    array = array [ 0 ]
if isinstance ( array , basestring ) :
    header . update ( { 'mclass' : 'mxCHAR_CLASS' , 'mtp' : 'miUTF8' , 'dims' : ( 1 if len ( array ) > 0 else 0 , len ( array ) ) } )
elif isinstance ( array , Sequence ) and len ( array ) == 0 : # empty ( int ) array
    header . update ( { 'mclass' : 'mxINT32_CLASS' , 'mtp' : 'miINT32' , 'dims' : ( 0 , 0 ) } )
elif isinstance ( array , Mapping ) : # test if cells ( values ) of all fields are of equal type and
# have equal length
    field_types = [ type ( j ) for j in array . values ( ) ]
    field_lengths = [ 1 if isinstance ( j , ( basestring , int , float ) ) else len ( j ) for j in array . values ( ) ]
    if len ( field_lengths ) == 1 :
        equal_lengths = True
        equal_types = True
    else :
        equal_lengths = not any ( diff ( field_lengths ) )
        equal_types = all ( [ field_types [ 0 ] == f for f in field_types ] )
    # if of unqeual lengths or unequal types , then treat each value
    # as a cell in a 1x1 struct
    header . update ( { 'mclass' : 'mxSTRUCT_CLASS' , 'dims' : ( 1 , field_lengths [ 0 ] if equal_lengths and equal_types else 1 ) } )
elif isinstance ( array , int ) :
    header . update ( { 'mclass' : 'mxINT32_CLASS' , 'mtp' : 'miINT32' , 'dims' : ( 1 , 1 ) } )
elif isinstance ( array , float ) :
    header . update ( { 'mclass' : 'mxDOUBLE_CLASS' , 'mtp' : 'miDOUBLE' , 'dims' : ( 1 , 1 ) } )
elif isinstance ( array , Sequence ) :
    if isarray ( array , lambda i : isinstance ( i , int ) , 1 ) : # 1D int array
        header . update ( { 'mclass' : 'mxINT32_CLASS' , 'mtp' : 'miINT32' , 'dims' : ( 1 , len ( array ) ) } )
    elif isarray ( array , lambda i : isinstance ( i , ( int , float ) ) , 1 ) : # 1D double array
        header . update ( { 'mclass' : 'mxDOUBLE_CLASS' , 'mtp' : 'miDOUBLE' , 'dims' : ( 1 , len ( array ) ) } )
    elif ( isarray ( array , lambda i : isinstance ( i , Sequence ) , 1 ) and any ( diff ( len ( s ) for s in array ) ) ) : # sequence of unequal length , assume cell array
        header . update ( { 'mclass' : 'mxCELL_CLASS' , 'dims' : ( 1 , len ( array ) ) } )
    elif isarray ( array , lambda i : isinstance ( i , basestring ) , 1 ) : # char array
        header . update ( { 'mclass' : 'mxCHAR_CLASS' , 'mtp' : 'miUTF8' , 'dims' : ( len ( array ) , len ( array [ 0 ] ) ) } )
    elif isarray ( array , lambda i : isinstance ( i , Sequence ) , 1 ) : # 2D array
        if any ( diff ( len ( j ) for j in array ) ) : # rows are of unequal length , make it a cell array
            header . update ( { 'mclass' : 'mxCELL_CLASS' , 'dims' : ( len ( array ) , len ( array [ 0 ] ) ) } )
        elif isarray ( array , lambda i : isinstance ( i , int ) ) : # 2D int array
            header . update ( { 'mclass' : 'mxINT32_CLASS' , 'mtp' : 'miINT32' , 'dims' : ( len ( array ) , len ( array [ 0 ] ) ) } )
        elif isarray ( array , lambda i : isinstance ( i , ( int , float ) ) ) : # 2D double array
            header . update ( { 'mclass' : 'mxDOUBLE_CLASS' , 'mtp' : 'miDOUBLE' , 'dims' : ( len ( array ) , len ( array [ 0 ] ) ) } )
    elif isarray ( array , lambda i : isinstance ( i , ( int , float , basestring , Sequence , Mapping ) ) ) : # mixed contents , make it a cell array
        header . update ( { 'mclass' : 'mxCELL_CLASS' , 'dims' : ( 1 , len ( array ) ) } )
if not header :
    raise ValueError ( 'Only dicts, two dimensional numeric, ' 'and char arrays are currently supported' )
header [ 'name' ] = name
return header , array


signature:
def savemat ( filename , data ) :
    """Save data to MAT - file :
 savemat ( filename , data )
 The filename argument is either a string with the filename , or
 a file like object .
 The parameter ` ` data ` ` shall be a dict with the variables .
 A ` ` ValueError ` ` exception is raised if data has invalid format , or if the
 data structure cannot be mapped to a known MAT array type ."""

implementation:
if not isinstance ( data , Mapping ) :
    raise ValueError ( 'Data should be a dict of variable arrays' )
if isinstance ( filename , basestring ) :
    fd = open ( filename , 'wb' )
else :
    fd = filename
write_file_header ( fd )
# write variables
for name , array in data . items ( ) :
    write_compressed_var_array ( fd , array , name )
fd . close ( )


signature:
def _execute ( self , command , data = None , unpack = True ) :
    """Private method to execute command .
 Args :
 command ( Command ) : The defined command .
 data ( dict ) : The uri variable and body .
 uppack ( bool ) : If unpack value from result .
 Returns :
 The unwrapped value field in the json response ."""

implementation:
if not data :
    data = { }
if self . session_id is not None :
    data . setdefault ( 'session_id' , self . session_id )
data = self . _wrap_el ( data )
res = self . remote_invoker . execute ( command , data )
ret = WebDriverResult . from_object ( res )
ret . raise_for_status ( )
ret . value = self . _unwrap_el ( ret . value )
if not unpack :
    return ret
return ret . value


signature:
def _unwrap_el ( self , value ) :
    """Convert { ' Element ' : 1234 } to WebElement Object
 Args :
 value ( str | list | dict ) : The value field in the json response .
 Returns :
 The unwrapped value ."""

implementation:
if isinstance ( value , dict ) and 'ELEMENT' in value :
    element_id = value . get ( 'ELEMENT' )
    return WebElement ( element_id , self )
elif isinstance ( value , list ) and not isinstance ( value , str ) :
    return [ self . _unwrap_el ( item ) for item in value ]
else :
    return value


signature:
def _wrap_el ( self , value ) :
    """Convert WebElement Object to { ' Element ' : 1234}
 Args :
 value ( str | list | dict ) : The local value .
 Returns :
 The wrapped value ."""

implementation:
if isinstance ( value , dict ) :
    return { k : self . _wrap_el ( v ) for k , v in value . items ( ) }
elif isinstance ( value , WebElement ) :
    return { 'ELEMENT' : value . element_id }
elif isinstance ( value , list ) and not isinstance ( value , str ) :
    return [ self . _wrap_el ( item ) for item in value ]
else :
    return value


signature:
def init ( self ) :
    """Create Session by desiredCapabilities
 Support :
 Android iOS Web ( WebView )
 Returns :
 WebDriver Object ."""

implementation:
resp = self . _execute ( Command . NEW_SESSION , { 'desiredCapabilities' : self . desired_capabilities } , False )
resp . raise_for_status ( )
self . session_id = str ( resp . session_id )
self . capabilities = resp . value


signature:
def switch_to_window ( self , window_name ) :
    """Switch to the given window .
 Support :
 Web ( WebView )
 Args :
 window _ name ( str ) : The window to change focus to .
 Returns :
 WebDriver Object ."""

implementation:
data = { 'name' : window_name }
self . _execute ( Command . SWITCH_TO_WINDOW , data )


signature:
def set_window_size ( self , width , height , window_handle = 'current' ) :
    """Sets the width and height of the current window .
 Support :
 Web ( WebView )
 Args :
 width ( int ) : the width in pixels .
 height ( int ) : the height in pixels .
 window _ handle ( str ) : Identifier of window _ handle ,
 default to ' current ' .
 Returns :
 WebDriver Object ."""

implementation: self . _execute ( Command . SET_WINDOW_SIZE , { 'width' : int ( width ) , 'height' : int ( height ) , 'window_handle' : window_handle } )


signature:
def set_window_position ( self , x , y , window_handle = 'current' ) :
    """Sets the x , y position of the current window .
 Support :
 Web ( WebView )
 Args :
 x ( int ) : the x - coordinate in pixels .
 y ( int ) : the y - coordinate in pixels .
 window _ handle ( str ) : Identifier of window _ handle ,
 default to ' current ' .
 Returns :
 WebDriver Object ."""

implementation: self . _execute ( Command . SET_WINDOW_POSITION , { 'x' : int ( x ) , 'y' : int ( y ) , 'window_handle' : window_handle } )


signature:
def move_to ( self , element , x = 0 , y = 0 ) :
    """Deprecated use element . touch ( ' drag ' , { toX , toY , duration ( s ) } ) instead .
 Move the mouse by an offset of the specificed element .
 Support :
 Android
 Args :
 element ( WebElement ) : WebElement Object .
 x ( float ) : X offset to move to , relative to the
 top - left corner of the element .
 y ( float ) : Y offset to move to , relative to the
 top - left corner of the element .
 Returns :
 WebDriver object ."""

implementation: self . _execute ( Command . MOVE_TO , { 'element' : element . element_id , 'x' : x , 'y' : y } )


signature:
def flick ( self , element , x , y , speed ) :
    """Deprecated use touch ( ' drag ' , { fromX , fromY , toX , toY , duration ( s ) } ) instead .
 Flick on the touch screen using finger motion events .
 This flickcommand starts at a particulat screen location .
 Support :
 iOS
 Args :
 element ( WebElement ) : WebElement Object where the flick starts .
 x ( float } : The x offset in pixels to flick by .
 y ( float ) : The y offset in pixels to flick by .
 speed ( float ) The speed in pixels per seconds .
 Returns :
 WebDriver object ."""

implementation: self . _execute ( Command . FLICK , { 'element' : element . element_id , 'x' : x , 'y' : y , 'speed' : speed } )


signature:
def switch_to_frame ( self , frame_reference = None ) :
    """Switches focus to the specified frame , by index , name , or webelement .
 Support :
 Web ( WebView )
 Args :
 frame _ reference ( None | int | WebElement ) :
 The identifier of the frame to switch to .
 None means to set to the default context .
 An integer representing the index .
 A webelement means that is an ( i ) frame to switch to .
 Otherwise throw an error .
 Returns :
 WebDriver Object ."""

implementation:
if frame_reference is not None and type ( frame_reference ) not in [ int , WebElement ] :
    raise TypeError ( 'Type of frame_reference must be None or int or WebElement' )
self . _execute ( Command . SWITCH_TO_FRAME , { 'id' : frame_reference } )


signature:
def execute_script ( self , script , * args ) :
    """Execute JavaScript Synchronously in current context .
 Support :
 Web ( WebView )
 Args :
 script : The JavaScript to execute .
 * args : Arguments for your JavaScript .
 Returns :
 Returns the return value of the function ."""

implementation: return self . _execute ( Command . EXECUTE_SCRIPT , { 'script' : script , 'args' : list ( args ) } )


signature:
def execute_async_script ( self , script , * args ) :
    """Execute JavaScript Asynchronously in current context .
 Support :
 Web ( WebView )
 Args :
 script : The JavaScript to execute .
 * args : Arguments for your JavaScript .
 Returns :
 Returns the return value of the function ."""

implementation: return self . _execute ( Command . EXECUTE_ASYNC_SCRIPT , { 'script' : script , 'args' : list ( args ) } )


signature:
def add_cookie ( self , cookie_dict ) :
    """Set a cookie .
 Support :
 Web ( WebView )
 Args :
 cookie _ dict : A dictionary contain keys : " name " , " value " ,
 [ " path " ] , [ " domain " ] , [ " secure " ] , [ " httpOnly " ] , [ " expiry " ] .
 Returns :
 WebElement Object ."""

implementation:
if not isinstance ( cookie_dict , dict ) :
    raise TypeError ( 'Type of the cookie must be a dict.' )
if not cookie_dict . get ( 'name' , None ) or not cookie_dict . get ( 'value' , None ) :
    raise KeyError ( 'Missing required keys, \'name\' and \'value\' must be provided.' )
self . _execute ( Command . ADD_COOKIE , { 'cookie' : cookie_dict } )


signature:
def save_screenshot ( self , filename , quietly = False ) :
    """Save the screenshot to local .
 Support :
 Android iOS Web ( WebView )
 Args :
 filename ( str ) : The path to save the image .
 quietly ( bool ) : If True , omit the IOError when
 failed to save the image .
 Returns :
 WebElement Object .
 Raises :
 WebDriverException .
 IOError ."""

implementation:
imgData = self . take_screenshot ( )
try :
    with open ( filename , "wb" ) as f :
        f . write ( b64decode ( imgData . encode ( 'ascii' ) ) )
except IOError as err :
    if not quietly :
        raise err


signature:
def element ( self , using , value ) :
    """Find an element in the current context .
 Support :
 Android iOS Web ( WebView )
 Args :
 using ( str ) : The element location strategy .
 value ( str ) : The value of the location strategy .
 Returns :
 WebElement Object .
 Raises :
 WebDriverException ."""

implementation: return self . _execute ( Command . FIND_ELEMENT , { 'using' : using , 'value' : value } )


signature:
def element_if_exists ( self , using , value ) :
    """Check if an element in the current context .
 Support :
 Android iOS Web ( WebView )
 Args :
 using ( str ) : The element location strategy .
 value ( str ) : The value of the location strategy .
 Returns :
 Return True if the element does exists and return False otherwise .
 Raises :
 WebDriverException ."""

implementation:
try :
    self . _execute ( Command . FIND_ELEMENT , { 'using' : using , 'value' : value } )
    return True
except :
    return False


signature:
def element_or_none ( self , using , value ) :
    """Check if an element in the current context .
 Support :
 Android iOS Web ( WebView )
 Args :
 using ( str ) : The element location strategy .
 value ( str ) : The value of the location strategy .
 Returns :
 Return Element if the element does exists and return None otherwise .
 Raises :
 WebDriverException ."""

implementation:
try :
    return self . _execute ( Command . FIND_ELEMENT , { 'using' : using , 'value' : value } )
except :
    return None


signature:
def elements ( self , using , value ) :
    """Find elements in the current context .
 Support :
 Android iOS Web ( WebView )
 Args :
 using ( str ) : The element location strategy .
 value ( str ) : The value of the location strategy .
 Returns :
 Return a List < Element | None > , if no element matched , the list is empty .
 Raises :
 WebDriverException ."""

implementation: return self . _execute ( Command . FIND_ELEMENTS , { 'using' : using , 'value' : value } )


signature:
def wait_for ( self , timeout = 10000 , interval = 1000 , asserter = lambda x : x ) :
    """Wait for driver till satisfy the given condition
 Support :
 Android iOS Web ( WebView )
 Args :
 timeout ( int ) : How long we should be retrying stuff .
 interval ( int ) : How long between retries .
 asserter ( callable ) : The asserter func to determine the result .
 Returns :
 Return the driver .
 Raises :
 WebDriverException ."""

implementation:
if not callable ( asserter ) :
    raise TypeError ( 'Asserter must be callable.' )
@ retry ( retry_on_exception = lambda ex : isinstance ( ex , WebDriverException ) , stop_max_delay = timeout , wait_fixed = interval )
def _wait_for ( driver ) :
    asserter ( driver )
    return driver
return _wait_for ( self )


signature:
def wait_for_element ( self , using , value , timeout = 10000 , interval = 1000 , asserter = is_displayed ) :
    """Wait for element till satisfy the given condition
 Support :
 Android iOS Web ( WebView )
 Args :
 using ( str ) : The element location strategy .
 value ( str ) : The value of the location strategy .
 timeout ( int ) : How long we should be retrying stuff .
 interval ( int ) : How long between retries .
 asserter ( callable ) : The asserter func to determine the result .
 Returns :
 Return the Element .
 Raises :
 WebDriverException ."""

implementation:
if not callable ( asserter ) :
    raise TypeError ( 'Asserter must be callable.' )
@ retry ( retry_on_exception = lambda ex : isinstance ( ex , WebDriverException ) , stop_max_delay = timeout , wait_fixed = interval )
def _wait_for_element ( ctx , using , value ) :
    el = ctx . element ( using , value )
    asserter ( el )
    return el
return _wait_for_element ( self , using , value )


signature:
def wait_for_elements ( self , using , value , timeout = 10000 , interval = 1000 , asserter = is_displayed ) :
    """Wait for elements till satisfy the given condition
 Support :
 Android iOS Web ( WebView )
 Args :
 using ( str ) : The element location strategy .
 value ( str ) : The value of the location strategy .
 timeout ( int ) : How long we should be retrying stuff .
 interval ( int ) : How long between retries .
 asserter ( callable ) : The asserter func to determine the result .
 Returns :
 Return the list of Element if any of them satisfy the condition .
 Raises :
 WebDriverException ."""

implementation:
if not callable ( asserter ) :
    raise TypeError ( 'Asserter must be callable.' )
@ retry ( retry_on_exception = lambda ex : isinstance ( ex , WebDriverException ) , stop_max_delay = timeout , wait_fixed = interval )
def _wait_for_elements ( ctx , using , value ) :
    els = ctx . elements ( using , value )
    if not len ( els ) :
        raise WebDriverException ( 'no such element' )
    else :
        el = els [ 0 ]
        asserter ( el )
        return els
return _wait_for_elements ( self , using , value )


signature:
def from_object ( cls , obj ) :
    """The factory method to create WebDriverResult from JSON Object .
 Args :
 obj ( dict ) : The JSON Object returned by server ."""

implementation: return cls ( obj . get ( 'sessionId' , None ) , obj . get ( 'status' , 0 ) , obj . get ( 'value' , None ) )


signature:
def raise_for_status ( self ) :
    """Raise WebDriverException if returned status is not zero ."""

implementation:
if not self . status :
    return
error = find_exception_by_code ( self . status )
message = None
screen = None
stacktrace = None
if isinstance ( self . value , str ) :
    message = self . value
elif isinstance ( self . value , dict ) :
    message = self . value . get ( 'message' , None )
    screen = self . value . get ( 'screen' , None )
    stacktrace = self . value . get ( 'stacktrace' , None )
raise WebDriverException ( error , message , screen , stacktrace )


signature:
def add_element_extension_method ( Klass ) :
    """Add element _ by alias and extension ' methods ( if _ exists / or _ none ) ."""

implementation:
def add_element_method ( Klass , using ) :
    locator = using . name . lower ( )
    find_element_name = "element_by_" + locator
    find_element_if_exists_name = "element_by_" + locator + "_if_exists"
    find_element_or_none_name = "element_by_" + locator + "_or_none"
    wait_for_element_name = "wait_for_element_by_" + locator
    find_elements_name = "elements_by_" + locator
    wait_for_elements_name = "wait_for_elements_by_" + locator
    def find_element ( self , value ) :
        return self . element ( using . value , value )
    find_element . __name__ = find_element_name
    find_element . __doc__ = ( "Set parameter 'using' to '{0}'.\n" . format ( using . value ) + "See more in \'element\' method." )
    def find_element_if_exists ( self , value ) :
        return self . element_if_exists ( using . value , value )
    find_element_if_exists . __name__ = find_element_if_exists_name
    find_element_if_exists . __doc__ = ( "Set parameter 'using' to '{0}'.\n" . format ( using . value ) + "See more in \'element_if_exists\' method." )
    def find_element_or_none ( self , value ) :
        return self . element_or_none ( using . value , value )
    find_element_or_none . __name__ = find_element_or_none_name
    find_element_or_none . __doc__ = ( "Set parameter 'using' to '{0}'.\n" . format ( using . value ) + "See more in \'element_or_none\' method." )
    def wait_for_element_by ( self , * args , ** kwargs ) :
        return self . wait_for_element ( using . value , * args , ** kwargs )
    wait_for_element_by . __name__ = wait_for_element_name
    wait_for_element_by . __doc__ = ( "Set parameter 'using' to '{0}'.\n" . format ( using . value ) + "See more in \'wait_for_element\' method." )
    def find_elements ( self , value ) :
        return self . elements ( using . value , value )
    find_elements . __name__ = find_elements_name
    find_elements . __doc__ = ( "Set parameter 'using' to '{0}'.\n" . format ( using . value ) + "See more in \'elements\' method." )
    def wait_for_elements_available ( self , * args , ** kwargs ) :
        return self . wait_for_elements ( using . value , * args , ** kwargs )
    wait_for_elements_available . __name__ = wait_for_elements_name
    wait_for_elements_available . __doc__ = ( "Set parameter 'using' to '{0}'.\n" . format ( using . value ) + "See more in \'wait_for_elements\' method." )
    setattr ( Klass , find_element_name , find_element )
    setattr ( Klass , find_element_if_exists_name , find_element_if_exists )
    setattr ( Klass , find_element_or_none_name , find_element_or_none )
    setattr ( Klass , wait_for_element_name , wait_for_element_by )
    setattr ( Klass , find_elements_name , find_elements )
    setattr ( Klass , wait_for_elements_name , wait_for_elements_available )
for locator in iter ( Locator ) :
    add_element_method ( Klass , locator )


signature:
def fluent ( func ) :
    """Fluent interface decorator to return self if method return None ."""

implementation:
@ wraps ( func )
def fluent_interface ( instance , * args , ** kwargs ) :
    ret = func ( instance , * args , ** kwargs )
    if ret is not None :
        return ret
    return instance
return fluent_interface


signature:
def value_to_key_strokes ( value ) :
    """Convert value to a list of key strokes
 > > > value _ to _ key _ strokes ( 123)
 [ ' 123 ' ]
 > > > value _ to _ key _ strokes ( ' 123 ' )
 [ ' 123 ' ]
 > > > value _ to _ key _ strokes ( [ 1 , 2 , 3 ] )
 [ ' 123 ' ]
 > > > value _ to _ key _ strokes ( [ ' 1 ' , ' 2 ' , ' 3 ' ] )
 [ ' 123 ' ]
 Args :
 value ( int | str | list )
 Returns :
 A list of string ."""

implementation:
result = ''
if isinstance ( value , Integral ) :
    value = str ( value )
for v in value :
    if isinstance ( v , Keys ) :
        result += v . value
    elif isinstance ( v , Integral ) :
        result += str ( v )
    else :
        result += v
return [ result ]


signature:
def value_to_single_key_strokes ( value ) :
    """Convert value to a list of key strokes
 > > > value _ to _ single _ key _ strokes ( 123)
 [ ' 1 ' , ' 2 ' , ' 3 ' ]
 > > > value _ to _ single _ key _ strokes ( ' 123 ' )
 [ ' 1 ' , ' 2 ' , ' 3 ' ]
 > > > value _ to _ single _ key _ strokes ( [ 1 , 2 , 3 ] )
 [ ' 1 ' , ' 2 ' , ' 3 ' ]
 > > > value _ to _ single _ key _ strokes ( [ ' 1 ' , ' 2 ' , ' 3 ' ] )
 [ ' 1 ' , ' 2 ' , ' 3 ' ]
 Args :
 value ( int | str | list )
 Returns :
 A list of string ."""

implementation:
result = [ ]
if isinstance ( value , Integral ) :
    value = str ( value )
for v in value :
    if isinstance ( v , Keys ) :
        result . append ( v . value )
    elif isinstance ( v , Integral ) :
        result . append ( str ( v ) )
    else :
        result . append ( v )
return result


signature:
def check_unused_args ( self , used_args , args , kwargs ) :
    """Implement the check _ unused _ args in superclass ."""

implementation:
for k , v in kwargs . items ( ) :
    if k in used_args :
        self . _used_kwargs . update ( { k : v } )
    else :
        self . _unused_kwargs . update ( { k : v } )


signature:
def vformat ( self , format_string , args , kwargs ) :
    """Clear used and unused dicts before each formatting ."""

implementation:
self . _used_kwargs = { }
self . _unused_kwargs = { }
return super ( MemorizeFormatter , self ) . vformat ( format_string , args , kwargs )


signature:
def format_map ( self , format_string , mapping ) :
    """format a string by a map
 Args :
 format _ string ( str ) : A format string
 mapping ( dict ) : A map to format the string
 Returns :
 A formatted string .
 Raises :
 KeyError : if key is not provided by the given map ."""

implementation: return self . vformat ( format_string , args = None , kwargs = mapping )


signature:
def find_exception_by_code ( code ) :
    """Find name of exception by WebDriver defined error code .
 Args :
 code ( str ) : Error code defined in protocol .
 Returns :
 The error name defined in protocol ."""

implementation:
errorName = None
for error in WebDriverError :
    if error . value . code == code :
        errorName = error
        break
return errorName


signature:
def execute ( self , command , data = { } ) :
    """Format the endpoint url by data and then request the remote server .
 Args :
 command ( Command ) : WebDriver command to be executed .
 data ( dict ) : Data fulfill the uri template and json body .
 Returns :
 A dict represent the json body from server response .
 Raises :
 KeyError : Data cannot fulfill the variable which command needed .
 ConnectionError : Meet network problem ( e . g . DNS failure ,
 refused connection , etc ) .
 Timeout : A request times out .
 HTTPError : HTTP request returned an unsuccessful status code ."""

implementation:
method , uri = command
try :
    path = self . _formatter . format_map ( uri , data )
    body = self . _formatter . get_unused_kwargs ( )
    url = "{0}{1}" . format ( self . _url , path )
    return self . _request ( method , url , body )
except KeyError as err :
    LOGGER . debug ( 'Endpoint {0} is missing argument {1}' . format ( uri , err ) )
    raise


signature:
def _request ( self , method , url , body ) :
    """Internal method to send request to the remote server .
 Args :
 method ( str ) : HTTP Method ( GET / POST / PUT / DELET / HEAD ) .
 url ( str ) : The request url .
 body ( dict ) : The JSON object to be sent .
 Returns :
 A dict represent the json body from server response .
 Raises :
 ConnectionError : Meet network problem ( e . g . DNS failure ,
 refused connection , etc ) .
 Timeout : A request times out .
 HTTPError : HTTP request returned an unsuccessful status code ."""

implementation:
if method != 'POST' and method != 'PUT' :
    body = None
s = Session ( )
LOGGER . debug ( 'Method: {0}, Url: {1}, Body: {2}.' . format ( method , url , body ) )
req = Request ( method , url , json = body )
prepped = s . prepare_request ( req )
res = s . send ( prepped , timeout = self . _timeout or None )
res . raise_for_status ( )
# TODO try catch
return res . json ( )


signature:
def _execute ( self , command , data = None , unpack = True ) :
    """Private method to execute command with data .
 Args :
 command ( Command ) : The defined command .
 data ( dict ) : The uri variable and body .
 Returns :
 The unwrapped value field in the json response ."""

implementation:
if not data :
    data = { }
data . setdefault ( 'element_id' , self . element_id )
return self . _driver . _execute ( command , data , unpack )


signature:
def element ( self , using , value ) :
    """find an element in the current element .
 Support :
 Android iOS Web ( WebView )
 Args :
 using ( str ) : The element location strategy .
 value ( str ) : The value of the location strategy .
 Returns :
 WebElement Object .
 Raises :
 WebDriverException ."""

implementation: return self . _execute ( Command . FIND_CHILD_ELEMENT , { 'using' : using , 'value' : value } )


signature:
def element_or_none ( self , using , value ) :
    """Check if an element in the current element .
 Support :
 Android iOS Web ( WebView )
 Args :
 using ( str ) : The element location strategy .
 value ( str ) : The value of the location strategy .
 Returns :
 Return Element if the element does exists and return None otherwise .
 Raises :
 WebDriverException ."""

implementation:
try :
    return self . _execute ( Command . FIND_CHILD_ELEMENT , { 'using' : using , 'value' : value } )
except :
    return None


signature:
def elements ( self , using , value ) :
    """find elements in the current element .
 Support :
 Android iOS Web ( WebView )
 Args :
 using ( str ) : The element location strategy .
 value ( str ) : The value of the location strategy .
 Returns :
 Return a List < Element | None > , if no element matched , the list is empty .
 Raises :
 WebDriverException ."""

implementation: return self . _execute ( Command . FIND_CHILD_ELEMENTS , { 'using' : using , 'value' : value } )


signature:
def move_to ( self , x = 0 , y = 0 ) :
    """Deprecated use element . touch ( ' drag ' , { toX , toY , duration ( s ) } ) instead .
 Move the mouse by an offset of the specificed element .
 Support :
 Android
 Args :
 x ( float ) : X offset to move to , relative to the
 top - left corner of the element .
 y ( float ) : Y offset to move to , relative to the
 top - left corner of the element .
 Returns :
 WebElement object ."""

implementation: self . _driver . move_to ( self , x , y )


signature:
def flick ( self , x , y , speed ) :
    """Deprecated use touch ( ' drag ' , { fromX , fromY , toX , toY , duration ( s ) } ) instead .
 Flick on the touch screen using finger motion events .
 This flickcommand starts at a particulat screen location .
 Support :
 iOS
 Args :
 x ( float } : The x offset in pixels to flick by .
 y ( float ) : The y offset in pixels to flick by .
 speed ( float ) The speed in pixels per seconds .
 Returns :
 WebElement object ."""

implementation: self . _driver . flick ( self , x , y , speed )


signature:
def touch ( self , name , args = None ) :
    """Apply touch actions on devices . Such as , tap / doubleTap / press / pinch / rotate / drag .
 See more on https : / / github . com / alibaba / macaca / issues / 366.
 Support :
 Android iOS
 Args :
 name ( str ) : Name of the action
 args ( dict ) : Arguments of the action
 Returns :
 WebDriver Object .
 Raises :
 WebDriverException ."""

implementation:
if isinstance ( name , list ) and not isinstance ( name , str ) :
    for obj in name :
        obj [ 'element' ] = self . element_id
    actions = name
elif isinstance ( name , str ) :
    if not args :
        args = { }
    args [ 'type' ] = name
    args [ 'element' ] = self . element_id
    actions = [ args ]
else :
    raise TypeError ( 'Invalid parameters.' )
self . _driver . _execute ( Command . PERFORM_ACTIONS , { 'actions' : actions } )


signature:
def is_displayed ( target ) :
    """Assert whether the target is displayed
 Args :
 target ( WebElement ) : WebElement Object .
 Returns :
 Return True if the element is displayed or return False otherwise ."""

implementation:
is_displayed = getattr ( target , 'is_displayed' , None )
if not is_displayed or not callable ( is_displayed ) :
    raise TypeError ( 'Target has no attribute \'is_displayed\' or not callable' )
if not is_displayed ( ) :
    raise WebDriverException ( 'element not visible' )


signature:
def PlugIn ( self ) :
    """Take next available controller id and plug in to Virtual USB Bus"""

implementation:
ids = self . available_ids ( )
if len ( ids ) == 0 :
    raise MaxInputsReachedError ( 'Max Inputs Reached' )
self . id = ids [ 0 ]
_xinput . PlugIn ( self . id )
while self . id in self . available_ids ( ) :
    pass


signature:
def UnPlug ( self , force = False ) :
    """Unplug controller from Virtual USB Bus and free up ID"""

implementation:
if force :
    _xinput . UnPlugForce ( c_uint ( self . id ) )
else :
    _xinput . UnPlug ( c_uint ( self . id ) )
while self . id not in self . available_ids ( ) :
    if self . id == 0 :
        break


signature:
def set_value ( self , control , value = None ) :
    """Set a value on the controller
 If percent is True all controls will accept a value between - 1.0 and 1.0
 If not then :
 Triggers are 0 to 255
 Axis are - 32768 to 32767
 Control List :
 AxisLx , Left Stick X - Axis
 AxisLy , Left Stick Y - Axis
 AxisRx , Right Stick X - Axis
 AxisRy , Right Stick Y - Axis
 BtnBack , Menu / Back Button
 BtnStart , Start Button
 BtnA , A Button
 BtnB , B Button
 BtnX , X Button
 BtnY , Y Button
 BtnThumbL , Left Thumbstick Click
 BtnThumbR , Right Thumbstick Click
 BtnShoulderL , Left Shoulder Button
 BtnShoulderR , Right Shoulder Button
 Dpad , Set Dpad Value ( 0 = Off , Use DPAD _ # # # Constants )
 TriggerL , Left Trigger
 TriggerR , Right Trigger"""

implementation:
func = getattr ( _xinput , 'Set' + control )
if 'Axis' in control :
    target_type = c_short
    if self . percent :
        target_value = int ( 32767 * value )
    else :
        target_value = value
elif 'Btn' in control :
    target_type = c_bool
    target_value = bool ( value )
elif 'Trigger' in control :
    target_type = c_byte
    if self . percent :
        target_value = int ( 255 * value )
    else :
        target_value = value
elif 'Dpad' in control :
    target_type = c_int
    target_value = int ( value )
func ( c_uint ( self . id ) , target_type ( target_value ) )


signature:
def main ( ) :
    """Test the functionality of the rController object"""

implementation:
import time
print ( 'Testing controller in position 1:' )
print ( 'Running 3 x 3 seconds tests' )
# Initialise Controller
con = rController ( 1 )
# Loop printing controller state and buttons held
for i in range ( 3 ) :
    print ( 'Waiting...' )
    time . sleep ( 2.5 )
    print ( 'State: ' , con . gamepad )
    print ( 'Buttons: ' , con . buttons )
    time . sleep ( 0.5 )
print ( 'Done!' )


signature:
def gamepad ( self ) :
    """Returns the current gamepad state . Buttons pressed is shown as a raw integer value .
 Use rController . buttons for a list of buttons pressed ."""

implementation:
state = _xinput_state ( )
_xinput . XInputGetState ( self . ControllerID - 1 , pointer ( state ) )
self . dwPacketNumber = state . dwPacketNumber
return state . XINPUT_GAMEPAD


signature:
def buttons ( self ) :
    """Returns a list of buttons currently pressed"""

implementation: return [ name for name , value in rController . _buttons . items ( ) if self . gamepad . wButtons & value == value ]


signature:
def maybe_decode_header ( header ) :
    """Decodes an encoded 7 - bit ASCII header value into it ' s actual value ."""

implementation:
value , encoding = decode_header ( header ) [ 0 ]
if encoding :
    return value . decode ( encoding )
else :
    return value


signature:
def autodiscover ( ) :
    """Imports all available previews classes ."""

implementation:
from django . conf import settings
for application in settings . INSTALLED_APPS :
    module = import_module ( application )
    if module_has_submodule ( module , 'emails' ) :
        emails = import_module ( '%s.emails' % application )
        try :
            import_module ( '%s.emails.previews' % application )
        except ImportError : # Only raise the exception if this module contains previews and
        # there was a problem importing them . ( An emails module that
        # does not contain previews is not an error . )
            if module_has_submodule ( emails , 'previews' ) :
                raise


signature:
def register ( self , cls ) :
    """Adds a preview to the index ."""

implementation:
preview = cls ( site = self )
logger . debug ( 'Registering %r with %r' , preview , self )
index = self . __previews . setdefault ( preview . module , { } )
index [ cls . __name__ ] = preview


signature:
def detail_view ( self , request , module , preview ) :
    """Looks up a preview in the index , returning a detail view response ."""

implementation:
try :
    preview = self . __previews [ module ] [ preview ]
except KeyError :
    raise Http404
    # The provided module / preview does not exist in the index .
return preview . detail_view ( request )


signature:
def url ( self ) :
    """The URL to access this preview ."""

implementation: return reverse ( '%s:detail' % URL_NAMESPACE , kwargs = { 'module' : self . module , 'preview' : type ( self ) . __name__ , } )


signature:
def detail_view ( self , request ) :
    """Renders the message view to a response ."""

implementation:
context = { 'preview' : self , }
kwargs = { }
if self . form_class :
    if request . GET :
        form = self . form_class ( data = request . GET )
    else :
        form = self . form_class ( )
    context [ 'form' ] = form
    if not form . is_bound or not form . is_valid ( ) :
        return render ( request , 'mailviews/previews/detail.html' , context )
    kwargs . update ( form . get_message_view_kwargs ( ) )
message_view = self . get_message_view ( request , ** kwargs )
message = message_view . render_to_message ( )
raw = message . message ( )
headers = OrderedDict ( ( header , maybe_decode_header ( raw [ header ] ) ) for header in self . headers )
context . update ( { 'message' : message , 'subject' : message . subject , 'body' : message . body , 'headers' : headers , 'raw' : raw . as_string ( ) , } )
alternatives = getattr ( message , 'alternatives' , [ ] )
try :
    html = next ( alternative [ 0 ] for alternative in alternatives if alternative [ 1 ] == 'text/html' )
    context . update ( { 'html' : html , 'escaped_html' : b64encode ( html . encode ( 'utf-8' ) ) , } )
except StopIteration :
    pass
return render ( request , self . template_name , context )


signature:
def split_docstring ( value ) :
    """Splits the docstring of the given value into it ' s summary and body .
 : returns : a 2 - tuple of the format ` ` ( summary , body ) ` `"""

implementation:
docstring = textwrap . dedent ( getattr ( value , '__doc__' , '' ) )
if not docstring :
    return None
pieces = docstring . strip ( ) . split ( '\n\n' , 1 )
try :
    body = pieces [ 1 ]
except IndexError :
    body = None
return Docstring ( pieces [ 0 ] , body )


signature:
def render_to_message ( self , extra_context = None , ** kwargs ) :
    """Renders and returns an unsent message with the provided context .
 Any extra keyword arguments passed will be passed through as keyword
 arguments to the message constructor .
 : param extra _ context : Any additional context to use when rendering the
 templated content .
 : type extra _ context : : class : ` dict `
 : returns : A message instance .
 : rtype : : attr : ` . message _ class `"""

implementation:
if extra_context is None :
    extra_context = { }
# Ensure our custom headers are added to the underlying message class .
kwargs . setdefault ( 'headers' , { } ) . update ( self . headers )
context = self . get_context_data ( ** extra_context )
return self . message_class ( subject = self . render_subject ( context ) , body = self . render_body ( context ) , ** kwargs )


signature:
def send ( self , extra_context = None , ** kwargs ) :
    """Renders and sends an email message .
 All keyword arguments other than ` ` extra _ context ` ` are passed through
 as keyword arguments when constructing a new : attr : ` message _ class `
 instance for this message .
 This method exists primarily for convenience , and the proper
 rendering of your message should not depend on the behavior of this
 method . To alter how a message is created , override
 : meth : ` ` render _ to _ message ` ` instead , since that should always be
 called , even if a message is not sent .
 : param extra _ context : Any additional context data that will be used
 when rendering this message .
 : type extra _ context : : class : ` dict `"""

implementation:
message = self . render_to_message ( extra_context = extra_context , ** kwargs )
return message . send ( )


signature:
def render_subject ( self , context ) :
    """Renders the message subject for the given context .
 The context data is automatically unescaped to avoid rendering HTML
 entities in ` ` text / plain ` ` content .
 : param context : The context to use when rendering the subject template .
 : type context : : class : ` ~ django . template . Context `
 : returns : A rendered subject .
 : rtype : : class : ` str `"""

implementation:
rendered = self . subject_template . render ( unescape ( context ) )
return rendered . strip ( )


signature:
def render_to_message ( self , extra_context = None , * args , ** kwargs ) :
    """Renders and returns an unsent message with the given context .
 Any extra keyword arguments passed will be passed through as keyword
 arguments to the message constructor .
 : param extra _ context : Any additional context to use when rendering
 templated content .
 : type extra _ context : : class : ` dict `
 : returns : A message instance .
 : rtype : : attr : ` . message _ class `"""

implementation:
message = super ( TemplatedHTMLEmailMessageView , self ) . render_to_message ( extra_context , * args , ** kwargs )
if extra_context is None :
    extra_context = { }
context = self . get_context_data ( ** extra_context )
content = self . render_html_body ( context )
message . attach_alternative ( content , mimetype = 'text/html' )
return message


signature:
def timestamp ( _ , dt ) :
    'get microseconds since 2000-01-01 00:00'

implementation:
# see http : / / stackoverflow . com / questions / 2956886/
dt = util . to_utc ( dt )
unix_timestamp = calendar . timegm ( dt . timetuple ( ) )
# timetuple doesn ' t maintain microseconds
# see http : / / stackoverflow . com / a / 14369386/519015
val = ( ( unix_timestamp - psql_epoch ) * 1000000 ) + dt . microsecond
return ( 'iq' , ( 8 , val ) )


signature:
def numeric ( _ , n ) :
    """NBASE = 1000
 ndigits = total number of base - NBASE digits
 weight = base - NBASE weight of first digit
 sign = 0x0000 if positive , 0x4000 if negative , 0xC000 if nan
 dscale = decimal digits after decimal place"""

implementation:
try :
    nt = n . as_tuple ( )
except AttributeError :
    raise TypeError ( 'numeric field requires Decimal value (got %r)' % n )
digits = [ ]
if isinstance ( nt . exponent , str ) : # NaN , Inf , - Inf
    ndigits = 0
    weight = 0
    sign = 0xC000
    dscale = 0
else :
    decdigits = list ( reversed ( nt . digits + ( nt . exponent % 4 ) * ( 0 , ) ) )
    weight = 0
    while decdigits :
        if any ( decdigits [ : 4 ] ) :
            break
        weight += 1
        del decdigits [ : 4 ]
    while decdigits :
        digits . insert ( 0 , ndig ( decdigits [ : 4 ] ) )
        del decdigits [ : 4 ]
    ndigits = len ( digits )
    weight += nt . exponent // 4 + ndigits - 1
    sign = nt . sign * 0x4000
    dscale = - min ( 0 , nt . exponent )
data = [ ndigits , weight , sign , dscale ] + digits
return ( 'ihhHH%dH' % ndigits , [ 2 * len ( data ) ] + data )


signature:
def set_default_subparser ( self , name , args = None ) :
    """default subparser selection . Call after setup , just before parse _ args ( )
 name : is the name of the subparser to call by default
 args : if set is the argument list handed to parse _ args ( )
 , tested with 2.7 , 3.2 , 3.3 , 3.4
 it works with 2.6 assuming argparse is installed"""

implementation:
subparser_found = False
for arg in sys . argv [ 1 : ] :
    if arg in [ '-h' , '--help' ] : # global help if no subparser
        break
else :
    for x in self . _subparsers . _actions :
        if not isinstance ( x , argparse . _SubParsersAction ) :
            continue
        for sp_name in x . _name_parser_map . keys ( ) :
            if sp_name in sys . argv [ 1 : ] :
                subparser_found = True
    if not subparser_found : # insert default in first position , this implies no
    # global options without a sub _ parsers specified
        if args is None :
            sys . argv . insert ( 1 , name )
        else :
            args . insert ( 0 , name )


signature:
def execute_from_command_line ( argv = None ) :
    """A simple method that runs a ManagementUtility ."""

implementation:
parser = argparse . ArgumentParser ( description = __doc__ )
parser . add_argument ( '--monitors-dir' , default = MONITORS_DIR )
parser . add_argument ( '--alerts-dir' , default = ALERTS_DIR )
parser . add_argument ( '--config' , default = SMA_INI_FILE )
parser . add_argument ( '--warning' , help = 'set logging to warning' , action = 'store_const' , dest = 'loglevel' , const = logging . WARNING , default = logging . INFO )
parser . add_argument ( '--quiet' , help = 'set logging to ERROR' , action = 'store_const' , dest = 'loglevel' , const = logging . ERROR , default = logging . INFO )
parser . add_argument ( '--debug' , help = 'set logging to DEBUG' , action = 'store_const' , dest = 'loglevel' , const = logging . DEBUG , default = logging . INFO )
parser . add_argument ( '--verbose' , help = 'set logging to COMM' , action = 'store_const' , dest = 'loglevel' , const = 5 , default = logging . INFO )
parser . sub = parser . add_subparsers ( )
parse_service = parser . sub . add_parser ( 'service' , help = 'Run SMA as service (daemon).' )
parse_service . set_defaults ( which = 'service' )
parse_oneshot = parser . sub . add_parser ( 'one-shot' , help = 'Run SMA once and exit' )
parse_oneshot . set_defaults ( which = 'one-shot' )
parse_alerts = parser . sub . add_parser ( 'alerts' , help = 'Alerts options.' )
parse_alerts . set_defaults ( which = 'alerts' )
parse_alerts . add_argument ( '--test' , help = 'Test alert' , action = 'store_true' )
parse_alerts . add_argument ( 'alert_section' , nargs = '?' , help = 'Alert section to see' )
parse_results = parser . sub . add_parser ( 'results' , help = 'Monitors results' )
parse_results . set_defaults ( which = 'results' )
parser . set_default_subparser ( 'one-shot' )
args = parser . parse_args ( argv [ 1 : ] )
create_logger ( 'sma' , args . loglevel )
if not getattr ( args , 'which' , None ) or args . which == 'one-shot' :
    sma = SMA ( args . monitors_dir , args . alerts_dir , args . config )
    sma . evaluate_and_alert ( )
elif args . which == 'service' :
    sma = SMAService ( args . monitors_dir , args . alerts_dir , args . config )
    sma . start ( )
elif args . which == 'alerts' and args . test :
    sma = SMA ( args . monitors_dir , args . alerts_dir , args . config )
    sma . alerts . test ( )
elif args . which == 'results' :
    print ( SMA ( args . monitors_dir , args . alerts_dir , args . config ) . results )


signature:
def _getCallingContext ( ) :
    """Utility function for the RedisLogRecord .
 Returns the module , function , and lineno of the function
 that called the logger .
 We look way up in the stack . The stack at this point is :
 [0 ] logger . py _ getCallingContext ( hey , that ' s me ! )
 [1 ] logger . py _ _ init _ _
 [2 ] logger . py makeRecord
 [3 ] _ log
 [4 ] < logging method >
 [5 ] caller of logging method"""

implementation:
frames = inspect . stack ( )
if len ( frames ) > 4 :
    context = frames [ 5 ]
else :
    context = frames [ 0 ]
modname = context [ 1 ]
lineno = context [ 2 ]
if context [ 3 ] :
    funcname = context [ 3 ]
else :
    funcname = ""
# python docs say you don ' t want references to
# frames lying around . Bad things can happen .
del context
del frames
return modname , funcname , lineno


signature:
def format ( self , record ) :
    """JSON - encode a record for serializing through redis .
 Convert date to iso format , and stringify any exceptions ."""

implementation:
data = record . _raw . copy ( )
# serialize the datetime date as utc string
data [ 'time' ] = data [ 'time' ] . isoformat ( )
# stringify exception data
if data . get ( 'traceback' ) :
    data [ 'traceback' ] = self . formatException ( data [ 'traceback' ] )
return json . dumps ( data )


signature:
def emit ( self , record ) :
    """Publish record to redis logging channel"""

implementation:
try :
    self . redis_client . publish ( self . channel , self . format ( record ) )
except redis . RedisError :
    pass


signature:
def emit ( self , record ) :
    """Publish record to redis logging list"""

implementation:
try :
    if self . max_messages :
        p = self . redis_client . pipeline ( )
        p . rpush ( self . key , self . format ( record ) )
        p . ltrim ( self . key , - self . max_messages , - 1 )
        p . execute ( )
    else :
        self . redis_client . rpush ( self . key , self . format ( record ) )
except redis . RedisError :
    pass


signature:
def require_template_debug ( f ) :
    """Decorated function is a no - op if TEMPLATE _ DEBUG is False"""

implementation:
def _ ( * args , ** kwargs ) :
    TEMPLATE_DEBUG = getattr ( settings , 'TEMPLATE_DEBUG' , False )
    return f ( * args , ** kwargs ) if TEMPLATE_DEBUG else ''
return _


signature:
def _display_details ( var_data ) :
    """Given a dictionary of variable attribute data from get _ details display the
 data in the terminal ."""

implementation:
meta_keys = ( key for key in list ( var_data . keys ( ) ) if key . startswith ( 'META_' ) )
for key in meta_keys :
    display_key = key [ 5 : ] . capitalize ( )
    pprint ( '{0}: {1}' . format ( display_key , var_data . pop ( key ) ) )
pprint ( var_data )


signature:
def set_trace ( context ) :
    """Start a pdb set _ trace inside of the template with the context available as
 ' context ' . Uses ipdb if available ."""

implementation:
try :
    import ipdb as pdb
except ImportError :
    import pdb
    print ( "For best results, pip install ipdb." )
print ( "Variables that are available in the current context:" )
render = lambda s : template . Template ( s ) . render ( context )
availables = get_variables ( context )
pprint ( availables )
print ( 'Type `availables` to show this list.' )
print ( 'Type <variable_name> to access one.' )
print ( 'Use render("template string") to test template rendering' )
# Cram context variables into the local scope
for var in availables :
    locals ( ) [ var ] = context [ var ]
pdb . set_trace ( )
return ''


signature:
def pydevd ( context ) :
    """Start a pydev settrace"""

implementation:
global pdevd_not_available
if pdevd_not_available :
    return ''
try :
    import pydevd
except ImportError :
    pdevd_not_available = True
    return ''
render = lambda s : template . Template ( s ) . render ( context )
availables = get_variables ( context )
for var in availables :
    locals ( ) [ var ] = context [ var ]
# catch the case where no client is listening
try :
    pydevd . settrace ( )
except socket . error :
    pdevd_not_available = True
return ''


signature:
def _flatten ( iterable ) :
    """Given an iterable with nested iterables , generate a flat iterable"""

implementation:
for i in iterable :
    if isinstance ( i , Iterable ) and not isinstance ( i , string_types ) :
        for sub_i in _flatten ( i ) :
            yield sub_i
    else :
        yield i


signature:
def get_details ( var ) :
    """Given a variable inside the context , obtain the attributes / callables ,
 their values where possible , and the module name and class name if possible"""

implementation:
var_data = { }
# Obtain module and class details if available and add them in
module = getattr ( var , '__module__' , '' )
kls = getattr ( getattr ( var , '__class__' , '' ) , '__name__' , '' )
if module :
    var_data [ 'META_module_name' ] = module
if kls :
    var_data [ 'META_class_name' ] = kls
for attr in get_attributes ( var ) :
    value = _get_detail_value ( var , attr )
    if value is not None :
        var_data [ attr ] = value
return var_data


signature:
def _get_detail_value ( var , attr ) :
    """Given a variable and one of its attributes that are available inside of
 a template , return its ' method ' if it is a callable , its class name if it
 is a model manager , otherwise return its value"""

implementation:
value = getattr ( var , attr )
# Rename common Django class names
kls = getattr ( getattr ( value , '__class__' , '' ) , '__name__' , '' )
if kls in ( 'ManyRelatedManager' , 'RelatedManager' , 'EmptyManager' ) :
    return kls
if callable ( value ) :
    return 'routine'
return value


signature:
def get_attributes ( var ) :
    """Given a varaible , return the list of attributes that are available inside
 of a template"""

implementation:
is_valid = partial ( is_valid_in_template , var )
return list ( filter ( is_valid , dir ( var ) ) )


signature:
def is_valid_in_template ( var , attr ) :
    """Given a variable and one of its attributes , determine if the attribute is
 accessible inside of a Django template and return True or False accordingly"""

implementation:
# Remove private variables or methods
if attr . startswith ( '_' ) :
    return False
# Remove any attributes that raise an acception when read
try :
    value = getattr ( var , attr )
except :
    return False
if isroutine ( value ) : # Remove any routines that are flagged with ' alters _ data '
    if getattr ( value , 'alters_data' , False ) :
        return False
    else : # Remove any routines that require arguments
        try :
            argspec = getargspec ( value )
            num_args = len ( argspec . args ) if argspec . args else 0
            num_defaults = len ( argspec . defaults ) if argspec . defaults else 0
            if num_args - num_defaults > 1 :
                return False
        except TypeError : # C extension callables are routines , but getargspec fails with
        # a TypeError when these are passed .
            pass
return True