signature:
def many ( prompt , * args , ** kwargs ) :
    """Calls ` pick ` in a while loop to allow user to pick many
 options . Returns a list of chosen options ."""

implementation:
def get_options ( options , chosen ) :
    return [ options [ i ] for i , c in enumerate ( chosen ) if c ]
def get_verbose_options ( verbose_options , chosen ) :
    no , yes = ' ' , '✔'
    if sys . version_info < ( 3 , 3 ) :
        no , yes = ' ' , '@'
    opts = [ '{} {}' . format ( yes if c else no , verbose_options [ i ] ) for i , c in enumerate ( chosen ) ]
    return opts + [ '{}{}' . format ( '  ' , kwargs . get ( 'done' , 'done...' ) ) ]
options , verbose_options = prepare_options ( args )
chosen = [ False ] * len ( options )
index = kwargs . get ( 'idx' , 0 )
default = kwargs . get ( 'default' , None )
if isinstance ( default , list ) :
    for idx in default :
        chosen [ idx ] = True
if isinstance ( default , int ) :
    chosen [ default ] = True
while True :
    try :
        index = one ( prompt , * get_verbose_options ( verbose_options , chosen ) , return_index = True , idx = index )
    except QuestionnaireGoBack :
        if any ( chosen ) :
            raise QuestionnaireGoBack ( 0 )
        else :
            raise QuestionnaireGoBack
    if index == len ( options ) :
        return get_options ( options , chosen )
    chosen [ index ] = not chosen [ index ]


signature:
def prepare_options ( options ) :
    """Create options and verbose options from strings and non - string iterables in
 ` options ` array ."""

implementation:
options_ , verbose_options = [ ] , [ ]
for option in options :
    if is_string ( option ) :
        options_ . append ( option )
        verbose_options . append ( option )
    else :
        options_ . append ( option [ 0 ] )
        verbose_options . append ( option [ 1 ] )
return options_ , verbose_options


signature:
def raw ( prompt , * args , ** kwargs ) :
    """Calls input to allow user to input an arbitrary string . User can go
 back by entering the ` go _ back ` string . Works in both Python 2 and 3."""

implementation:
go_back = kwargs . get ( 'go_back' , '<' )
type_ = kwargs . get ( 'type' , str )
default = kwargs . get ( 'default' , '' )
with stdout_redirected ( sys . stderr ) :
    while True :
        try :
            if kwargs . get ( 'secret' , False ) :
                answer = getpass . getpass ( prompt )
            elif sys . version_info < ( 3 , 0 ) :
                answer = raw_input ( prompt )
            else :
                answer = input ( prompt )
            if not answer :
                answer = default
            if answer == go_back :
                raise QuestionnaireGoBack
            return type_ ( answer )
        except ValueError :
            eprint ( '\n`{}` is not a valid `{}`\n' . format ( answer , type_ ) )


signature:
def stdout_redirected ( to ) :
    """Lifted from : https : / / stackoverflow . com / questions / 4675728 / redirect - stdout - to - a - file - in - python
 This is the only way I ' ve found to redirect stdout with curses . This way the
 output from questionnaire can be piped to another program , without piping
 what ' s written to the terminal by the prompters ."""

implementation:
stdout = sys . stdout
stdout_fd = fileno ( stdout )
# copy stdout _ fd before it is overwritten
with os . fdopen ( os . dup ( stdout_fd ) , 'wb' ) as copied :
    stdout . flush ( )
    # flush library buffers that dup2 knows nothing about
    try :
        os . dup2 ( fileno ( to ) , stdout_fd )
        # $ exec > & to
    except ValueError : # filename
        with open ( to , 'wb' ) as to_file :
            os . dup2 ( to_file . fileno ( ) , stdout_fd )
            # $ exec > to
    try :
        yield stdout
        # allow code to be run with the redirected stdout
    finally : # restore stdout to its previous value
        stdout . flush ( )
        os . dup2 ( copied . fileno ( ) , stdout_fd )


signature:
def exit_on_keyboard_interrupt ( f ) :
    """Decorator that allows user to exit script by sending a keyboard interrupt
 ( ctrl + c ) without raising an exception ."""

implementation:
@ wraps ( f )
def wrapper ( * args , ** kwargs ) :
    raise_exception = kwargs . pop ( 'raise_exception' , False )
    try :
        return f ( * args , ** kwargs )
    except KeyboardInterrupt :
        if not raise_exception :
            sys . exit ( )
        raise KeyboardInterrupt
return wrapper


signature:
def get_operator ( self , op ) :
    """Assigns function to the operators property of the instance ."""

implementation:
if op in self . OPERATORS :
    return self . OPERATORS . get ( op )
try :
    n_args = len ( inspect . getargspec ( op ) [ 0 ] )
    if n_args != 2 :
        raise TypeError
except :
    eprint ( 'Error: invalid operator function. Operators must accept two args.' )
    raise
else :
    return op


signature:
def assign_prompter ( self , prompter ) :
    """If you want to change the core prompters registry , you can
 override this method in a Question subclass ."""

implementation:
if is_string ( prompter ) :
    if prompter not in prompters :
        eprint ( "Error: '{}' is not a core prompter" . format ( prompter ) )
        sys . exit ( )
    self . prompter = prompters [ prompter ]
else :
    self . prompter = prompter


signature:
def add ( self , * args , ** kwargs ) :
    """Add a Question instance to the questions dict . Each key points
 to a list of Question instances with that key . Use the ` question `
 kwarg to pass a Question instance if you want , or pass in the same
 args you would pass to instantiate a question ."""

implementation:
if 'question' in kwargs and isinstance ( kwargs [ 'question' ] , Question ) :
    question = kwargs [ 'question' ]
else :
    question = Question ( * args , ** kwargs )
self . questions . setdefault ( question . key , [ ] ) . append ( question )
return question


signature:
def ask ( self , error = None ) :
    """Asks the next question in the questionnaire and returns the answer ,
 unless user goes back ."""

implementation:
q = self . next_question
if q is None :
    return
try :
    answer = q . prompter ( self . get_prompt ( q , error ) , * q . prompter_args , ** q . prompter_kwargs )
except QuestionnaireGoBack as e :
    steps = e . args [ 0 ] if e . args else 1
    if steps == 0 :
        self . ask ( )
        # user can redo current question even if ` can _ go _ back ` is ` False `
        return
    self . go_back ( steps )
else :
    if q . _validate :
        error = q . _validate ( answer )
        if error :
            self . ask ( error )
            return
    if q . _transform :
        answer = q . _transform ( answer )
    self . answers [ q . key ] = answer
    return answer


signature:
def next_question ( self ) :
    """Returns the next ` Question ` in the questionnaire , or ` None ` if there
 are no questions left . Returns first question for whose key there is no
 answer and for which condition is satisfied , or for which there is no
 condition ."""

implementation:
for key , questions in self . questions . items ( ) :
    if key in self . answers :
        continue
    for question in questions :
        if self . check_condition ( question . _condition ) :
            return question
return None


signature:
def check_condition ( self , condition ) :
    """Helper that returns True if condition is satisfied / doesn ' t exist ."""

implementation:
if not condition :
    return True
for c in condition . conditions :
    key , value , operator = c
    if not operator ( self . answers [ key ] , value ) :
        return False
return True


signature:
def go_back ( self , n = 1 ) :
    """Move ` n ` questions back in the questionnaire by removing the last ` n `
 answers ."""

implementation:
if not self . can_go_back :
    return
N = max ( len ( self . answers ) - abs ( n ) , 0 )
self . answers = OrderedDict ( islice ( self . answers . items ( ) , N ) )


signature:
def format_answers ( self , fmt = 'obj' ) :
    """Formats answers depending on ` fmt ` ."""

implementation:
fmts = ( 'obj' , 'array' , 'plain' )
if fmt not in fmts :
    eprint ( "Error: '{}' not in {}" . format ( fmt , fmts ) )
    return
def stringify ( val ) :
    if type ( val ) in ( list , tuple ) :
        return ', ' . join ( str ( e ) for e in val )
    return val
if fmt == 'obj' :
    return json . dumps ( self . answers )
elif fmt == 'array' :
    answers = [ [ k , v ] for k , v in self . answers . items ( ) ]
    return json . dumps ( answers )
elif fmt == 'plain' :
    answers = '\n' . join ( '{}: {}' . format ( k , stringify ( v ) ) for k , v in self . answers . items ( ) )
    return answers


signature:
def answer_display ( self , s = '' ) :
    """Helper method for displaying the answers so far ."""

implementation:
padding = len ( max ( self . questions . keys ( ) , key = len ) ) + 5
for key in list ( self . answers . keys ( ) ) :
    s += '{:>{}} : {}\n' . format ( key , padding , self . answers [ key ] )
return s


signature:
def add_intent ( self , name , lines , reload_cache = False ) :
    """Creates a new intent , optionally checking the cache first
 Args :
 name ( str ) : The associated name of the intent
 lines ( list < str > ) : All the sentences that should activate the intent
 reload _ cache : Whether to ignore cached intent if exists"""

implementation:
self . intents . add ( name , lines , reload_cache )
self . padaos . add_intent ( name , lines )
self . must_train = True


signature:
def add_entity ( self , name , lines , reload_cache = False ) :
    """Adds an entity that matches the given lines .
 Example :
 self . add _ intent ( ' weather ' , [ ' will it rain on { weekday } ? ' ] )
 self . add _ entity ( ' { weekday } ' , [ ' monday ' , ' tuesday ' , ' wednesday ' ] ) # . . .
 Args :
 name ( str ) : The name of the entity
 lines ( list < str > ) : Lines of example extracted entities
 reload _ cache ( bool ) : Whether to refresh all of cache"""

implementation:
Entity . verify_name ( name )
self . entities . add ( Entity . wrap_name ( name ) , lines , reload_cache )
self . padaos . add_entity ( name , lines )
self . must_train = True


signature:
def load_entity ( self , name , file_name , reload_cache = False ) :
    """Loads an entity , optionally checking the cache first
 Args :
 name ( str ) : The associated name of the entity
 file _ name ( str ) : The location of the entity file
 reload _ cache ( bool ) : Whether to refresh all of cache"""

implementation:
Entity . verify_name ( name )
self . entities . load ( Entity . wrap_name ( name ) , file_name , reload_cache )
with open ( file_name ) as f :
    self . padaos . add_entity ( name , f . read ( ) . split ( '\n' ) )
self . must_train = True


signature:
def load_intent ( self , name , file_name , reload_cache = False ) :
    """Loads an intent , optionally checking the cache first
 Args :
 name ( str ) : The associated name of the intent
 file _ name ( str ) : The location of the intent file
 reload _ cache ( bool ) : Whether to refresh all of cache"""

implementation:
self . intents . load ( name , file_name , reload_cache )
with open ( file_name ) as f :
    self . padaos . add_intent ( name , f . read ( ) . split ( '\n' ) )
self . must_train = True


signature:
def remove_intent ( self , name ) :
    """Unload an intent"""

implementation:
self . intents . remove ( name )
self . padaos . remove_intent ( name )
self . must_train = True


signature:
def remove_entity ( self , name ) :
    """Unload an entity"""

implementation:
self . entities . remove ( name )
self . padaos . remove_entity ( name )


signature:
def train ( self , debug = True , force = False , single_thread = False , timeout = 20 ) :
    """Trains all the loaded intents that need to be updated
 If a cache file exists with the same hash as the intent file ,
 the intent will not be trained and just loaded from file
 Args :
 debug ( bool ) : Whether to print a message to stdout each time a new intent is trained
 force ( bool ) : Whether to force training if already finished
 single _ thread ( bool ) : Whether to force running in a single thread
 timeout ( float ) : Seconds before cancelling training
 Returns :
 bool : True if training succeeded without timeout"""

implementation:
if not self . must_train and not force :
    return
self . padaos . compile ( )
self . train_thread = Thread ( target = self . _train , kwargs = dict ( debug = debug , single_thread = single_thread , timeout = timeout ) , daemon = True )
self . train_thread . start ( )
self . train_thread . join ( timeout )
self . must_train = False
return not self . train_thread . is_alive ( )


signature:
def train_subprocess ( self , * args , ** kwargs ) :
    """Trains in a subprocess which provides a timeout guarantees everything shuts down properly
 Args :
 See < train >
 Returns :
 bool : True for success , False if timed out"""

implementation:
ret = call ( [ sys . executable , '-m' , 'padatious' , 'train' , self . cache_dir , '-d' , json . dumps ( self . serialized_args ) , '-a' , json . dumps ( args ) , '-k' , json . dumps ( kwargs ) , ] )
if ret == 2 :
    raise TypeError ( 'Invalid train arguments: {} {}' . format ( args , kwargs ) )
data = self . serialized_args
self . clear ( )
self . apply_training_args ( data )
self . padaos . compile ( )
if ret == 0 :
    self . must_train = False
    return True
elif ret == 10 : # timeout
    return False
else :
    raise ValueError ( 'Training failed and returned code: {}' . format ( ret ) )


signature:
def calc_intents ( self , query ) :
    """Tests all the intents against the query and returns
 data on how well each one matched against the query
 Args :
 query ( str ) : Input sentence to test against intents
 Returns :
 list < MatchData > : List of intent matches
 See calc _ intent ( ) for a description of the returned MatchData"""

implementation:
if self . must_train :
    self . train ( )
intents = { } if self . train_thread and self . train_thread . is_alive ( ) else { i . name : i for i in self . intents . calc_intents ( query , self . entities ) }
sent = tokenize ( query )
for perfect_match in self . padaos . calc_intents ( query ) :
    name = perfect_match [ 'name' ]
    intents [ name ] = MatchData ( name , sent , matches = perfect_match [ 'entities' ] , conf = 1.0 )
return list ( intents . values ( ) )


signature:
def calc_intent ( self , query ) :
    """Tests all the intents against the query and returns
 match data of the best intent
 Args :
 query ( str ) : Input sentence to test against intents
 Returns :
 MatchData : Best intent match"""

implementation:
matches = self . calc_intents ( query )
if len ( matches ) == 0 :
    return MatchData ( '' , '' )
best_match = max ( matches , key = lambda x : x . conf )
best_matches = ( match for match in matches if match . conf == best_match . conf )
return min ( best_matches , key = lambda x : sum ( map ( len , x . matches . values ( ) ) ) )


signature:
def expand ( self ) :
    """Creates a combination of all sub - sentences .
 Returns :
 List < List < str > > : A list with all subsentence expansions combined in
 every possible way"""

implementation:
old_expanded = [ [ ] ]
for sub in self . _tree :
    sub_expanded = sub . expand ( )
    new_expanded = [ ]
    while len ( old_expanded ) > 0 :
        sentence = old_expanded . pop ( )
        for new in sub_expanded :
            new_expanded . append ( sentence + new )
    old_expanded = new_expanded
return old_expanded


signature:
def expand ( self ) :
    """Returns all of its options as seperated sub - sentences .
 Returns :
 List < List < str > > : A list containing the sentences created by all
 expansions of its sub - sentences"""

implementation:
options = [ ]
for option in self . _tree :
    options . extend ( option . expand ( ) )
return options


signature:
def _parse_expr ( self ) :
    """Generate sentence token trees from the current position to
 the next closing parentheses / end of the list and return it
 [ ' 1 ' , ' ( ' , ' 2 ' , ' | ' , ' 3 , ' ) ' ] - > [ ' 1 ' , [ [ ' 2 ' ] , [ ' 3 ' ] ] ]
 [ ' 2 ' , ' | ' , ' 3 ' ] - > [ [ ' 2 ' ] , [ ' 3 ' ] ]"""

implementation:
# List of all generated sentences
sentence_list = [ ]
# Currently active sentence
cur_sentence = [ ]
sentence_list . append ( Sentence ( cur_sentence ) )
# Determine which form the current expression has
while self . _current_position < len ( self . tokens ) :
    cur = self . tokens [ self . _current_position ]
    self . _current_position += 1
    if cur == '(' : # Parse the subexpression
        subexpr = self . _parse_expr ( )
        # Check if the subexpression only has one branch
        # - > If so , append " ( " and " ) " and add it as is
        normal_brackets = False
        if len ( subexpr . tree ( ) ) == 1 :
            normal_brackets = True
            cur_sentence . append ( Word ( '(' ) )
        # add it to the sentence
        cur_sentence . append ( subexpr )
        if normal_brackets :
            cur_sentence . append ( Word ( ')' ) )
    elif cur == '|' : # Begin parsing a new sentence
        cur_sentence = [ ]
        sentence_list . append ( Sentence ( cur_sentence ) )
    elif cur == ')' : # End parsing the current subexpression
        break
    # TODO anything special about { sth } ?
    else :
        cur_sentence . append ( Word ( cur ) )
return Options ( sentence_list )


signature:
def _train_and_save ( obj , cache , data , print_updates ) :
    """Internal pickleable function used to train objects in another process"""

implementation:
obj . train ( data )
if print_updates :
    print ( 'Regenerated ' + obj . name + '.' )
obj . save ( cache )


signature:
def wrap_name ( name ) :
    """Wraps SkillName : entity into SkillName : { entity }"""

implementation:
if ':' in name :
    parts = name . split ( ':' )
    intent_name , ent_name = parts [ 0 ] , parts [ 1 : ]
    return intent_name + ':{' + ':' . join ( ent_name ) + '}'
else :
    return '{' + name + '}'


signature:
def lines_hash ( lines ) :
    """Creates a unique binary id for the given lines
 Args :
 lines ( list < str > ) : List of strings that should be collectively hashed
 Returns :
 bytearray : Binary hash"""

implementation:
x = xxh32 ( )
for i in lines :
    x . update ( i . encode ( ) )
return x . digest ( )


signature:
def tokenize ( sentence ) :
    """Converts a single sentence into a list of individual significant units
 Args :
 sentence ( str ) : Input string ie . ' This is a sentence . '
 Returns :
 list < str > : List of tokens ie . [ ' this ' , ' is ' , ' a ' , ' sentence ' ]"""

implementation:
tokens = [ ]
class Vars :
    start_pos = - 1
    last_type = 'o'
def update ( c , i ) :
    if c . isalpha ( ) or c in '-{}' :
        t = 'a'
    elif c . isdigit ( ) or c == '#' :
        t = 'n'
    elif c . isspace ( ) :
        t = 's'
    else :
        t = 'o'
    if t != Vars . last_type or t == 'o' :
        if Vars . start_pos >= 0 :
            token = sentence [ Vars . start_pos : i ] . lower ( )
            if token not in '.!?' :
                tokens . append ( token )
        Vars . start_pos = - 1 if t == 's' else i
    Vars . last_type = t
for i , char in enumerate ( sentence ) :
    update ( char , i )
update ( ' ' , len ( sentence ) )
return tokens


signature:
def resolve_conflicts ( inputs , outputs ) :
    """Checks for duplicate inputs and if there are any ,
 remove one and set the output to the max of the two outputs
 Args :
 inputs ( list < list < float > > ) : Array of input vectors
 outputs ( list < list < float > > ) : Array of output vectors
 Returns :
 tuple < inputs , outputs > : The modified inputs and outputs"""

implementation:
data = { }
for inp , out in zip ( inputs , outputs ) :
    tup = tuple ( inp )
    if tup in data :
        data [ tup ] . append ( out )
    else :
        data [ tup ] = [ out ]
inputs , outputs = [ ] , [ ]
for inp , outs in data . items ( ) :
    inputs . append ( list ( inp ) )
    combined = [ 0 ] * len ( outs [ 0 ] )
    for i in range ( len ( combined ) ) :
        combined [ i ] = max ( j [ i ] for j in outs )
    outputs . append ( combined )
return inputs , outputs


signature:
def main ( src , pyi_dir , target_dir , incremental , quiet , replace_any , hg , traceback ) :
    """Re - apply type annotations from . pyi stubs to your codebase ."""

implementation:
Config . incremental = incremental
Config . replace_any = replace_any
returncode = 0
for src_entry in src :
    for file , error , exc_type , tb in retype_path ( Path ( src_entry ) , pyi_dir = Path ( pyi_dir ) , targets = Path ( target_dir ) , src_explicitly_given = True , quiet = quiet , hg = hg , ) :
        print ( f'error: {file}: {error}' , file = sys . stderr )
        if traceback :
            print ( 'Traceback (most recent call last):' , file = sys . stderr )
            for line in tb :
                print ( line , file = sys . stderr , end = '' )
            print ( f'{exc_type.__name__}: {error}' , file = sys . stderr )
        returncode += 1
if not src and not quiet :
    print ( 'warning: no sources given' , file = sys . stderr )
# According to http : / / tldp . org / LDP / abs / html / index . html starting with 126
# we have special returncodes .
sys . exit ( min ( returncode , 125 ) )


signature:
def retype_path ( src , pyi_dir , targets , * , src_explicitly_given = False , quiet = False , hg = False ) :
    """Recursively retype files or directories given . Generate errors ."""

implementation:
if src . is_dir ( ) :
    for child in src . iterdir ( ) :
        if child == pyi_dir or child == targets :
            continue
        yield from retype_path ( child , pyi_dir / src . name , targets / src . name , quiet = quiet , hg = hg , )
elif src . suffix == '.py' or src_explicitly_given :
    try :
        retype_file ( src , pyi_dir , targets , quiet = quiet , hg = hg )
    except Exception as e :
        yield ( src , str ( e ) , type ( e ) , traceback . format_tb ( e . __traceback__ ) , )


signature:
def retype_file ( src , pyi_dir , targets , * , quiet = False , hg = False ) :
    """Retype ` src ` , finding types in ` pyi _ dir ` . Save in ` targets ` .
 The file should remain formatted exactly as it was before , save for :
 - annotations
 - additional imports needed to satisfy annotations
 - additional module - level names needed to satisfy annotations
 Type comments in sources are normalized to type annotations ."""

implementation:
with tokenize . open ( src ) as src_buffer :
    src_encoding = src_buffer . encoding
    src_node = lib2to3_parse ( src_buffer . read ( ) )
try :
    with open ( ( pyi_dir / src . name ) . with_suffix ( '.pyi' ) ) as pyi_file :
        pyi_txt = pyi_file . read ( )
except FileNotFoundError :
    if not quiet :
        print ( f'warning: .pyi file for source {src} not found in {pyi_dir}' , file = sys . stderr , )
else :
    pyi_ast = ast3 . parse ( pyi_txt )
    assert isinstance ( pyi_ast , ast3 . Module )
    reapply_all ( pyi_ast . body , src_node )
fix_remaining_type_comments ( src_node )
targets . mkdir ( parents = True , exist_ok = True )
with open ( targets / src . name , 'w' , encoding = src_encoding ) as target_file :
    target_file . write ( lib2to3_unparse ( src_node , hg = hg ) )
return targets / src . name


signature:
def lib2to3_parse ( src_txt ) :
    """Given a string with source , return the lib2to3 Node ."""

implementation:
grammar = pygram . python_grammar_no_print_statement
drv = driver . Driver ( grammar , pytree . convert )
if src_txt [ - 1 ] != '\n' :
    nl = '\r\n' if '\r\n' in src_txt [ : 1024 ] else '\n'
    src_txt += nl
try :
    result = drv . parse_string ( src_txt , True )
except ParseError as pe :
    lineno , column = pe . context [ 1 ]
    lines = src_txt . splitlines ( )
    try :
        faulty_line = lines [ lineno - 1 ]
    except IndexError :
        faulty_line = "<line number missing in source>"
    raise ValueError ( f"Cannot parse: {lineno}:{column}: {faulty_line}" ) from None
if isinstance ( result , Leaf ) :
    result = Node ( syms . file_input , [ result ] )
return result


signature:
def lib2to3_unparse ( node , * , hg = False ) :
    """Given a lib2to3 node , return its string representation ."""

implementation:
code = str ( node )
if hg :
    from retype_hgext import apply_job_security
    code = apply_job_security ( code )
return code


signature:
def reapply_all ( ast_node , lib2to3_node ) :
    """Reapplies the typed _ ast node into the lib2to3 tree .
 Also does post - processing . This is done in reverse order to enable placing
 TypeVars and aliases that depend on one another ."""

implementation:
late_processing = reapply ( ast_node , lib2to3_node )
for lazy_func in reversed ( late_processing ) :
    lazy_func ( )


signature:
def fix_remaining_type_comments ( node ) :
    """Converts type comments in ` node ` to proper annotated assignments ."""

implementation:
assert node . type == syms . file_input
last_n = None
for n in node . post_order ( ) :
    if last_n is not None :
        if n . type == token . NEWLINE and is_assignment ( last_n ) :
            fix_variable_annotation_type_comment ( n , last_n )
        elif n . type == syms . funcdef and last_n . type == syms . suite :
            fix_signature_annotation_type_comment ( n , last_n , offset = 1 )
        elif n . type == syms . async_funcdef and last_n . type == syms . suite :
            fix_signature_annotation_type_comment ( n , last_n , offset = 2 )
    last_n = n


signature:
def get_function_signature ( fun , * , is_method = False ) :
    """Returns ( args , returns ) .
 ` args ` is ast3 . arguments , ` returns ` is the return type AST node . The kicker
 about this function is that it pushes type comments into proper annotation
 fields , standardizing type handling ."""

implementation:
args = fun . args
returns = fun . returns
if fun . type_comment :
    try :
        args_tc , returns_tc = parse_signature_type_comment ( fun . type_comment )
        if returns and returns_tc :
            raise ValueError ( "using both a type annotation and a type comment is not allowed" )
        returns = returns_tc
        copy_arguments_to_annotations ( args , args_tc , is_method = is_method )
    except ( SyntaxError , ValueError ) as exc :
        raise ValueError ( f"Annotation problem in function {fun.name!r}: " + f"{fun.lineno}:{fun.col_offset + 1}: {exc}" )
copy_type_comments_to_annotations ( args )
return args , returns


signature:
def parse_signature_type_comment ( type_comment ) :
    """Parse the fugly signature type comment into AST nodes .
 Caveats : ASTifying * * kwargs is impossible with the current grammar so we
 hack it into unary subtraction ( to differentiate from Starred in vararg ) .
 For example from :
 " ( str , int , * int , * * Any ) - > ' SomeReturnType ' "
 To :
 ( [ ast3 . Name , ast . Name , ast3 . Name , ast . Name ] , ast3 . Str )"""

implementation:
try :
    result = ast3 . parse ( type_comment , '<func_type>' , 'func_type' )
except SyntaxError :
    raise ValueError ( f"invalid function signature type comment: {type_comment!r}" )
assert isinstance ( result , ast3 . FunctionType )
if len ( result . argtypes ) == 1 :
    argtypes = result . argtypes [ 0 ]
else :
    argtypes = result . argtypes
return argtypes , result . returns


signature:
def parse_type_comment ( type_comment ) :
    """Parse a type comment string into AST nodes ."""

implementation:
try :
    result = ast3 . parse ( type_comment , '<type_comment>' , 'eval' )
except SyntaxError :
    raise ValueError ( f"invalid type comment: {type_comment!r}" ) from None
assert isinstance ( result , ast3 . Expression )
return result . body


signature:
def parse_arguments ( arguments ) :
    """parse _ arguments ( ' ( a , b , * , c = False , * * d ) ' ) - > ast3 . arguments
 Parse a string with function arguments into an AST node ."""

implementation:
arguments = f"def f{arguments}: ..."
try :
    result = ast3 . parse ( arguments , '<arguments>' , 'exec' )
except SyntaxError :
    raise ValueError ( f"invalid arguments: {arguments!r}" ) from None
assert isinstance ( result , ast3 . Module )
assert len ( result . body ) == 1
assert isinstance ( result . body [ 0 ] , ast3 . FunctionDef )
args = result . body [ 0 ] . args
copy_type_comments_to_annotations ( args )
return args


signature:
def copy_arguments_to_annotations ( args , type_comment , * , is_method = False ) :
    """Copies AST nodes from ` type _ comment ` into the ast3 . arguments in ` args ` .
 Does validaation of argument count ( allowing for untyped self / cls )
 and type ( vararg and kwarg ) ."""

implementation:
if isinstance ( type_comment , ast3 . Ellipsis ) :
    return
expected = len ( args . args )
if args . vararg :
    expected += 1
expected += len ( args . kwonlyargs )
if args . kwarg :
    expected += 1
actual = len ( type_comment ) if isinstance ( type_comment , list ) else 1
if expected != actual :
    if is_method and expected - actual == 1 :
        pass
        # fine , we ' re just skipping ` self ` , ` cls ` , etc .
    else :
        raise ValueError ( f"number of arguments in type comment doesn't match; " + f"expected {expected}, found {actual}" )
if isinstance ( type_comment , list ) :
    next_value = type_comment . pop
else : # If there ' s just one value , only one of the loops and ifs below will
# be populated . We ensure this with the expected / actual length check
# above .
    _tc = type_comment
    def next_value ( index : int = 0 ) -> ast3 . expr :
        return _tc
for arg in args . args [ expected - actual : ] :
    ensure_no_annotation ( arg . annotation )
    arg . annotation = next_value ( 0 )
if args . vararg :
    ensure_no_annotation ( args . vararg . annotation )
    args . vararg . annotation = next_value ( 0 )
for arg in args . kwonlyargs :
    ensure_no_annotation ( arg . annotation )
    arg . annotation = next_value ( 0 )
if args . kwarg :
    ensure_no_annotation ( args . kwarg . annotation )
    args . kwarg . annotation = next_value ( 0 )


signature:
def copy_type_comments_to_annotations ( args ) :
    """Copies argument type comments from the legacy long form to annotations
 in the entire function signature ."""

implementation:
for arg in args . args :
    copy_type_comment_to_annotation ( arg )
if args . vararg :
    copy_type_comment_to_annotation ( args . vararg )
for arg in args . kwonlyargs :
    copy_type_comment_to_annotation ( arg )
if args . kwarg :
    copy_type_comment_to_annotation ( args . kwarg )


signature:
def maybe_replace_any_if_equal ( name , expected , actual ) :
    """Return the type given in ` expected ` .
 Raise ValueError if ` expected ` isn ' t equal to ` actual ` . If - - replace - any is
 used , the Any type in ` actual ` is considered equal .
 The implementation is naively checking if the string representation of
 ` actual ` is one of " Any " , " typing . Any " , or " t . Any " . This is done for two
 reasons :
 1 . I ' m lazy .
 2 . We want people to be able to explicitly state that they want Any without it
 being replaced . This way they can use an alias ."""

implementation:
is_equal = expected == actual
if not is_equal and Config . replace_any :
    actual_str = minimize_whitespace ( str ( actual ) )
    if actual_str and actual_str [ 0 ] in { '"' , "'" } :
        actual_str = actual_str [ 1 : - 1 ]
    is_equal = actual_str in { 'Any' , 'typing.Any' , 't.Any' }
if not is_equal :
    expected_annotation = minimize_whitespace ( str ( expected ) )
    actual_annotation = minimize_whitespace ( str ( actual ) )
    raise ValueError ( f"incompatible existing {name}. " + f"Expected: {expected_annotation!r}, actual: {actual_annotation!r}" )
return expected or actual


signature:
def remove_function_signature_type_comment ( body ) :
    """Removes the legacy signature type comment , leaving other comments if any ."""

implementation:
for node in body . children :
    if node . type == token . INDENT :
        prefix = node . prefix . lstrip ( )
        if prefix . startswith ( '# type: ' ) :
            node . prefix = '\n' . join ( prefix . split ( '\n' ) [ 1 : ] )
        break


signature:
def flatten_some ( children ) :
    """Generates nodes or leaves , unpacking bodies of try : except : finally : statements ."""

implementation:
for node in children :
    if node . type in ( syms . try_stmt , syms . suite ) :
        yield from flatten_some ( node . children )
    else :
        yield node


signature:
def pop_param ( params ) :
    """Pops the parameter and the " remainder " ( comma , default value ) .
 Returns a tuple of ( ' name ' , default ) or ( _ star , ' name ' ) or ( _ dstar , ' name ' ) ."""

implementation:
default = None
name = params . pop ( 0 )
if name in ( _star , _dstar ) :
    default = params . pop ( 0 )
    if default == _comma :
        return name , default
try :
    remainder = params . pop ( 0 )
    if remainder == _eq :
        default = params . pop ( 0 )
        remainder = params . pop ( 0 )
    if remainder != _comma :
        raise ValueError ( f"unexpected token: {remainder}" )
except IndexError :
    pass
return name , default


signature:
def get_offset_and_prefix ( body , skip_assignments = False ) :
    """Returns the offset after which a statement can be inserted to the ` body ` .
 This offset is calculated to come after all imports , and maybe existing
 ( possibly annotated ) assignments if ` skip _ assignments ` is True .
 Also returns the indentation prefix that should be applied to the inserted
 node ."""

implementation:
assert body . type in ( syms . file_input , syms . suite )
_offset = 0
prefix = ''
for _offset , child in enumerate ( body . children ) :
    if child . type == syms . simple_stmt :
        stmt = child . children [ 0 ]
        if stmt . type == syms . expr_stmt :
            expr = stmt . children
            if not skip_assignments :
                break
            if ( len ( expr ) != 2 or expr [ 0 ] . type != token . NAME or expr [ 1 ] . type != syms . annassign or _eq in expr [ 1 ] . children ) :
                break
        elif stmt . type not in ( syms . import_name , syms . import_from , token . STRING ) :
            break
    elif child . type == token . INDENT :
        assert isinstance ( child , Leaf )
        prefix = child . value
    elif child . type != token . NEWLINE :
        break
prefix , child . prefix = child . prefix , prefix
return _offset , prefix


signature:
def fix_line_numbers ( body ) :
    r"""Recomputes all line numbers based on the number of \ n characters ."""

implementation:
maxline = 0
for node in body . pre_order ( ) :
    maxline += node . prefix . count ( '\n' )
    if isinstance ( node , Leaf ) :
        node . lineno = maxline
        maxline += str ( node . value ) . count ( '\n' )


signature:
def new ( n , prefix = None ) :
    """lib2to3 ' s AST requires unique objects as children ."""

implementation:
if isinstance ( n , Leaf ) :
    return Leaf ( n . type , n . value , prefix = n . prefix if prefix is None else prefix )
# this is hacky , we assume complex nodes are just being reused once from the
# original AST .
n . parent = None
if prefix is not None :
    n . prefix = prefix
return n


signature:
def apply_job_security ( code ) :
    """Treat input ` code ` like Python 2 ( implicit strings are byte literals ) .
 The implementation is horribly inefficient but the goal is to be compatible
 with what Mercurial does at runtime ."""

implementation:
buf = io . BytesIO ( code . encode ( 'utf8' ) )
tokens = tokenize . tokenize ( buf . readline )
# NOTE : by setting the fullname to ` mercurial . pycompat ` below , we ' re
# ensuring that hg - specific pycompat imports aren ' t inserted to the code .
data = tokenize . untokenize ( replacetokens ( list ( tokens ) , 'mercurial.pycompat' ) )
return cast ( str , data . decode ( 'utf8' ) )


signature:
def _load_info ( self ) :
    '''Get user info for GBDX S3 , put into instance vars for convenience .
 Args :
 None .
 Returns :
 Dictionary with S3 access key , S3 secret key , S3 session token ,
 user bucket and user prefix ( dict ) .'''

implementation:
url = '%s/prefix?duration=36000' % self . base_url
r = self . gbdx_connection . get ( url )
r . raise_for_status ( )
return r . json ( )


signature:
def download ( self , location , local_dir = '.' ) :
    '''Download content from bucket / prefix / location .
 Location can be a directory or a file ( e . g . , my _ dir or my _ dir / my _ image . tif )
 If location is a directory , all files in the directory are
 downloaded . If it is a file , then that file is downloaded .
 Args :
 location ( str ) : S3 location within prefix .
 local _ dir ( str ) : Local directory where file ( s ) will be stored .
 Default is here .'''

implementation:
self . logger . debug ( 'Getting S3 info' )
bucket = self . info [ 'bucket' ]
prefix = self . info [ 'prefix' ]
self . logger . debug ( 'Connecting to S3' )
s3conn = self . client
# remove head and / or trail backslash from location
location = location . strip ( '/' )
self . logger . debug ( 'Downloading contents' )
objects = s3conn . list_objects ( Bucket = bucket , Prefix = ( prefix + '/' + location ) )
if 'Contents' not in objects :
    raise ValueError ( 'Download target {}/{}/{} was not found or inaccessible.' . format ( bucket , prefix , location ) )
for s3key in objects [ 'Contents' ] :
    key = s3key [ 'Key' ]
    # skip directory keys
    if not key or key . endswith ( '/' ) :
        continue
    # get path to each file
    filepath = key . replace ( prefix + '/' + location , '' , 1 ) . lstrip ( '/' )
    filename = key . split ( '/' ) [ - 1 ]
    # self . logger . debug ( filename )
    file_dir = filepath . split ( '/' ) [ : - 1 ]
    file_dir = '/' . join ( file_dir )
    full_dir = os . path . join ( local_dir , file_dir )
    # make sure directory exists
    if not os . path . isdir ( full_dir ) :
        os . makedirs ( full_dir )
    # download file
    s3conn . download_file ( bucket , key , os . path . join ( full_dir , filename ) )
self . logger . debug ( 'Done!' )


signature:
def delete ( self , location ) :
    '''Delete content in bucket / prefix / location .
 Location can be a directory or a file ( e . g . , my _ dir or my _ dir / my _ image . tif )
 If location is a directory , all files in the directory are deleted .
 If it is a file , then that file is deleted .
 Args :
 location ( str ) : S3 location within prefix . Can be a directory or
 a file ( e . g . , my _ dir or my _ dir / my _ image . tif ) .'''

implementation:
bucket = self . info [ 'bucket' ]
prefix = self . info [ 'prefix' ]
self . logger . debug ( 'Connecting to S3' )
s3conn = self . client
# remove head and / or trail backslash from location
if location [ 0 ] == '/' :
    location = location [ 1 : ]
if location [ - 1 ] == '/' :
    location = location [ : - 2 ]
self . logger . debug ( 'Deleting contents' )
for s3key in s3conn . list_objects ( Bucket = bucket , Prefix = ( prefix + '/' + location ) ) [ 'Contents' ] :
    s3conn . delete_object ( Bucket = bucket , Key = s3key [ 'Key' ] )
self . logger . debug ( 'Done!' )


signature:
def upload ( self , local_file , s3_path = None ) :
    '''Upload files to your DG S3 bucket / prefix .
 Args :
 local _ file ( str ) : a path to a local file to upload , directory structures are not mirrored
 s3 _ path : a key ( location ) on s3 to upload the file to
 Returns :
 str : s3 path file was saved to
 Examples :
 > > > upload ( ' path / to / image . tif ' )
 ' mybucket / myprefix / image . tif '
 > > > upload ( ' . / images / image . tif ' )
 ' mybucket / myprefix / image . tif '
 > > > upload ( ' . / images / image . tif ' , s3 _ path = ' images / image . tif ' )
 ' mybucket / myprefix / images / image . tif ' '''

implementation:
if not os . path . exists ( local_file ) :
    raise Exception ( local_file + " does not exist." )
if s3_path is None :
    s3_path = os . path . basename ( local_file )
bucket = self . info [ 'bucket' ]
prefix = self . info [ 'prefix' ]
self . logger . debug ( 'Connecting to S3' )
s3conn = self . client
self . logger . debug ( 'Uploading file {}' . format ( local_file ) )
s3conn . upload_file ( local_file , bucket , prefix + '/' + s3_path )
self . logger . debug ( 'Done!' )
return '{}/{}/{}' . format ( bucket , prefix , s3_path )


signature:
def rgb ( self , ** kwargs ) :
    '''Convert the image to a 3 band RGB for plotting
 This method shares the same arguments as plot ( ) . It will perform visual adjustment on the
 image and prepare the data for plotting in MatplotLib . Values are converted to an
 appropriate precision and the axis order is changed to put the band axis last .'''

implementation:
if "bands" in kwargs :
    use_bands = kwargs [ "bands" ]
    assert len ( use_bands ) == 3 , 'Plot method only supports single or 3-band outputs'
    del kwargs [ "bands" ]
else :
    use_bands = self . _rgb_bands
if kwargs . get ( 'blm' ) == True :
    return self . histogram_match ( use_bands , ** kwargs )
# if not specified or DRA ' ed , default to a 2-98 stretch
if "histogram" not in kwargs :
    if "stretch" not in kwargs :
        if not self . options . get ( 'dra' ) :
            kwargs [ 'stretch' ] = [ 2 , 98 ]
    return self . histogram_stretch ( use_bands , ** kwargs )
elif kwargs [ "histogram" ] == "equalize" :
    return self . histogram_equalize ( use_bands , ** kwargs )
elif kwargs [ "histogram" ] == "match" :
    return self . histogram_match ( use_bands , ** kwargs )
elif kwargs [ "histogram" ] == "minmax" :
    return self . histogram_stretch ( use_bands , stretch = [ 0 , 100 ] , ** kwargs )
# DRA ' ed images should be left alone if not explicitly adjusted
elif kwargs [ "histogram" ] == "ignore" or self . options . get ( 'dra' ) :
    data = self . _read ( self [ use_bands , ... ] , ** kwargs )
    return np . rollaxis ( data , 0 , 3 )
else :
    raise KeyError ( 'Unknown histogram parameter, use "equalize", "match", "minmax", or "ignore"' )


signature:
def histogram_equalize ( self , use_bands , ** kwargs ) :
    '''Equalize and the histogram and normalize value range
 Equalization is on all three bands , not per - band'''

implementation:
data = self . _read ( self [ use_bands , ... ] , ** kwargs )
data = np . rollaxis ( data . astype ( np . float32 ) , 0 , 3 )
flattened = data . flatten ( )
if 0 in data :
    masked = np . ma . masked_values ( data , 0 ) . compressed ( )
    image_histogram , bin_edges = np . histogram ( masked , 256 )
else :
    image_histogram , bin_edges = np . histogram ( flattened , 256 )
bins = ( bin_edges [ : - 1 ] + bin_edges [ 1 : ] ) / 2.0
cdf = image_histogram . cumsum ( )
cdf = cdf / float ( cdf [ - 1 ] )
image_equalized = np . interp ( flattened , bins , cdf ) . reshape ( data . shape )
if 'stretch' in kwargs or 'gamma' in kwargs :
    return self . _histogram_stretch ( image_equalized , ** kwargs )
else :
    return image_equalized


signature:
def histogram_match ( self , use_bands , blm_source = None , ** kwargs ) :
    '''Match the histogram to existing imagery'''

implementation:
assert has_rio , "To match image histograms please install rio_hist"
data = self . _read ( self [ use_bands , ... ] , ** kwargs )
data = np . rollaxis ( data . astype ( np . float32 ) , 0 , 3 )
if 0 in data :
    data = np . ma . masked_values ( data , 0 )
bounds = self . _reproject ( box ( * self . bounds ) , from_proj = self . proj , to_proj = "EPSG:4326" ) . bounds
if blm_source == 'browse' :
    from gbdxtools . images . browse_image import BrowseImage
    ref = BrowseImage ( self . cat_id , bbox = bounds ) . read ( )
else :
    from gbdxtools . images . tms_image import TmsImage
    tms = TmsImage ( zoom = self . _calc_tms_zoom ( self . affine [ 0 ] ) , bbox = bounds , ** kwargs )
    ref = np . rollaxis ( tms . read ( ) , 0 , 3 )
out = np . dstack ( [ rio_match ( data [ : , : , idx ] , ref [ : , : , idx ] . astype ( np . double ) / 255.0 ) for idx in range ( data . shape [ - 1 ] ) ] )
if 'stretch' in kwargs or 'gamma' in kwargs :
    return self . _histogram_stretch ( out , ** kwargs )
else :
    return out


signature:
def histogram_stretch ( self , use_bands , ** kwargs ) :
    '''entry point for contrast stretching'''

implementation:
data = self . _read ( self [ use_bands , ... ] , ** kwargs )
data = np . rollaxis ( data . astype ( np . float32 ) , 0 , 3 )
return self . _histogram_stretch ( data , ** kwargs )


signature:
def _histogram_stretch ( self , data , ** kwargs ) :
    '''perform a contrast stretch and / or gamma adjustment'''

implementation:
limits = { }
# get the image min - max statistics
for x in range ( 3 ) :
    band = data [ : , : , x ]
    try :
        limits [ x ] = np . percentile ( band , kwargs . get ( "stretch" , [ 0 , 100 ] ) )
    except IndexError : # this band has no dynamic range and cannot be stretched
        return data
# compute the stretch
for x in range ( 3 ) :
    band = data [ : , : , x ]
    if 0 in band :
        band = np . ma . masked_values ( band , 0 ) . compressed ( )
    top = limits [ x ] [ 1 ]
    bottom = limits [ x ] [ 0 ]
    if top != bottom : # catch divide by zero
        data [ : , : , x ] = ( data [ : , : , x ] - bottom ) / float ( top - bottom ) * 255.0
data = np . clip ( data , 0 , 255 ) . astype ( "uint8" )
# gamma adjust
if "gamma" in kwargs :
    invGamma = 1.0 / kwargs [ 'gamma' ]
    lut = np . array ( [ ( ( i / 255.0 ) ** invGamma ) * 255 for i in np . arange ( 0 , 256 ) ] ) . astype ( "uint8" )
    data = np . take ( lut , data )
return data


signature:
def ndvi ( self , ** kwargs ) :
    """Calculates Normalized Difference Vegetation Index using NIR and Red of an image .
 Returns : numpy array with ndvi values"""

implementation:
data = self . _read ( self [ self . _ndvi_bands , ... ] ) . astype ( np . float32 )
return ( data [ 0 , : , : ] - data [ 1 , : , : ] ) / ( data [ 0 , : , : ] + data [ 1 , : , : ] )


signature:
def ndwi ( self ) :
    """Calculates Normalized Difference Water Index using Coastal and NIR2 bands for WV02 , WV03.
 For Landsat8 and sentinel2 calculated by using Green and NIR bands .
 Returns : numpy array of ndwi values"""

implementation:
data = self . _read ( self [ self . _ndwi_bands , ... ] ) . astype ( np . float32 )
return ( data [ 1 , : , : ] - data [ 0 , : , : ] ) / ( data [ 0 , : , : ] + data [ 1 , : , : ] )


signature:
def plot ( self , spec = "rgb" , ** kwargs ) :
    '''Plot the image with MatplotLib
 Plot sizing includes default borders and spacing . If the image is shown in Jupyter the outside whitespace will be automatically cropped to save size , resulting in a smaller sized image than expected .
 Histogram options :
 * ' equalize ' : performs histogram equalization on the image .
 * ' minmax ' : stretch the pixel range to the minimum and maximum input pixel values . Equivalent to stretch = [ 0,100 ] .
 * ' match ' : match the histogram to the Maps API imagery . Pass the additional keyword blm _ source = ' browse ' to match to the Browse Service ( image thumbnail ) instead .
 * ' ignore ' : Skip dynamic range adjustment , in the event the image is already correctly balanced and the values are in the correct range .
 Gamma values greater than 1 will brighten the image midtones , values less than 1 will darken the midtones .
 Plots generated with the histogram options of ' match ' and ' equalize ' can be combined with the stretch and gamma options . The stretch and gamma adjustments will be applied after the histogram adjustments .
 Args :
 w ( float or int ) : width of plot in inches at 72 dpi , default is 10
 h ( float or int ) : height of plot in inches at 72 dpi , default is 10
 title ( str ) : Title to use on the plot
 fontsize ( int ) : Size of title font , default is 22 . Size is measured in points .
 bands ( list ) : bands to use for plotting , such as bands = [ 4,2,1 ] . Defaults to the image ' s natural RGB bands . This option is useful for generating pseudocolor images when passed a list of three bands . If only a single band is provided , a colormapped plot will be generated instead .
 cmap ( str ) : MatPlotLib colormap name to use for single band images . Default is colormap = ' Grey _ R ' .
 histogram ( str ) : either ' equalize ' , ' minmax ' , ' match ' , or ignore
 stretch ( list ) : stretch the histogram between two percentile values , default is [ 2,98]
 gamma ( float ) : adjust image gamma , default is 1.0'''

implementation:
if self . shape [ 0 ] == 1 or ( "bands" in kwargs and len ( kwargs [ "bands" ] ) == 1 ) :
    if "cmap" in kwargs :
        cmap = kwargs [ "cmap" ]
        del kwargs [ "cmap" ]
    else :
        cmap = "Greys_r"
    self . _plot ( tfm = self . _single_band , cmap = cmap , ** kwargs )
else :
    if spec == "rgb" and self . _has_token ( ** kwargs ) :
        self . _plot ( tfm = self . rgb , ** kwargs )
    else :
        self . _plot ( tfm = getattr ( self , spec ) , ** kwargs )


signature:
def get_images_by_catid_and_aoi ( self , catid , aoi_wkt ) :
    """Retrieves the IDAHO image records associated with a given catid .
 Args :
 catid ( str ) : The source catalog ID from the platform catalog .
 aoi _ wkt ( str ) : The well known text of the area of interest .
 Returns :
 results ( json ) : The full catalog - search response for IDAHO images
 within the catID ."""

implementation:
self . logger . debug ( 'Retrieving IDAHO metadata' )
# use the footprint to get the IDAHO id
url = '%s/search' % self . base_url
body = { "filters" : [ "catalogID = '%s'" % catid ] , "types" : [ "IDAHOImage" ] , "searchAreaWkt" : aoi_wkt }
r = self . gbdx_connection . post ( url , data = json . dumps ( body ) )
r . raise_for_status ( )
if r . status_code == 200 :
    results = r . json ( )
    numresults = len ( results [ 'results' ] )
    self . logger . debug ( '%s IDAHO images found associated with catid %s' % ( numresults , catid ) )
    return results


signature:
def get_images_by_catid ( self , catid ) :
    """Retrieves the IDAHO image records associated with a given catid .
 Args :
 catid ( str ) : The source catalog ID from the platform catalog .
 Returns :
 results ( json ) : The full catalog - search response for IDAHO images
 within the catID ."""

implementation:
self . logger . debug ( 'Retrieving IDAHO metadata' )
# get the footprint of the catid ' s strip
footprint = self . catalog . get_strip_footprint_wkt ( catid )
# try to convert from multipolygon to polygon :
try :
    footprint = from_wkt ( footprint ) . geoms [ 0 ] . wkt
except :
    pass
if not footprint :
    self . logger . debug ( """Cannot get IDAHO metadata for strip %s,
                                 footprint not found""" % catid )
    return None
return self . get_images_by_catid_and_aoi ( catid = catid , aoi_wkt = footprint )


signature:
def describe_images ( self , idaho_image_results ) :
    """Describe the result set of a catalog search for IDAHO images .
 Args :
 idaho _ image _ results ( dict ) : Result set of catalog search .
 Returns :
 results ( json ) : The full catalog - search response for IDAHO images
 corresponding to the given catID ."""

implementation:
results = idaho_image_results [ 'results' ]
# filter only idaho images :
results = [ r for r in results if 'IDAHOImage' in r [ 'type' ] ]
self . logger . debug ( 'Describing %s IDAHO images.' % len ( results ) )
# figure out which catids are represented in this set of images
catids = set ( [ r [ 'properties' ] [ 'catalogID' ] for r in results ] )
description = { }
for catid in catids : # images associated with a single catid
    description [ catid ] = { }
    description [ catid ] [ 'parts' ] = { }
    images = [ r for r in results if r [ 'properties' ] [ 'catalogID' ] == catid ]
    for image in images :
        description [ catid ] [ 'sensorPlatformName' ] = image [ 'properties' ] [ 'sensorPlatformName' ]
        part = int ( image [ 'properties' ] [ 'vendorDatasetIdentifier' ] . split ( ':' ) [ 1 ] [ - 3 : ] )
        color = image [ 'properties' ] [ 'colorInterpretation' ]
        bucket = image [ 'properties' ] [ 'tileBucketName' ]
        identifier = image [ 'identifier' ]
        boundstr = image [ 'properties' ] [ 'footprintWkt' ]
        try :
            description [ catid ] [ 'parts' ] [ part ]
        except :
            description [ catid ] [ 'parts' ] [ part ] = { }
        description [ catid ] [ 'parts' ] [ part ] [ color ] = { }
        description [ catid ] [ 'parts' ] [ part ] [ color ] [ 'id' ] = identifier
        description [ catid ] [ 'parts' ] [ part ] [ color ] [ 'bucket' ] = bucket
        description [ catid ] [ 'parts' ] [ part ] [ color ] [ 'boundstr' ] = boundstr
return description


signature:
def get_chip ( self , coordinates , catid , chip_type = 'PAN' , chip_format = 'TIF' , filename = 'chip.tif' ) :
    """Downloads a native resolution , orthorectified chip in tif format
 from a user - specified catalog id .
 Args :
 coordinates ( list ) : Rectangle coordinates in order West , South , East , North .
 West and East are longitudes , North and South are latitudes .
 The maximum chip size is ( 2048 pix ) x ( 2048 pix )
 catid ( str ) : The image catalog id .
 chip _ type ( str ) : ' PAN ' ( panchromatic ) , ' MS ' ( multispectral ) , ' PS ' ( pansharpened ) .
 ' MS ' is 4 or 8 bands depending on sensor .
 chip _ format ( str ) : ' TIF ' or ' PNG '
 filename ( str ) : Where to save chip .
 Returns :
 True if chip is successfully downloaded ; else False ."""

implementation:
def t2s1 ( t ) : # Tuple to string 1
    return str ( t ) . strip ( '(,)' ) . replace ( ',' , '' )
def t2s2 ( t ) : # Tuple to string 2
    return str ( t ) . strip ( '(,)' ) . replace ( ' ' , '' )
if len ( coordinates ) != 4 :
    print ( 'Wrong coordinate entry' )
    return False
W , S , E , N = coordinates
box = ( ( W , S ) , ( W , N ) , ( E , N ) , ( E , S ) , ( W , S ) )
box_wkt = 'POLYGON ((' + ',' . join ( [ t2s1 ( corner ) for corner in box ] ) + '))'
# get IDAHO images which intersect box
results = self . get_images_by_catid_and_aoi ( catid = catid , aoi_wkt = box_wkt )
description = self . describe_images ( results )
pan_id , ms_id , num_bands = None , None , 0
for catid , images in description . items ( ) :
    for partnum , part in images [ 'parts' ] . items ( ) :
        if 'PAN' in part . keys ( ) :
            pan_id = part [ 'PAN' ] [ 'id' ]
            bucket = part [ 'PAN' ] [ 'bucket' ]
        if 'WORLDVIEW_8_BAND' in part . keys ( ) :
            ms_id = part [ 'WORLDVIEW_8_BAND' ] [ 'id' ]
            num_bands = 8
            bucket = part [ 'WORLDVIEW_8_BAND' ] [ 'bucket' ]
        elif 'RGBN' in part . keys ( ) :
            ms_id = part [ 'RGBN' ] [ 'id' ]
            num_bands = 4
            bucket = part [ 'RGBN' ] [ 'bucket' ]
# specify band information
band_str = ''
if chip_type == 'PAN' :
    band_str = pan_id + '?bands=0'
elif chip_type == 'MS' :
    band_str = ms_id + '?'
elif chip_type == 'PS' :
    if num_bands == 8 :
        band_str = ms_id + '?bands=4,2,1&panId=' + pan_id
    elif num_bands == 4 :
        band_str = ms_id + '?bands=0,1,2&panId=' + pan_id
# specify location information
location_str = '&upperLeft={}&lowerRight={}' . format ( t2s2 ( ( W , N ) ) , t2s2 ( ( E , S ) ) )
service_url = 'https://idaho.geobigdata.io/v1/chip/bbox/' + bucket + '/'
url = service_url + band_str + location_str
url += '&format=' + chip_format + '&token=' + self . gbdx_connection . access_token
r = requests . get ( url )
if r . status_code == 200 :
    with open ( filename , 'wb' ) as f :
        f . write ( r . content )
        return True
else :
    print ( 'Cannot download chip' )
    return False


signature:
def get_tms_layers ( self , catid , bands = '4,2,1' , gamma = 1.3 , highcutoff = 0.98 , lowcutoff = 0.02 , brightness = 1.0 , contrast = 1.0 ) :
    """Get list of urls and bounding boxes corrsponding to idaho images for a given catalog id .
 Args :
 catid ( str ) : Catalog id
 bands ( str ) : Bands to display , separated by commas ( 0-7 ) .
 gamma ( float ) : gamma coefficient . This is for on - the - fly pansharpening .
 highcutoff ( float ) : High cut off coefficient ( 0.0 to 1.0 ) . This is for on - the - fly pansharpening .
 lowcutoff ( float ) : Low cut off coefficient ( 0.0 to 1.0 ) . This is for on - the - fly pansharpening .
 brightness ( float ) : Brightness coefficient ( 0.0 to 1.0 ) . This is for on - the - fly pansharpening .
 contrast ( float ) : Contrast coefficient ( 0.0 to 1.0 ) . This is for on - the - fly pansharpening .
 Returns :
 urls ( list ) : TMS urls .
 bboxes ( list of tuples ) : Each tuple is ( W , S , E , N ) where ( W , S , E , N ) are the bounds of the corresponding idaho part ."""

implementation:
description = self . describe_images ( self . get_images_by_catid ( catid ) )
service_url = 'http://idaho.geobigdata.io/v1/tile/'
urls , bboxes = [ ] , [ ]
for catid , images in description . items ( ) :
    for partnum , part in images [ 'parts' ] . items ( ) :
        if 'PAN' in part . keys ( ) :
            pan_id = part [ 'PAN' ] [ 'id' ]
        if 'WORLDVIEW_8_BAND' in part . keys ( ) :
            ms_id = part [ 'WORLDVIEW_8_BAND' ] [ 'id' ]
            ms_partname = 'WORLDVIEW_8_BAND'
        elif 'RGBN' in part . keys ( ) :
            ms_id = part [ 'RGBN' ] [ 'id' ]
            ms_partname = 'RGBN'
        if ms_id :
            if pan_id :
                band_str = ms_id + '/{z}/{x}/{y}?bands=' + bands + '&panId=' + pan_id
            else :
                band_str = ms_id + '/{z}/{x}/{y}?bands=' + bands
            bbox = from_wkt ( part [ ms_partname ] [ 'boundstr' ] ) . bounds
        elif not ms_id and pan_id :
            band_str = pan_id + '/{z}/{x}/{y}?bands=0'
            bbox = from_wkt ( part [ 'PAN' ] [ 'boundstr' ] ) . bounds
        else :
            continue
        bboxes . append ( bbox )
        # Get the bucket . It has to be the same for all entries in the part .
        bucket = part [ list ( part . keys ( ) ) [ 0 ] ] [ 'bucket' ]
        # Get the token
        token = self . gbdx_connection . access_token
        # Assemble url
        url = ( service_url + bucket + '/' + band_str + """&gamma={}
                                        &highCutoff={}
                                        &lowCutoff={}
                                        &brightness={}
                                        &contrast={}
                                        &token={}""" . format ( gamma , highcutoff , lowcutoff , brightness , contrast , token ) )
        urls . append ( url )
return urls , bboxes


signature:
def create_leaflet_viewer ( self , idaho_image_results , filename ) :
    """Create a leaflet viewer html file for viewing idaho images .
 Args :
 idaho _ image _ results ( dict ) : IDAHO image result set as returned from
 the catalog .
 filename ( str ) : Where to save output html file ."""

implementation:
description = self . describe_images ( idaho_image_results )
if len ( description ) > 0 :
    functionstring = ''
    for catid , images in description . items ( ) :
        for partnum , part in images [ 'parts' ] . items ( ) :
            num_images = len ( list ( part . keys ( ) ) )
            partname = None
            if num_images == 1 : # there is only one image , use the PAN
                partname = [ p for p in list ( part . keys ( ) ) ] [ 0 ]
                pan_image_id = ''
            elif num_images == 2 : # there are two images in this part , use the multi ( or pansharpen )
                partname = [ p for p in list ( part . keys ( ) ) if p is not 'PAN' ] [ 0 ]
                pan_image_id = part [ 'PAN' ] [ 'id' ]
            if not partname :
                self . logger . debug ( "Cannot find part for idaho image." )
                continue
            bandstr = { 'RGBN' : '0,1,2' , 'WORLDVIEW_8_BAND' : '4,2,1' , 'PAN' : '0' } . get ( partname , '0,1,2' )
            part_boundstr_wkt = part [ partname ] [ 'boundstr' ]
            part_polygon = from_wkt ( part_boundstr_wkt )
            bucketname = part [ partname ] [ 'bucket' ]
            image_id = part [ partname ] [ 'id' ]
            W , S , E , N = part_polygon . bounds
            functionstring += "addLayerToMap('%s','%s',%s,%s,%s,%s,'%s');\n" % ( bucketname , image_id , W , S , E , N , pan_image_id )
    __location__ = os . path . realpath ( os . path . join ( os . getcwd ( ) , os . path . dirname ( __file__ ) ) )
    try :
        with open ( os . path . join ( __location__ , 'leafletmap_template.html' ) , 'r' ) as htmlfile :
            data = htmlfile . read ( ) . decode ( "utf8" )
    except AttributeError :
        with open ( os . path . join ( __location__ , 'leafletmap_template.html' ) , 'r' ) as htmlfile :
            data = htmlfile . read ( )
    data = data . replace ( 'FUNCTIONSTRING' , functionstring )
    data = data . replace ( 'CENTERLAT' , str ( S ) )
    data = data . replace ( 'CENTERLON' , str ( W ) )
    data = data . replace ( 'BANDS' , bandstr )
    data = data . replace ( 'TOKEN' , self . gbdx_connection . access_token )
    with codecs . open ( filename , 'w' , 'utf8' ) as outputfile :
        self . logger . debug ( "Saving %s" % filename )
        outputfile . write ( data )
else :
    print ( 'No items returned.' )


signature:
def is_ordered ( cat_id ) :
    """Checks to see if a CatalogID has been ordered or not .
 Args :
 catalogID ( str ) : The catalog ID from the platform catalog .
 Returns :
 ordered ( bool ) : Whether or not the image has been ordered"""

implementation:
url = 'https://rda.geobigdata.io/v1/stripMetadata/{}' . format ( cat_id )
auth = Auth ( )
r = _req_with_retries ( auth . gbdx_connection , url )
if r is not None :
    return r . status_code == 200
return False


signature:
def can_acomp ( cat_id ) :
    """Checks to see if a CatalogID can be atmos . compensated or not .
 Args :
 catalogID ( str ) : The catalog ID from the platform catalog .
 Returns :
 available ( bool ) : Whether or not the image can be acomp ' d"""

implementation:
url = 'https://rda.geobigdata.io/v1/stripMetadata/{}/capabilities' . format ( cat_id )
auth = Auth ( )
r = _req_with_retries ( auth . gbdx_connection , url )
try :
    data = r . json ( )
    return data [ 'acompVersion' ] is not None
except :
    return False


signature:
def deprecate_module_attr ( mod , deprecated ) :
    """Return a wrapped object that warns about deprecated accesses"""

implementation:
deprecated = set ( deprecated )
class Wrapper ( object ) :
    def __getattr__ ( self , attr ) :
        if attr in deprecated :
            warnings . warn ( "Property {} is deprecated" . format ( attr ) , GBDXDeprecation )
        return getattr ( mod , attr )
    def __setattr__ ( self , attr , value ) :
        if attr in deprecated :
            warnings . warn ( "Property {} is deprecated" . format ( attr ) , GBDXDeprecation )
        return setattr ( mod , attr , value )
return Wrapper ( )


signature:
def get_matching_multiplex_port ( self , name ) :
    """Given a name , figure out if a multiplex port prefixes this name and return it . Otherwise return none ."""

implementation:
# short circuit : if the attribute name already exists return none
# if name in self . _ portnames : return None
# if not len ( [ p for p in self . _ portnames if name . startswith ( p ) and name ! = p ] ) : return None
matching_multiplex_ports = [ self . __getattribute__ ( p ) for p in self . _portnames if name . startswith ( p ) and name != p and hasattr ( self , p ) and self . __getattribute__ ( p ) . is_multiplex ]
for port in matching_multiplex_ports :
    return port
return None


signature:
def set ( self , ** kwargs ) :
    """Set input values on task
 Args :
 arbitrary _ keys : values for the keys
 Returns :
 None"""

implementation:
for port_name , port_value in kwargs . items ( ) : # Support both port and port . value
    if hasattr ( port_value , 'value' ) :
        port_value = port_value . value
    self . inputs . __setattr__ ( port_name , port_value )


signature:
def savedata ( self , output , location = None ) :
    '''Save output data from any task in this workflow to S3
 Args :
 output : Reference task output ( e . g . task . outputs . output1 ) .
 location ( optional ) : Subfolder under which the output will be saved .
 It will be placed under the account directory in gbd - customer - data bucket :
 s3 : / / gbd - customer - data / { account _ id } / { location }
 Leave blank to save to : workflow _ output / { workflow _ id } / { task _ name } / { port _ name }
 Returns :
 None'''

implementation:
output . persist = True
if location :
    output . persist_location = location


signature:
def list_workflow_outputs ( self ) :
    '''Get a list of outputs from the workflow that are saved to S3 . To get resolved locations call workflow status .
 Args :
 None
 Returns :
 list'''

implementation:
workflow_outputs = [ ]
for task in self . tasks :
    for output_port_name in task . outputs . _portnames :
        if task . outputs . __getattribute__ ( output_port_name ) . persist :
            workflow_outputs . append ( task . name + ':' + output_port_name )
return workflow_outputs


signature:
def generate_workflow_description ( self ) :
    '''Generate workflow json for launching the workflow against the gbdx api
 Args :
 None
 Returns :
 json string'''

implementation:
if not self . tasks :
    raise WorkflowError ( 'Workflow contains no tasks, and cannot be executed.' )
self . definition = self . workflow_skeleton ( )
if self . batch_values :
    self . definition [ "batch_values" ] = self . batch_values
all_input_port_values = [ t . inputs . __getattribute__ ( input_port_name ) . value for t in self . tasks for input_port_name in t . inputs . _portnames ]
for task in self . tasks : # only include multiplex output ports in this task if other tasks refer to them in their inputs .
# 1 . find the multplex output port _ names in this task
# 2 . see if they are referred to in any other tasks inputs
# 3 . If not , exclude them from the workflow _ def
    output_multiplex_ports_to_exclude = [ ]
    multiplex_output_port_names = [ portname for portname in task . outputs . _portnames if task . outputs . __getattribute__ ( portname ) . is_multiplex ]
    for p in multiplex_output_port_names :
        output_port_reference = 'source:' + task . name + ':' + p
        if output_port_reference not in all_input_port_values :
            output_multiplex_ports_to_exclude . append ( p )
    task_def = task . generate_task_workflow_json ( output_multiplex_ports_to_exclude = output_multiplex_ports_to_exclude )
    self . definition [ 'tasks' ] . append ( task_def )
if self . callback :
    self . definition [ 'callback' ] = self . callback
return self . definition


signature:
def execute ( self ) :
    '''Execute the workflow .
 Args :
 None
 Returns :
 Workflow _ id'''

implementation:
# if not self . tasks :
# raise WorkflowError ( ' Workflow contains no tasks , and cannot be executed . ' )
# for task in self . tasks :
# self . definition [ ' tasks ' ] . append ( task . generate _ task _ workflow _ json ( ) )
self . generate_workflow_description ( )
# hit batch workflow endpoint if batch values
if self . batch_values :
    self . id = self . workflow . launch_batch_workflow ( self . definition )
# use regular workflow endpoint if no batch values
else :
    self . id = self . workflow . launch ( self . definition )
return self . id


signature:
def task_ids ( self ) :
    '''Get the task IDs of a running workflow
 Args :
 None
 Returns :
 List of task IDs'''

implementation:
if not self . id :
    raise WorkflowError ( 'Workflow is not running.  Cannot get task IDs.' )
if self . batch_values :
    raise NotImplementedError ( "Query Each Workflow Id within the Batch Workflow for task IDs." )
wf = self . workflow . get ( self . id )
return [ task [ 'id' ] for task in wf [ 'tasks' ] ]


signature:
def cancel ( self ) :
    '''Cancel a running workflow .
 Args :
 None
 Returns :
 None'''

implementation:
if not self . id :
    raise WorkflowError ( 'Workflow is not running.  Cannot cancel.' )
if self . batch_values :
    self . workflow . batch_workflow_cancel ( self . id )
else :
    self . workflow . cancel ( self . id )


signature:
def stdout ( self ) :
    '''Get stdout from all the tasks of a workflow .
 Returns :
 ( list ) : tasks with their stdout
 Example :
 > > > workflow . stdout
 " id " : " 4488895771403082552 " ,
 " taskType " : " AOP _ Strip _ Processor " ,
 " name " : " Task1 " ,
 " stdout " : " . . . . . "'''

implementation:
if not self . id :
    raise WorkflowError ( 'Workflow is not running.  Cannot get stdout.' )
if self . batch_values :
    raise NotImplementedError ( "Query Each Workflow Id within the Batch Workflow for stdout." )
wf = self . workflow . get ( self . id )
stdout_list = [ ]
for task in wf [ 'tasks' ] :
    stdout_list . append ( { 'id' : task [ 'id' ] , 'taskType' : task [ 'taskType' ] , 'name' : task [ 'name' ] , 'stdout' : self . workflow . get_stdout ( self . id , task [ 'id' ] ) } )
return stdout_list


signature:
def stderr ( self ) :
    '''Get stderr from all the tasks of a workflow .
 Returns :
 ( list ) : tasks with their stderr
 Example :
 > > > workflow . stderr
 " id " : " 4488895771403082552 " ,
 " taskType " : " AOP _ Strip _ Processor " ,
 " name " : " Task1 " ,
 " stderr " : " . . . . . "'''

implementation:
if not self . id :
    raise WorkflowError ( 'Workflow is not running.  Cannot get stderr.' )
if self . batch_values :
    raise NotImplementedError ( "Query Each Workflow Id within the Batch Workflow for stderr." )
wf = self . workflow . get ( self . id )
stderr_list = [ ]
for task in wf [ 'tasks' ] :
    stderr_list . append ( { 'id' : task [ 'id' ] , 'taskType' : task [ 'taskType' ] , 'name' : task [ 'name' ] , 'stderr' : self . workflow . get_stderr ( self . id , task [ 'id' ] ) } )
return stderr_list


signature:
def layers ( self ) :
    """Renders the list of layers to add to the map .
 Returns :
 layers ( list ) : list of layer entries suitable for use in mapbox - gl ' map . addLayer ( ) ' call"""

implementation:
layers = [ self . _layer_def ( style ) for style in self . styles ]
return layers


signature:
def get_proj ( prj_code ) :
    """Helper method for handling projection codes that are unknown to pyproj
 Args :
 prj _ code ( str ) : an epsg proj code
 Returns :
 projection : a pyproj projection"""

implementation:
if prj_code in CUSTOM_PRJ :
    proj = pyproj . Proj ( CUSTOM_PRJ [ prj_code ] )
else :
    proj = pyproj . Proj ( init = prj_code )
return proj


signature:
def preview ( image , ** kwargs ) :
    '''Show a slippy map preview of the image . Requires iPython .
 Args :
 image ( image ) : image object to display
 zoom ( int ) : zoom level to intialize the map , default is 16
 center ( list ) : center coordinates to initialize the map , defaults to center of image
 bands ( list ) : bands of image to display , defaults to the image ' s default RGB bands'''

implementation:
try :
    from IPython . display import Javascript , HTML , display
    from gbdxtools . rda . interface import RDA
    from gbdxtools import Interface
    gbdx = Interface ( )
except :
    print ( "IPython is required to produce maps." )
    return
zoom = kwargs . get ( "zoom" , 16 )
bands = kwargs . get ( "bands" )
if bands is None :
    bands = image . _rgb_bands
wgs84_bounds = kwargs . get ( "bounds" , list ( loads ( image . metadata [ "image" ] [ "imageBoundsWGS84" ] ) . bounds ) )
center = kwargs . get ( "center" , list ( shape ( image ) . centroid . bounds [ 0 : 2 ] ) )
if image . proj != 'EPSG:4326' :
    code = image . proj . split ( ':' ) [ 1 ]
    conn = gbdx . gbdx_connection
    proj_info = conn . get ( 'https://ughlicoordinates.geobigdata.io/ughli/v1/projinfo/{}' . format ( code ) ) . json ( )
    tfm = partial ( pyproj . transform , pyproj . Proj ( init = 'EPSG:4326' ) , pyproj . Proj ( init = image . proj ) )
    bounds = list ( ops . transform ( tfm , box ( * wgs84_bounds ) ) . bounds )
else :
    proj_info = { }
    bounds = wgs84_bounds
# Applying DRA to a DRA ' ed image looks bad , skip if already in graph
if not image . options . get ( 'dra' ) :
    rda = RDA ( )
    # Need some simple DRA to get the image in range for display .
    dra = rda . HistogramDRA ( image )
    image = dra . aoi ( bbox = image . bounds )
graph_id = image . rda_id
node_id = image . rda . graph ( ) [ 'nodes' ] [ 0 ] [ 'id' ]
map_id = "map_{}" . format ( str ( int ( time . time ( ) ) ) )
scales = ',' . join ( [ '1' ] * len ( bands ) )
offsets = ',' . join ( [ '0' ] * len ( bands ) )
display ( HTML ( Template ( '''
       <div id="$map_id"/>
       <link href='https://openlayers.org/en/v4.6.4/css/ol.css' rel='stylesheet' />
       <script src="https://cdn.polyfill.io/v2/polyfill.min.js?features=requestAnimationFrame,Element.prototype.classList,URL"></script>
       <style>body{margin:0;padding:0;}#$map_id{position:relative;top:0;bottom:0;width:100%;height:400px;}</style>
       <style></style>
    ''' ) . substitute ( { "map_id" : map_id } ) ) )
js = Template ( """
        require.config({
            paths: {
                oljs: 'https://cdnjs.cloudflare.com/ajax/libs/openlayers/4.6.4/ol',
                proj4: 'https://cdnjs.cloudflare.com/ajax/libs/proj4js/2.4.4/proj4'
            }
        });

        require(['oljs', 'proj4'], function(oljs, proj4) {
            oljs.proj.setProj4(proj4)
            var md = $md;
            var georef = $georef;
            var graphId = '$graphId';
            var nodeId = '$nodeId';
            var extents = $bounds;

            var x1 = md.minTileX * md.tileXSize;
            var y1 = ((md.minTileY + md.numYTiles) * md.tileYSize + md.tileYSize);
            var x2 = ((md.minTileX + md.numXTiles) * md.tileXSize + md.tileXSize);
            var y2 = md.minTileY * md.tileYSize;
            var tileLayerResolutions = [georef.scaleX];

            var url = '$url' + '/tile/';
            url += graphId + '/' + nodeId;
            url += "/{x}/{y}.png?token=$token&display_bands=$bands&display_scales=$scales&display_offsets=$offsets";

            var proj = '$proj';
            var projInfo = $projInfo;

            if ( proj !== 'EPSG:4326' ) {
                var proj4def = projInfo["proj4"];
                proj4.defs(proj, proj4def);
                var area = projInfo["area_of_use"];
                var bbox = [area["area_west_bound_lon"], area["area_south_bound_lat"],
                            area["area_east_bound_lon"], area["area_north_bound_lat"]]
                var projection = oljs.proj.get(proj);
                var fromLonLat = oljs.proj.getTransform('EPSG:4326', projection);
                var extent = oljs.extent.applyTransform(
                    [bbox[0], bbox[1], bbox[2], bbox[3]], fromLonLat);
                projection.setExtent(extent);
            } else {
                var projection = oljs.proj.get(proj);
            }

            var rda = new oljs.layer.Tile({
              title: 'RDA',
              opacity: 1,
              extent: extents,
              source: new oljs.source.TileImage({
                      crossOrigin: null,
                      projection: projection,
                      extent: extents,

                      tileGrid: new oljs.tilegrid.TileGrid({
                          extent: extents,
                          origin: [extents[0], extents[3]],
                          resolutions: tileLayerResolutions,
                          tileSize: [md.tileXSize, md.tileYSize],
                      }),
                      tileUrlFunction: function (coordinate) {
                          if (coordinate === null) return undefined;
                          const x = coordinate[1] + md.minTileX;
                          const y = -(coordinate[2] + 1 - md.minTileY);
                          if (x < md.minTileX || x > md.maxTileX) return undefined;
                          if (y < md.minTileY || y > md.maxTileY) return undefined;
                          return url.replace('{x}', x).replace('{y}', y);
                      }
                  })
            });

            var map = new oljs.Map({
              layers: [ rda ],
              target: '$map_id',
              view: new oljs.View({
                projection: projection,
                center: $center,
                zoom: $zoom
              })
            });
        });
    """ ) . substitute ( { "map_id" : map_id , "proj" : image . proj , "projInfo" : json . dumps ( proj_info ) , "graphId" : graph_id , "bounds" : bounds , "bands" : "," . join ( map ( str , bands ) ) , "nodeId" : node_id , "md" : json . dumps ( image . metadata [ "image" ] ) , "georef" : json . dumps ( image . metadata [ "georef" ] ) , "center" : center , "zoom" : zoom , "token" : gbdx . gbdx_connection . access_token , "scales" : scales , "offsets" : offsets , "url" : VIRTUAL_RDA_URL } )
display ( Javascript ( js ) )


signature:
def calc_toa_gain_offset ( meta ) :
    """Compute ( gain , offset ) tuples for each band of the specified image metadata"""

implementation:
# Set satellite index to look up cal factors
sat_index = meta [ 'satid' ] . upper ( ) + "_" + meta [ 'bandid' ] . upper ( )
# Set scale for at sensor radiance
# Eq is :
# L = GAIN * DN * ( ACF / EBW ) + Offset
# ACF abscal factor from meta data
# EBW effectiveBandwidth from meta data
# Gain provided by abscal from const
# Offset provided by abscal from const
acf = np . asarray ( meta [ 'abscalfactor' ] )
# Should be nbands length
ebw = np . asarray ( meta [ 'effbandwidth' ] )
# Should be nbands length
gain = np . asarray ( constants . DG_ABSCAL_GAIN [ sat_index ] )
scale = ( acf / ebw ) * gain
offset = np . asarray ( constants . DG_ABSCAL_OFFSET [ sat_index ] )
e_sun_index = meta [ 'satid' ] . upper ( ) + "_" + meta [ 'bandid' ] . upper ( )
e_sun = np . asarray ( constants . DG_ESUN [ e_sun_index ] )
sun = ephem . Sun ( )
img_obs = ephem . Observer ( )
img_obs . lon = meta [ 'latlonhae' ] [ 1 ]
img_obs . lat = meta [ 'latlonhae' ] [ 0 ]
img_obs . elevation = meta [ 'latlonhae' ] [ 2 ]
img_obs . date = datetime . datetime . fromtimestamp ( meta [ 'img_datetime_obj_utc' ] [ '$date' ] / 1000.0 ) . strftime ( '%Y-%m-%d %H:%M:%S.%f' )
sun . compute ( img_obs )
d_es = sun . earth_distance
# Pull sun elevation from the image metadata
# theta _ s can be zenith or elevation - the calc below will us either
# a cos or s in respectively
# theta _ s = float ( self . meta _ dg . IMD . IMAGE . MEANSUNEL )
theta_s = 90 - float ( meta [ 'mean_sun_el' ] )
scale2 = ( d_es ** 2 * np . pi ) / ( e_sun * np . cos ( np . deg2rad ( theta_s ) ) )
# Return scaled data
# Radiance = Scale * Image + offset , Reflectance = Radiance * Scale2
return zip ( scale , scale2 , offset )


signature:
def materialize ( self , node = None , bounds = None , callback = None , out_format = 'TILE_STREAM' , ** kwargs ) :
    """Materializes images into gbdx user buckets in s3.
 Note : This method is only available to RDA based image classes .
 Args :
 node ( str ) : the node in the graph to materialize
 bounds ( list ) : optional bbox for cropping what gets materialized in s3
 out _ format ( str ) : VECTOR _ TILE , VECTOR , TIF , TILE _ STREAM
 callback ( str ) : a callback url like an ` sns : / / `
 Returns :
 job _ id ( str ) : the job _ id of the materialization"""

implementation:
kwargs . update ( { "node" : node , "bounds" : bounds , "callback" : callback , "out_format" : out_format } )
return self . rda . _materialize ( ** kwargs )


signature:
def list ( self ) :
    """Lists available and visible GBDX tasks .
 Returns :
 List of tasks"""

implementation:
r = self . gbdx_connection . get ( self . _base_url )
raise_for_status ( r )
return r . json ( ) [ 'tasks' ]


signature:
def register ( self , task_json = None , json_filename = None ) :
    """Registers a new GBDX task .
 Args :
 task _ json ( dict ) : Dictionary representing task definition .
 json _ filename ( str ) : A full path of a file with json representing the task definition .
 Only one out of task _ json and json _ filename should be provided .
 Returns :
 Response ( str ) ."""

implementation:
if not task_json and not json_filename :
    raise Exception ( "Both task json and filename can't be none." )
if task_json and json_filename :
    raise Exception ( "Both task json and filename can't be provided." )
if json_filename :
    task_json = json . load ( open ( json_filename , 'r' ) )
r = self . gbdx_connection . post ( self . _base_url , json = task_json )
raise_for_status ( r )
return r . text


signature:
def get_definition ( self , task_name ) :
    """Gets definition of a registered GBDX task .
 Args :
 task _ name ( str ) : Task name .
 Returns :
 Dictionary representing the task definition ."""

implementation:
r = self . gbdx_connection . get ( self . _base_url + '/' + task_name )
raise_for_status ( r )
return r . json ( )


signature:
def delete ( self , task_name ) :
    """Deletes a GBDX task .
 Args :
 task _ name ( str ) : Task name .
 Returns :
 Response ( str ) ."""

implementation:
r = self . gbdx_connection . delete ( self . _base_url + '/' + task_name )
raise_for_status ( r )
return r . text


signature:
def update ( self , task_name , task_json ) :
    """Updates a GBDX task .
 Args :
 task _ name ( str ) : Task name .
 task _ json ( dict ) : Dictionary representing updated task definition .
 Returns :
 Dictionary representing the updated task definition ."""

implementation:
r = self . gbdx_connection . put ( self . _base_url + '/' + task_name , json = task_json )
raise_for_status ( r )
return r . json ( )


signature:
def to_geotiff ( arr , path = './output.tif' , proj = None , spec = None , bands = None , ** kwargs ) :
    '''Write out a geotiff file of the image
 Args :
 path ( str ) : path to write the geotiff file to , default is . / output . tif
 proj ( str ) : EPSG string of projection to reproject to
 spec ( str ) : if set to ' rgb ' , write out color - balanced 8 - bit RGB tif
 bands ( list ) : list of bands to export . If spec = ' rgb ' will default to RGB bands
 Returns :
 str : path the geotiff was written to'''

implementation:
assert has_rasterio , "To create geotiff images please install rasterio"
try :
    img_md = arr . rda . metadata [ "image" ]
    x_size = img_md [ "tileXSize" ]
    y_size = img_md [ "tileYSize" ]
except ( AttributeError , KeyError ) :
    x_size = kwargs . get ( "chunk_size" , 256 )
    y_size = kwargs . get ( "chunk_size" , 256 )
try :
    tfm = kwargs [ 'transform' ] if 'transform' in kwargs else arr . affine
except :
    tfm = None
dtype = arr . dtype . name if arr . dtype . name != 'int8' else 'uint8'
if spec is not None and spec . lower ( ) == 'rgb' :
    if bands is None :
        bands = arr . _rgb_bands
    # skip if already DRA ' ed
    if not arr . options . get ( 'dra' ) : # add the RDA HistogramDRA op to get a RGB 8 - bit image
        from gbdxtools . rda . interface import RDA
        rda = RDA ( )
        dra = rda . HistogramDRA ( arr )
        # Reset the bounds and select the bands on the new Dask
        arr = dra . aoi ( bbox = arr . bounds )
    arr = arr [ bands , ... ] . astype ( np . uint8 )
    dtype = 'uint8'
else :
    if bands is not None :
        arr = arr [ bands , ... ]
meta = { 'width' : arr . shape [ 2 ] , 'height' : arr . shape [ 1 ] , 'count' : arr . shape [ 0 ] , 'dtype' : dtype , 'driver' : 'GTiff' , 'transform' : tfm }
if proj is not None :
    meta [ "crs" ] = { 'init' : proj }
if "tiled" in kwargs and kwargs [ "tiled" ] :
    meta . update ( blockxsize = x_size , blockysize = y_size , tiled = "yes" )
with rasterio . open ( path , "w" , ** meta ) as dst :
    writer = rio_writer ( dst )
    result = store ( arr , writer , compute = False )
    result . compute ( scheduler = threaded_get )
return path


signature:
def ingest_vectors ( self , output_port_value ) :
    '''append two required tasks to the given output to ingest to VS'''

implementation:
# append two tasks to self [ ' definition ' ] [ ' tasks ' ]
ingest_task = Task ( 'IngestItemJsonToVectorServices' )
ingest_task . inputs . items = output_port_value
ingest_task . impersonation_allowed = True
stage_task = Task ( 'StageDataToS3' )
stage_task . inputs . destination = 's3://{vector_ingest_bucket}/{recipe_id}/{run_id}/{task_name}'
stage_task . inputs . data = ingest_task . outputs . result . value
self . definition [ 'tasks' ] . append ( ingest_task . generate_task_workflow_json ( ) )
self . definition [ 'tasks' ] . append ( stage_task . generate_task_workflow_json ( ) )


signature:
def get ( self , recipe_id ) :
    '''Retrieves an AnswerFactory Recipe by id
 Args :
 recipe _ id The id of the recipe
 Returns :
 A JSON representation of the recipe'''

implementation:
self . logger . debug ( 'Retrieving recipe by id: ' + recipe_id )
url = '%(base_url)s/recipe/%(recipe_id)s' % { 'base_url' : self . base_url , 'recipe_id' : recipe_id }
r = self . gbdx_connection . get ( url )
r . raise_for_status ( )
return r . json ( )


signature:
def save ( self , recipe ) :
    '''Saves an AnswerFactory Recipe
 Args :
 recipe ( dict ) : Dictionary specifying a recipe
 Returns :
 AnswerFactory Recipe id'''

implementation:
# test if this is a create vs . an update
if 'id' in recipe and recipe [ 'id' ] is not None : # update - > use put op
    self . logger . debug ( "Updating existing recipe: " + json . dumps ( recipe ) )
    url = '%(base_url)s/recipe/json/%(recipe_id)s' % { 'base_url' : self . base_url , 'recipe_id' : recipe [ 'id' ] }
    r = self . gbdx_connection . put ( url , json = recipe )
    try :
        r . raise_for_status ( )
    except :
        print ( r . text )
        raise
    return recipe [ 'id' ]
else : # create - > use post op
    self . logger . debug ( "Creating new recipe: " + json . dumps ( recipe ) )
    url = '%(base_url)s/recipe/json' % { 'base_url' : self . base_url }
    r = self . gbdx_connection . post ( url , json = recipe )
    try :
        r . raise_for_status ( )
    except :
        print ( r . text )
        raise
    recipe_json = r . json ( )
    return recipe_json [ 'id' ]


signature:
def save ( self , project ) :
    '''Saves an AnswerFactory Project
 Args :
 project ( dict ) : Dictionary specifying an AnswerFactory Project .
 Returns :
 AnswerFactory Project id'''

implementation:
# test if this is a create vs . an update
if 'id' in project and project [ 'id' ] is not None : # update - > use put op
    self . logger . debug ( 'Updating existing project: ' + json . dumps ( project ) )
    url = '%(base_url)s/%(project_id)s' % { 'base_url' : self . base_url , 'project_id' : project [ 'id' ] }
    r = self . gbdx_connection . put ( url , json = project )
    try :
        r . raise_for_status ( )
    except :
        print ( r . text )
        raise
    # updates only get the Accepted response - > return the original project id
    return project [ 'id' ]
else :
    self . logger . debug ( 'Creating new project: ' + json . dumps ( project ) )
    # create - > use post op
    url = self . base_url
    r = self . gbdx_connection . post ( url , json = project )
    try :
        r . raise_for_status ( )
    except :
        print ( r . text )
        raise
    project_json = r . json ( )
    # create returns the saved project - > return the project id that ' s saved
    return project_json [ 'id' ]


signature:
def delete ( self , project_id ) :
    '''Deletes a project by id
 Args :
 project _ id : The project id to delete
 Returns :
 Nothing'''

implementation:
self . logger . debug ( 'Deleting project by id: ' + project_id )
url = '%(base_url)s/%(project_id)s' % { 'base_url' : self . base_url , 'project_id' : project_id }
r = self . gbdx_connection . delete ( url )
r . raise_for_status ( )


signature:
def paint ( self ) :
    """Renders a javascript snippet suitable for use as a mapbox - gl line paint entry
 Returns :
 A dict that can be converted to a mapbox - gl javascript paint snippet"""

implementation:
# TODO Figure out why i cant use some of these props
snippet = { 'line-opacity' : VectorStyle . get_style_value ( self . opacity ) , 'line-color' : VectorStyle . get_style_value ( self . color ) , # ' line - cap ' : self . cap ,
# ' line - join ' : self . join ,
'line-width' : VectorStyle . get_style_value ( self . width ) , # ' line - gap - width ' : self . gap _ width ,
# ' line - blur ' : self . blur ,
}
if self . translate :
    snippet [ 'line-translate' ] = self . translate
if self . dasharray :
    snippet [ 'line-dasharray' ] = VectorStyle . get_style_value ( self . dasharray )
return snippet


signature:
def paint ( self ) :
    """Renders a javascript snippet suitable for use as a mapbox - gl fill paint entry
 Returns :
 A dict that can be converted to a mapbox - gl javascript paint snippet"""

implementation:
snippet = { 'fill-opacity' : VectorStyle . get_style_value ( self . opacity ) , 'fill-color' : VectorStyle . get_style_value ( self . color ) , 'fill-outline-color' : VectorStyle . get_style_value ( self . outline_color ) }
if self . translate :
    snippet [ 'fill-translate' ] = self . translate
return snippet


signature:
def paint ( self ) :
    """Renders a javascript snippet suitable for use as a mapbox - gl fill - extrusion paint entry
 Returns :
 A dict that can be converted to a mapbox - gl javascript paint snippet"""

implementation:
snippet = { 'fill-extrusion-opacity' : VectorStyle . get_style_value ( self . opacity ) , 'fill-extrusion-color' : VectorStyle . get_style_value ( self . color ) , 'fill-extrusion-base' : VectorStyle . get_style_value ( self . base ) , 'fill-extrusion-height' : VectorStyle . get_style_value ( self . height ) }
if self . translate :
    snippet [ 'fill-extrusion-translate' ] = self . translate
return snippet


signature:
def paint ( self ) :
    """Renders a javascript snippet suitable for use as a mapbox - gl heatmap paint entry
 Returns :
 A dict that can be converted to a mapbox - gl javascript paint snippet"""

implementation:
snippet = { 'heatmap-radius' : VectorStyle . get_style_value ( self . radius ) , 'heatmap-opacity' : VectorStyle . get_style_value ( self . opacity ) , 'heatmap-color' : VectorStyle . get_style_value ( self . color ) , 'heatmap-intensity' : VectorStyle . get_style_value ( self . intensity ) , 'heatmap-weight' : VectorStyle . get_style_value ( self . weight ) }
return snippet


signature:
def create ( self , vectors ) :
    """Create a vectors in the vector service .
 Args :
 vectors : A single geojson vector or a list of geojson vectors . Item _ type and ingest _ source are required .
 Returns :
 ( list ) : IDs of the vectors created
 Example :
 > > > vectors . create (
 . . . " type " : " Feature " ,
 . . . " geometry " : {
 . . . " type " : " Point " ,
 . . . " coordinates " : [ 1.0,1.0]
 . . . " properties " : {
 . . . " text " : " item text " ,
 . . . " name " : " item name " ,
 . . . " item _ type " : " type " ,
 . . . " ingest _ source " : " source " ,
 . . . " attributes " : {
 . . . " latitude " : 1,
 . . . " institute _ founded " : " 2015-07-17 " ,
 . . . " mascot " : " moth " """

implementation:
if type ( vectors ) is dict :
    vectors = [ vectors ]
# validate they all have item _ type and ingest _ source in properties
for vector in vectors :
    if not 'properties' in list ( vector . keys ( ) ) :
        raise Exception ( 'Vector does not contain "properties" field.' )
    if not 'item_type' in list ( vector [ 'properties' ] . keys ( ) ) :
        raise Exception ( 'Vector does not contain "item_type".' )
    if not 'ingest_source' in list ( vector [ 'properties' ] . keys ( ) ) :
        raise Exception ( 'Vector does not contain "ingest_source".' )
r = self . gbdx_connection . post ( self . create_url , data = json . dumps ( vectors ) )
r . raise_for_status ( )
return r . json ( )


signature:
def create_from_wkt ( self , wkt , item_type , ingest_source , ** attributes ) :
    '''Create a single vector in the vector service
 Args :
 wkt ( str ) : wkt representation of the geometry
 item _ type ( str ) : item _ type of the vector
 ingest _ source ( str ) : source of the vector
 attributes : a set of key - value pairs of attributes
 Returns :
 id ( str ) : string identifier of the vector created'''

implementation:
# verify the " depth " of the attributes is single layer
geojson = load_wkt ( wkt ) . __geo_interface__
vector = { 'type' : "Feature" , 'geometry' : geojson , 'properties' : { 'item_type' : item_type , 'ingest_source' : ingest_source , 'attributes' : attributes } }
return self . create ( vector ) [ 0 ]


signature:
def get ( self , ID , index = 'vector-web-s' ) :
    '''Retrieves a vector . Not usually necessary because searching is the best way to find & get stuff .
 Args :
 ID ( str ) : ID of the vector object
 index ( str ) : Optional . Index the object lives in . defaults to ' vector - web - s '
 Returns :
 record ( dict ) : A dict object identical to the json representation of the catalog record'''

implementation:
url = self . get_url % index
r = self . gbdx_connection . get ( url + ID )
r . raise_for_status ( )
return r . json ( )


signature:
def query ( self , searchAreaWkt , query , count = 100 , ttl = '5m' , index = default_index ) :
    '''Perform a vector services query using the QUERY API
 ( https : / / gbdxdocs . digitalglobe . com / docs / vs - query - list - vector - items - returns - default - fields )
 Args :
 searchAreaWkt : WKT Polygon of area to search
 query : Elastic Search query
 count : Maximum number of results to return
 ttl : Amount of time for each temporary vector page to exist
 Returns :
 List of vector results'''

implementation:
if count < 1000 : # issue a single page query
    search_area_polygon = from_wkt ( searchAreaWkt )
    left , lower , right , upper = search_area_polygon . bounds
    params = { "q" : query , "count" : min ( count , 1000 ) , "left" : left , "right" : right , "lower" : lower , "upper" : upper }
    url = self . query_index_url % index if index else self . query_url
    r = self . gbdx_connection . get ( url , params = params )
    r . raise_for_status ( )
    return r . json ( )
else :
    return list ( self . query_iteratively ( searchAreaWkt , query , count , ttl , index ) )


signature:
def query_iteratively ( self , searchAreaWkt , query , count = 100 , ttl = '5m' , index = default_index ) :
    '''Perform a vector services query using the QUERY API
 ( https : / / gbdxdocs . digitalglobe . com / docs / vs - query - list - vector - items - returns - default - fields )
 Args :
 searchAreaWkt : WKT Polygon of area to search
 query : Elastic Search query
 count : Maximum number of results to return
 ttl : Amount of time for each temporary vector page to exist
 Returns :
 generator of vector results'''

implementation:
search_area_polygon = from_wkt ( searchAreaWkt )
left , lower , right , upper = search_area_polygon . bounds
params = { "q" : query , "count" : min ( count , 1000 ) , "ttl" : ttl , "left" : left , "right" : right , "lower" : lower , "upper" : upper }
# initialize paging request
url = self . query_index_page_url % index if index else self . query_page_url
r = self . gbdx_connection . get ( url , params = params )
r . raise_for_status ( )
page = r . json ( )
paging_id = page [ 'next_paging_id' ]
item_count = int ( page [ 'item_count' ] )
data = page [ 'data' ]
num_results = 0
for vector in data :
    num_results += 1
    if num_results > count :
        break
    yield vector
if num_results == count :
    return
# get vectors from each page
while paging_id and item_count > 0 and num_results < count :
    headers = { 'Content-Type' : 'application/x-www-form-urlencoded' }
    data = { "pagingId" : paging_id , "ttl" : ttl }
    r = self . gbdx_connection . post ( self . page_url , headers = headers , data = data )
    r . raise_for_status ( )
    page = r . json ( )
    paging_id = page [ 'next_paging_id' ]
    item_count = int ( page [ 'item_count' ] )
    data = page [ 'data' ]
    for vector in data :
        num_results += 1
        if num_results > count :
            break
        yield vector


signature:
def aggregate_query ( self , searchAreaWkt , agg_def , query = None , start_date = None , end_date = None , count = 10 , index = default_index ) :
    """Aggregates results of a query into buckets defined by the ' agg _ def ' parameter . The aggregations are
 represented by dicts containing a ' name ' key and a ' terms ' key holding a list of the aggregation buckets .
 Each bucket element is a dict containing a ' term ' key containing the term used for this bucket , a ' count ' key
 containing the count of items that match this bucket , and an ' aggregations ' key containing any child
 aggregations .
 Args :
 searchAreaWkt ( str ) : wkt representation of the geometry
 agg _ def ( str or AggregationDef ) : the aggregation definitions
 query ( str ) : a valid Elasticsearch query string to constrain the items going into the aggregation
 start _ date ( str ) : either an ISO - 8601 date string or a ' now ' expression ( e . g . " now - 6d " or just " now " )
 end _ date ( str ) : either an ISO - 8601 date string or a ' now ' expression ( e . g . " now - 6d " or just " now " )
 count ( int ) : the number of buckets to include in the aggregations ( the top N will be returned )
 index ( str ) : the index ( or alias or wildcard index expression ) to run aggregations against , set to None for the entire set of vector indexes
 Returns :
 results ( list ) : A ( usually single - element ) list of dict objects containing the aggregation results ."""

implementation:
geojson = load_wkt ( searchAreaWkt ) . __geo_interface__
aggs_str = str ( agg_def )
# could be string or AggregationDef
params = { "count" : count , "aggs" : aggs_str }
if query :
    params [ 'query' ] = query
if start_date :
    params [ 'start_date' ] = start_date
if end_date :
    params [ 'end_date' ] = end_date
url = self . aggregations_by_index_url % index if index else self . aggregations_url
r = self . gbdx_connection . post ( url , params = params , json = geojson )
r . raise_for_status ( )
return r . json ( object_pairs_hook = OrderedDict ) [ 'aggregations' ]


signature:
def tilemap ( self , query , styles = { } , bbox = [ - 180 , - 90 , 180 , 90 ] , zoom = 16 , api_key = os . environ . get ( 'MAPBOX_API_KEY' , None ) , image = None , image_bounds = None , index = "vector-user-provided" , name = "GBDX_Task_Output" , ** kwargs ) :
    """Renders a mapbox gl map from a vector service query"""

implementation:
try :
    from IPython . display import display
except :
    print ( "IPython is required to produce maps." )
    return
assert api_key is not None , "No Mapbox API Key found. You can either pass in a token or set the MAPBOX_API_KEY environment variable."
wkt = box ( * bbox ) . wkt
features = self . query ( wkt , query , index = index )
union = cascaded_union ( [ shape ( f [ 'geometry' ] ) for f in features ] )
lon , lat = union . centroid . coords [ 0 ]
url = 'https://vector.geobigdata.io/insight-vector/api/mvt/{z}/{x}/{y}?' ;
url += 'q={}&index={}' . format ( query , index ) ;
if styles is not None and not isinstance ( styles , list ) :
    styles = [ styles ]
map_id = "map_{}" . format ( str ( int ( time . time ( ) ) ) )
map_data = VectorTileLayer ( url , source_name = name , styles = styles , ** kwargs )
image_layer = self . _build_image_layer ( image , image_bounds )
template = BaseTemplate ( map_id , ** { "lat" : lat , "lon" : lon , "zoom" : zoom , "datasource" : json . dumps ( map_data . datasource ) , "layers" : json . dumps ( map_data . layers ) , "image_layer" : image_layer , "mbkey" : api_key , "token" : self . gbdx_connection . access_token } )
template . inject ( )