signature:
def map ( self , features = None , query = None , styles = None , bbox = [ - 180 , - 90 , 180 , 90 ] , zoom = 10 , center = None , image = None , image_bounds = None , cmap = 'viridis' , api_key = os . environ . get ( 'MAPBOX_API_KEY' , None ) , ** kwargs ) :
    """Renders a mapbox gl map from a vector service query or a list of geojson features
 Args :
 features ( list ) : a list of geojson features
 query ( str ) : a VectorServices query
 styles ( list ) : a list of VectorStyles to apply to the features
 bbox ( list ) : a bounding box to query for features ( [ minx , miny , maxx , maxy ] )
 zoom ( int ) : the initial zoom level of the map
 center ( list ) : a list of [ lat , lon ] used to center the map
 api _ key ( str ) : a valid Mapbox API key
 image ( dict ) : a CatalogImage or a ndarray
 image _ bounds ( list ) : a list of bounds for image positioning
 Use outside of GBDX Notebooks requires a MapBox API key , sign up for free at https : / / www . mapbox . com / pricing /
 Pass the key using the ` api _ key ` keyword or set an environmental variable called ` MAPBOX API KEY `
 cmap ( str ) : MatPlotLib colormap to use for rendering single band images ( default : viridis )"""

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 key or set the MAPBOX_API_KEY environment variable. Use outside of GBDX Notebooks requires a MapBox API key, sign up for free at https://www.mapbox.com/pricing/"
if features is None and query is not None :
    wkt = box ( * bbox ) . wkt
    features = self . query ( wkt , query , index = None )
elif features is None and query is None and image is None :
    print ( 'Must provide either a list of features or a query or an image' )
    return
if styles is not None and not isinstance ( styles , list ) :
    styles = [ styles ]
geojson = { "type" : "FeatureCollection" , "features" : features }
if center is None and features is not None :
    union = cascaded_union ( [ shape ( f [ 'geometry' ] ) for f in features ] )
    lon , lat = union . centroid . coords [ 0 ]
elif center is None and image is not None :
    try :
        lon , lat = shape ( image ) . centroid . coords [ 0 ]
    except :
        lon , lat = box ( * image_bounds ) . centroid . coords [ 0 ]
else :
    lat , lon = center
map_id = "map_{}" . format ( str ( int ( time . time ( ) ) ) )
map_data = VectorGeojsonLayer ( geojson , styles = styles , ** kwargs )
image_layer = self . _build_image_layer ( image , image_bounds , cmap )
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" : 'dummy' } )
template . inject ( )


signature:
def read ( self , bands = None , ** kwargs ) :
    """Reads data from a dask array and returns the computed ndarray matching the given bands
 Args :
 bands ( list ) : band indices to read from the image . Returns bands in the order specified in the list of bands .
 Returns :
 ndarray : a numpy array of image data"""

implementation:
arr = self
if bands is not None :
    arr = self [ bands , ... ]
return arr . compute ( scheduler = threaded_get )


signature:
def randwindow ( self , window_shape ) :
    """Get a random window of a given shape from within an image
 Args :
 window _ shape ( tuple ) : The desired shape of the returned image as ( height , width ) in pixels .
 Returns :
 image : a new image object of the specified shape and same type"""

implementation:
row = random . randrange ( window_shape [ 0 ] , self . shape [ 1 ] )
col = random . randrange ( window_shape [ 1 ] , self . shape [ 2 ] )
return self [ : , row - window_shape [ 0 ] : row , col - window_shape [ 1 ] : col ]


signature:
def iterwindows ( self , count = 64 , window_shape = ( 256 , 256 ) ) :
    """Iterate over random windows of an image
 Args :
 count ( int ) : the number of the windows to generate . Defaults to 64 , if ` None ` will continue to iterate over random windows until stopped .
 window _ shape ( tuple ) : The desired shape of each image as ( height , width ) in pixels .
 Yields :
 image : an image of the given shape and same type ."""

implementation:
if count is None :
    while True :
        yield self . randwindow ( window_shape )
else :
    for i in xrange ( count ) :
        yield self . randwindow ( window_shape )


signature:
def window_at ( self , geom , window_shape ) :
    """Return a subsetted window of a given size , centered on a geometry object
 Useful for generating training sets from vector training data
 Will throw a ValueError if the window is not within the image bounds
 Args :
 geom ( shapely , geometry ) : Geometry to center the image on
 window _ shape ( tuple ) : The desired shape of the image as ( height , width ) in pixels .
 Returns :
 image : image object of same type"""

implementation:
# Centroids of the input geometry may not be centered on the object .
# For a covering image we use the bounds instead .
# This is also a workaround for issue 387.
y_size , x_size = window_shape [ 0 ] , window_shape [ 1 ]
bounds = box ( * geom . bounds )
px = ops . transform ( self . __geo_transform__ . rev , bounds ) . centroid
miny , maxy = int ( px . y - y_size / 2 ) , int ( px . y + y_size / 2 )
minx , maxx = int ( px . x - x_size / 2 ) , int ( px . x + x_size / 2 )
_ , y_max , x_max = self . shape
if minx < 0 or miny < 0 or maxx > x_max or maxy > y_max :
    raise ValueError ( "Input geometry resulted in a window outside of the image" )
return self [ : , miny : maxy , minx : maxx ]


signature:
def window_cover ( self , window_shape , pad = True ) :
    """Iterate over a grid of windows of a specified shape covering an image .
 The image is divided into a grid of tiles of size window _ shape . Each iteration returns
 the next window .
 Args :
 window _ shape ( tuple ) : The desired shape of each image as ( height ,
 width ) in pixels .
 pad : ( bool ) : Whether or not to pad edge cells . If False , cells that do not
 have the desired shape will not be returned . Defaults to True .
 Yields :
 image : image object of same type ."""

implementation:
size_y , size_x = window_shape [ 0 ] , window_shape [ 1 ]
_ndepth , _nheight , _nwidth = self . shape
nheight , _m = divmod ( _nheight , size_y )
nwidth , _n = divmod ( _nwidth , size_x )
img = self
if pad is True :
    new_height , new_width = _nheight , _nwidth
    if _m != 0 :
        new_height = ( nheight + 1 ) * size_y
    if _n != 0 :
        new_width = ( nwidth + 1 ) * size_x
    if ( new_height , new_width ) != ( _nheight , _nwidth ) :
        bounds = box ( 0 , 0 , new_width , new_height )
        geom = ops . transform ( self . __geo_transform__ . fwd , bounds )
        img = self [ geom ]
row_lims = range ( 0 , img . shape [ 1 ] , size_y )
col_lims = range ( 0 , img . shape [ 2 ] , size_x )
for maxy , maxx in product ( row_lims , col_lims ) :
    reg = img [ : , maxy : ( maxy + size_y ) , maxx : ( maxx + size_x ) ]
    if pad is False :
        if reg . shape [ 1 : ] == window_shape :
            yield reg
    else :
        yield reg


signature:
def aoi ( self , ** kwargs ) :
    """Subsets the Image by the given bounds
 Args :
 bbox ( list ) : optional . A bounding box array [ minx , miny , maxx , maxy ]
 wkt ( str ) : optional . A WKT geometry string
 geojson ( str ) : optional . A GeoJSON geometry dictionary
 Returns :
 image : an image instance of the same type"""

implementation:
g = self . _parse_geoms ( ** kwargs )
if g is None :
    return self
else :
    return self [ g ]


signature:
def pxbounds ( self , geom , clip = False ) :
    """Returns the bounds of a geometry object in pixel coordinates
 Args :
 geom : Shapely geometry object or GeoJSON as Python dictionary or WKT string
 clip ( bool ) : Clip the bounds to the min / max extent of the image
 Returns :
 list : bounds in pixels [ min x , min y , max x , max y ] clipped to image bounds"""

implementation:
try :
    if isinstance ( geom , dict ) :
        if 'geometry' in geom :
            geom = shape ( geom [ 'geometry' ] )
        else :
            geom = shape ( geom )
    elif isinstance ( geom , BaseGeometry ) :
        geom = shape ( geom )
    else :
        geom = wkt . loads ( geom )
except :
    raise TypeError ( "Invalid geometry object" )
# if geometry doesn ' t overlap the image , return an error
if geom . disjoint ( shape ( self ) ) :
    raise ValueError ( "Geometry outside of image bounds" )
# clip to pixels within the image
( xmin , ymin , xmax , ymax ) = ops . transform ( self . __geo_transform__ . rev , geom ) . bounds
_nbands , ysize , xsize = self . shape
if clip :
    xmin = max ( xmin , 0 )
    ymin = max ( ymin , 0 )
    xmax = min ( xmax , xsize )
    ymax = min ( ymax , ysize )
return ( xmin , ymin , xmax , ymax )


signature:
def geotiff ( self , ** kwargs ) :
    """Creates a geotiff on the filesystem
 Args :
 path ( str ) : optional , path to write the geotiff file to , default is . / output . tif
 proj ( str ) : optional , EPSG string of projection to reproject to
 spec ( str ) : optional , if set to ' rgb ' , write out color - balanced 8 - bit RGB tif
 bands ( list ) : optional , list of bands to export . If spec = ' rgb ' will default to RGB bands ,
 otherwise will export all bands
 Returns :
 str : path the geotiff was written to"""

implementation:
if 'proj' not in kwargs :
    kwargs [ 'proj' ] = self . proj
return to_geotiff ( self , ** kwargs )


signature:
def warp ( self , dem = None , proj = "EPSG:4326" , ** kwargs ) :
    """Delayed warp across an entire AOI or Image
 Creates a new dask image by deferring calls to the warp _ geometry on chunks
 Args :
 dem ( ndarray ) : optional . A DEM for warping to specific elevation planes
 proj ( str ) : optional . An EPSG proj string to project the image data into ( " EPSG : 32612 " )
 Returns :
 daskarray : a warped image as deferred image array"""

implementation:
try :
    img_md = self . 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 )
# Create an affine transform to convert between real - world and pixels
if self . proj is None :
    from_proj = "EPSG:4326"
else :
    from_proj = self . proj
try : # NOTE : this only works on images that have rda rpcs metadata
    center = wkt . loads ( self . rda . metadata [ "image" ] [ "imageBoundsWGS84" ] ) . centroid
    g = box ( * ( center . buffer ( self . rda . metadata [ "rpcs" ] [ "gsd" ] / 2 ) . bounds ) )
    tfm = partial ( pyproj . transform , pyproj . Proj ( init = "EPSG:4326" ) , pyproj . Proj ( init = proj ) )
    gsd = kwargs . get ( "gsd" , ops . transform ( tfm , g ) . area ** 0.5 )
    current_bounds = wkt . loads ( self . rda . metadata [ "image" ] [ "imageBoundsWGS84" ] ) . bounds
except ( AttributeError , KeyError , TypeError ) :
    tfm = partial ( pyproj . transform , pyproj . Proj ( init = self . proj ) , pyproj . Proj ( init = proj ) )
    gsd = kwargs . get ( "gsd" , ( ops . transform ( tfm , shape ( self ) ) . area / ( self . shape [ 1 ] * self . shape [ 2 ] ) ) ** 0.5 )
    current_bounds = self . bounds
tfm = partial ( pyproj . transform , pyproj . Proj ( init = from_proj ) , pyproj . Proj ( init = proj ) )
itfm = partial ( pyproj . transform , pyproj . Proj ( init = proj ) , pyproj . Proj ( init = from_proj ) )
output_bounds = ops . transform ( tfm , box ( * current_bounds ) ) . bounds
gtf = Affine . from_gdal ( output_bounds [ 0 ] , gsd , 0.0 , output_bounds [ 3 ] , 0.0 , - 1 * gsd )
ll = ~ gtf * ( output_bounds [ : 2 ] )
ur = ~ gtf * ( output_bounds [ 2 : ] )
x_chunks = int ( ( ur [ 0 ] - ll [ 0 ] ) / x_size ) + 1
y_chunks = int ( ( ll [ 1 ] - ur [ 1 ] ) / y_size ) + 1
num_bands = self . shape [ 0 ]
try :
    dtype = RDA_TO_DTYPE [ img_md [ "dataType" ] ]
except :
    dtype = 'uint8'
daskmeta = { "dask" : { } , "chunks" : ( num_bands , y_size , x_size ) , "dtype" : dtype , "name" : "warp-{}" . format ( self . name ) , "shape" : ( num_bands , y_chunks * y_size , x_chunks * x_size ) }
def px_to_geom ( xmin , ymin ) :
    xmax = int ( xmin + x_size )
    ymax = int ( ymin + y_size )
    bounds = list ( ( gtf * ( xmin , ymax ) ) + ( gtf * ( xmax , ymin ) ) )
    return box ( * bounds )
full_bounds = box ( * output_bounds )
dasks = [ ]
if isinstance ( dem , GeoDaskImage ) :
    if dem . proj != proj :
        dem = dem . warp ( proj = proj , dem = dem )
    dasks . append ( dem . dask )
for y in xrange ( y_chunks ) :
    for x in xrange ( x_chunks ) :
        xmin = x * x_size
        ymin = y * y_size
        geometry = px_to_geom ( xmin , ymin )
        daskmeta [ "dask" ] [ ( daskmeta [ "name" ] , 0 , y , x ) ] = ( self . _warp , geometry , gsd , dem , proj , dtype , 5 )
daskmeta [ "dask" ] , _ = optimization . cull ( HighLevelGraph . merge ( daskmeta [ "dask" ] , * dasks ) , list ( daskmeta [ "dask" ] . keys ( ) ) )
gi = mapping ( full_bounds )
gt = AffineTransform ( gtf , proj )
image = GeoDaskImage ( daskmeta , __geo_interface__ = gi , __geo_transform__ = gt )
return image [ box ( * output_bounds ) ]


signature:
def _parse_geoms ( self , ** kwargs ) :
    """Finds supported geometry types , parses them and returns the bbox"""

implementation:
bbox = kwargs . get ( 'bbox' , None )
wkt_geom = kwargs . get ( 'wkt' , None )
geojson = kwargs . get ( 'geojson' , None )
if bbox is not None :
    g = box ( * bbox )
elif wkt_geom is not None :
    g = wkt . loads ( wkt_geom )
elif geojson is not None :
    g = shape ( geojson )
else :
    return None
if self . proj is None :
    return g
else :
    return self . _reproject ( g , from_proj = kwargs . get ( 'from_proj' , 'EPSG:4326' ) )


signature:
def load_url ( url , shape = ( 8 , 256 , 256 ) ) :
    """Loads a geotiff url inside a thread and returns as an ndarray"""

implementation:
thread_id = threading . current_thread ( ) . ident
_curl = _curl_pool [ thread_id ]
_curl . setopt ( _curl . URL , url )
_curl . setopt ( pycurl . NOSIGNAL , 1 )
_ , ext = os . path . splitext ( urlparse ( url ) . path )
with NamedTemporaryFile ( prefix = "gbdxtools" , suffix = "." + ext , delete = False ) as temp : # TODO : apply correct file extension
    _curl . setopt ( _curl . WRITEDATA , temp . file )
    _curl . perform ( )
    code = _curl . getinfo ( pycurl . HTTP_CODE )
    try :
        if ( code != 200 ) :
            raise TypeError ( "Request for {} returned unexpected error code: {}" . format ( url , code ) )
        arr = np . rollaxis ( imread ( temp ) , 2 , 0 )
    except Exception as e :
        print ( e )
        temp . seek ( 0 )
        print ( temp . read ( ) )
        arr = np . zeros ( shape , dtype = np . uint8 )
        _curl . close ( )
        del _curl_pool [ thread_id ]
    finally :
        temp . file . flush ( )
        temp . close ( )
        os . remove ( temp . name )
    return arr


signature:
def _tile_coords ( self , bounds ) :
    """convert mercator bbox to tile index limits"""

implementation:
tfm = partial ( pyproj . transform , pyproj . Proj ( init = "epsg:3857" ) , pyproj . Proj ( init = "epsg:4326" ) )
bounds = ops . transform ( tfm , box ( * bounds ) ) . bounds
# because tiles have a common corner , the tiles that cover a
# given tile includes the adjacent neighbors .
# https : / / github . com / mapbox / mercantile / issues / 84 # issuecomment - 413113791
west , south , east , north = bounds
epsilon = 1.0e-10
if east != west and north != south : # 2D bbox
# shrink the bounds a small amount so that
# shapes / tiles round trip .
    west += epsilon
    south += epsilon
    east -= epsilon
    north -= epsilon
params = [ west , south , east , north , [ self . zoom_level ] ]
tile_coords = [ ( tile . x , tile . y ) for tile in mercantile . tiles ( * params ) ]
xtiles , ytiles = zip ( * tile_coords )
minx = min ( xtiles )
miny = min ( ytiles )
maxx = max ( xtiles )
maxy = max ( ytiles )
return minx , miny , maxx , maxy


signature:
def get ( self , key , default = None ) :
    """> > > inputs = InputPorts ( { " one " : 1 } )
 > > > " one " in inputs . _ ports
 True
 > > > " one " in inputs . _ vals
 True
 > > > inputs . get ( " one " , 2 ) = = 1
 True
 > > > inputs . get ( " two " , 2 ) = = 2
 True
 > > > " two " in inputs . _ ports
 True
 > > > " two " in inputs . _ vals
 False"""

implementation:
if key not in self . _ports :
    self . _ports [ key ] = self . _port_template ( key )
return self . _vals . get ( key , default )


signature:
def load_url ( url , token , shape = ( 8 , 256 , 256 ) ) :
    """Loads a geotiff url inside a thread and returns as an ndarray"""

implementation:
_ , ext = os . path . splitext ( urlparse ( url ) . path )
success = False
for i in xrange ( MAX_RETRIES ) :
    thread_id = threading . current_thread ( ) . ident
    _curl = _curl_pool [ thread_id ]
    _curl . setopt ( _curl . URL , url )
    _curl . setopt ( pycurl . NOSIGNAL , 1 )
    _curl . setopt ( pycurl . HTTPHEADER , [ 'Authorization: Bearer {}' . format ( token ) ] )
    with NamedTemporaryFile ( prefix = "gbdxtools" , suffix = ext , delete = False ) as temp : # TODO : apply correct file extension
        _curl . setopt ( _curl . WRITEDATA , temp . file )
        _curl . perform ( )
        code = _curl . getinfo ( pycurl . HTTP_CODE )
        try :
            if ( code != 200 ) :
                raise TypeError ( "Request for {} returned unexpected error code: {}" . format ( url , code ) )
            temp . file . flush ( )
            temp . close ( )
            arr = imread ( temp . name )
            if len ( arr . shape ) == 3 :
                arr = np . rollaxis ( arr , 2 , 0 )
            else :
                arr = np . expand_dims ( arr , axis = 0 )
            success = True
            return arr
        except Exception as e :
            _curl . close ( )
            del _curl_pool [ thread_id ]
        finally :
            temp . close ( )
            os . remove ( temp . name )
if success is False :
    raise TypeError ( "Request for {} returned unexpected error code: {}" . format ( url , code ) )
return arr


signature:
def launch ( self , workflow ) :
    """Launches GBDX workflow .
 Args :
 workflow ( dict ) : Dictionary specifying workflow tasks .
 Returns :
 Workflow id ( str ) ."""

implementation:
# hit workflow api
try :
    r = self . gbdx_connection . post ( self . workflows_url , json = workflow )
    try :
        r . raise_for_status ( )
    except :
        print ( "GBDX API Status Code: %s" % r . status_code )
        print ( "GBDX API Response: %s" % r . text )
        r . raise_for_status ( )
    workflow_id = r . json ( ) [ 'id' ]
    return workflow_id
except TypeError :
    self . logger . debug ( 'Workflow not launched!' )


signature:
def status ( self , workflow_id ) :
    """Checks workflow status .
 Args :
 workflow _ id ( str ) : Workflow id .
 Returns :
 Workflow status ( str ) ."""

implementation:
self . logger . debug ( 'Get status of workflow: ' + workflow_id )
url = '%(wf_url)s/%(wf_id)s' % { 'wf_url' : self . workflows_url , 'wf_id' : workflow_id }
r = self . gbdx_connection . get ( url )
r . raise_for_status ( )
return r . json ( ) [ 'state' ]


signature:
def get_stdout ( self , workflow_id , task_id ) :
    """Get stdout for a particular task .
 Args :
 workflow _ id ( str ) : Workflow id .
 task _ id ( str ) : Task id .
 Returns :
 Stdout of the task ( string ) ."""

implementation:
url = '%(wf_url)s/%(wf_id)s/tasks/%(task_id)s/stdout' % { 'wf_url' : self . workflows_url , 'wf_id' : workflow_id , 'task_id' : task_id }
r = self . gbdx_connection . get ( url )
r . raise_for_status ( )
return r . text


signature:
def cancel ( self , workflow_id ) :
    """Cancels a running workflow .
 Args :
 workflow _ id ( str ) : Workflow id .
 Returns :
 Nothing"""

implementation:
self . logger . debug ( 'Canceling workflow: ' + workflow_id )
url = '%(wf_url)s/%(wf_id)s/cancel' % { 'wf_url' : self . workflows_url , 'wf_id' : workflow_id }
r = self . gbdx_connection . post ( url , data = '' )
r . raise_for_status ( )


signature:
def launch_batch_workflow ( self , batch_workflow ) :
    """Launches GBDX batch workflow .
 Args :
 batch _ workflow ( dict ) : Dictionary specifying batch workflow tasks .
 Returns :
 Batch Workflow id ( str ) ."""

implementation:
# hit workflow api
url = '%(base_url)s/batch_workflows' % { 'base_url' : self . base_url }
try :
    r = self . gbdx_connection . post ( url , json = batch_workflow )
    batch_workflow_id = r . json ( ) [ 'batch_workflow_id' ]
    return batch_workflow_id
except TypeError as e :
    self . logger . debug ( 'Batch Workflow not launched, reason: {0}' . format ( e ) )


signature:
def batch_workflow_status ( self , batch_workflow_id ) :
    """Checks GBDX batch workflow status .
 Args :
 batch workflow _ id ( str ) : Batch workflow id .
 Returns :
 Batch Workflow status ( str ) ."""

implementation:
self . logger . debug ( 'Get status of batch workflow: ' + batch_workflow_id )
url = '%(base_url)s/batch_workflows/%(batch_id)s' % { 'base_url' : self . base_url , 'batch_id' : batch_workflow_id }
r = self . gbdx_connection . get ( url )
return r . json ( )


signature:
def batch_workflow_cancel ( self , batch_workflow_id ) :
    """Cancels GBDX batch workflow .
 Args :
 batch workflow _ id ( str ) : Batch workflow id .
 Returns :
 Batch Workflow status ( str ) ."""

implementation:
self . logger . debug ( 'Cancel batch workflow: ' + batch_workflow_id )
url = '%(base_url)s/batch_workflows/%(batch_id)s/cancel' % { 'base_url' : self . base_url , 'batch_id' : batch_workflow_id }
r = self . gbdx_connection . post ( url )
return r . json ( )


signature:
def search ( self , lookback_h = 12 , owner = None , state = "all" ) :
    """Cancels GBDX batch workflow .
 Params :
 lookback _ h ( int ) : Look back time in hours .
 owner ( str ) : Workflow owner to search by
 state ( str ) : State to filter by , eg :
 " submitted " ,
 " scheduled " ,
 " started " ,
 " canceled " ,
 " cancelling " ,
 " failed " ,
 " succeeded " ,
 " timedout " ,
 " pending " ,
 " running " ,
 " complete " ,
 " waiting " ,
 " all "
 Returns :
 Batch Workflow status ( str ) ."""

implementation:
postdata = { "lookback_h" : lookback_h , "state" : state }
if owner is not None :
    postdata [ 'owner' ] = owner
url = "{}/workflows/search" . format ( self . base_url )
headers = { 'Content-Type' : 'application/json' }
r = self . gbdx_connection . post ( url , headers = headers , data = json . dumps ( postdata ) )
return r . json ( )


signature:
def order ( self , image_catalog_ids , batch_size = 100 , callback = None ) :
    '''Orders images from GBDX .
 Args :
 image _ catalog _ ids ( str or list ) : A single catalog id or a list of
 catalog ids .
 batch _ size ( int ) : The image _ catalog _ ids will be split into
 batches of batch _ size . The ordering API max
 batch size is 100 , if batch _ size is greater
 than 100 it will be truncated .
 callback ( str ) : A url to call when ordering is completed .
 Returns :
 order _ ids ( str or list ) : If one batch , returns a string . If more
 than one batch , returns a list of order ids ,
 one for each batch .'''

implementation:
def _order_single_batch ( url_ , ids , results_list ) :
    data = json . dumps ( ids ) if callback is None else json . dumps ( { "acquisitionIds" : ids , "callback" : callback } )
    r = self . gbdx_connection . post ( url_ , data = data )
    r . raise_for_status ( )
    order_id = r . json ( ) . get ( "order_id" )
    if order_id :
        results_list . append ( order_id )
self . logger . debug ( 'Place order' )
url = ( '%s/order' if callback is None else '%s/ordercb' ) % self . base_url
batch_size = min ( 100 , batch_size )
if not isinstance ( image_catalog_ids , list ) :
    image_catalog_ids = [ image_catalog_ids ]
sanitized_ids = list ( set ( ( id for id in ( _id . strip ( ) for _id in image_catalog_ids ) if id ) ) )
res = [ ]
# Use itertool batch recipe
acq_ids_by_batch = zip ( * ( [ iter ( sanitized_ids ) ] * batch_size ) )
for ids_batch in acq_ids_by_batch :
    _order_single_batch ( url , ids_batch , res )
# Order reminder
remain_count = len ( sanitized_ids ) % batch_size
if remain_count > 0 :
    _order_single_batch ( url , sanitized_ids [ - remain_count : ] , res )
if len ( res ) == 1 :
    return res [ 0 ]
elif len ( res ) > 1 :
    return res


signature:
def status ( self , order_id ) :
    '''Checks imagery order status . There can be more than one image per
 order and this function returns the status of all images
 within the order .
 Args :
 order _ id ( str ) : The id of the order placed .
 Returns :
 List of dictionaries , one per image . Each dictionary consists
 of the keys ' acquisition _ id ' , ' location ' and ' state ' .'''

implementation:
self . logger . debug ( 'Get status of order ' + order_id )
url = '%(base_url)s/order/%(order_id)s' % { 'base_url' : self . base_url , 'order_id' : order_id }
r = self . gbdx_connection . get ( url )
r . raise_for_status ( )
return r . json ( ) . get ( "acquisitions" , { } )


signature:
def heartbeat ( self ) :
    '''Check the heartbeat of the ordering API
 Args : None
 Returns : True or False'''

implementation:
url = '%s/heartbeat' % self . base_url
# Auth is not required to hit the heartbeat
r = requests . get ( url )
try :
    return r . json ( ) == "ok"
except :
    return False


signature:
def get ( self , catID , includeRelationships = False ) :
    '''Retrieves the strip footprint WKT string given a cat ID .
 Args :
 catID ( str ) : The source catalog ID from the platform catalog .
 includeRelationships ( bool ) : whether to include graph links to related objects . Default False .
 Returns :
 record ( dict ) : A dict object identical to the json representation of the catalog record'''

implementation:
url = '%(base_url)s/record/%(catID)s' % { 'base_url' : self . base_url , 'catID' : catID }
r = self . gbdx_connection . get ( url )
r . raise_for_status ( )
return r . json ( )


signature:
def get_strip_metadata ( self , catID ) :
    '''Retrieves the strip catalog metadata given a cat ID .
 Args :
 catID ( str ) : The source catalog ID from the platform catalog .
 Returns :
 metadata ( dict ) : A metadata dictionary .
 TODO : have this return a class object with interesting information exposed .'''

implementation:
self . logger . debug ( 'Retrieving strip catalog metadata' )
url = '%(base_url)s/record/%(catID)s?includeRelationships=false' % { 'base_url' : self . base_url , 'catID' : catID }
r = self . gbdx_connection . get ( url )
if r . status_code == 200 :
    return r . json ( ) [ 'properties' ]
elif r . status_code == 404 :
    self . logger . debug ( 'Strip not found: %s' % catID )
    r . raise_for_status ( )
else :
    self . logger . debug ( 'There was a problem retrieving catid: %s' % catID )
    r . raise_for_status ( )


signature:
def get_address_coords ( self , address ) :
    '''Use the google geocoder to get latitude and longitude for an address string
 Args :
 address : any address string
 Returns :
 A tuple of ( lat , lng )'''

implementation:
url = "https://maps.googleapis.com/maps/api/geocode/json?&address=" + address
r = requests . get ( url )
r . raise_for_status ( )
results = r . json ( ) [ 'results' ]
lat = results [ 0 ] [ 'geometry' ] [ 'location' ] [ 'lat' ]
lng = results [ 0 ] [ 'geometry' ] [ 'location' ] [ 'lng' ]
return lat , lng


signature:
def search_address ( self , address , filters = None , startDate = None , endDate = None , types = None ) :
    '''Perform a catalog search over an address string
 Args :
 address : any address string
 filters : Array of filters . Optional . Example :
 " ( sensorPlatformName = ' WORLDVIEW01 ' OR sensorPlatformName = ' QUICKBIRD02 ' ) " ,
 " cloudCover < 10 " ,
 " offNadirAngle < 10"
 startDate : string . Optional . Example : " 2004-01-01T00:00:00.000Z "
 endDate : string . Optional . Example : " 2004-01-01T00:00:00.000Z "
 types : Array of types to search for . Optional . Example ( and default ) : [ " Acquisition " ]
 Returns :
 catalog search resultset'''

implementation:
lat , lng = self . get_address_coords ( address )
return self . search_point ( lat , lng , filters = filters , startDate = startDate , endDate = endDate , types = types )


signature:
def search_point ( self , lat , lng , filters = None , startDate = None , endDate = None , types = None , type = None ) :
    '''Perform a catalog search over a specific point , specified by lat , lng
 Args :
 lat : latitude
 lng : longitude
 filters : Array of filters . Optional . Example :
 " ( sensorPlatformName = ' WORLDVIEW01 ' OR sensorPlatformName = ' QUICKBIRD02 ' ) " ,
 " cloudCover < 10 " ,
 " offNadirAngle < 10"
 startDate : string . Optional . Example : " 2004-01-01T00:00:00.000Z "
 endDate : string . Optional . Example : " 2004-01-01T00:00:00.000Z "
 types : Array of types to search for . Optional . Example ( and default ) : [ " Acquisition " ]
 Returns :
 catalog search resultset'''

implementation:
searchAreaWkt = "POLYGON ((%s %s, %s %s, %s %s, %s %s, %s %s))" % ( lng , lat , lng , lat , lng , lat , lng , lat , lng , lat )
return self . search ( searchAreaWkt = searchAreaWkt , filters = filters , startDate = startDate , endDate = endDate , types = types )


signature:
def get_data_location ( self , catalog_id ) :
    """Find and return the S3 data location given a catalog _ id .
 Args :
 catalog _ id : The catalog ID
 Returns :
 A string containing the s3 location of the data associated with a catalog ID . Returns
 None if the catalog ID is not found , or if there is no data yet associated with it ."""

implementation:
try :
    record = self . get ( catalog_id )
except :
    return None
# Handle Landsat8
if 'Landsat8' in record [ 'type' ] and 'LandsatAcquisition' in record [ 'type' ] :
    bucket = record [ 'properties' ] [ 'bucketName' ]
    prefix = record [ 'properties' ] [ 'bucketPrefix' ]
    return 's3://' + bucket + '/' + prefix
# Handle DG Acquisition
if 'DigitalGlobeAcquisition' in record [ 'type' ] :
    o = Ordering ( )
    res = o . location ( [ catalog_id ] )
    return res [ 'acquisitions' ] [ 0 ] [ 'location' ]
return None


signature:
def search ( self , searchAreaWkt = None , filters = None , startDate = None , endDate = None , types = None ) :
    '''Perform a catalog search
 Args :
 searchAreaWkt : WKT Polygon of area to search . Optional .
 filters : Array of filters . Optional . Example :
 " ( sensorPlatformName = ' WORLDVIEW01 ' OR sensorPlatformName = ' QUICKBIRD02 ' ) " ,
 " cloudCover < 10 " ,
 " offNadirAngle < 10"
 startDate : string . Optional . Example : " 2004-01-01T00:00:00.000Z "
 endDate : string . Optional . Example : " 2004-01-01T00:00:00.000Z "
 types : Array of types to search for . Optional . Example ( and default ) : [ " Acquisition " ]
 Returns :
 catalog search resultset'''

implementation:
# Default to search for Acquisition type objects .
if not types :
    types = [ 'Acquisition' ]
# validation : we must have either a WKT or one - week of time window
if startDate :
    startDateTime = datetime . datetime . strptime ( startDate , '%Y-%m-%dT%H:%M:%S.%fZ' )
if endDate :
    endDateTime = datetime . datetime . strptime ( endDate , '%Y-%m-%dT%H:%M:%S.%fZ' )
if startDate and endDate :
    diff = endDateTime - startDateTime
    if diff . days < 0 :
        raise Exception ( "startDate must come before endDate." )
postdata = { "searchAreaWkt" : searchAreaWkt , "types" : types , "startDate" : startDate , "endDate" : endDate , }
if filters :
    postdata [ 'filters' ] = filters
if searchAreaWkt :
    postdata [ 'searchAreaWkt' ] = searchAreaWkt
url = '%(base_url)s/search' % { 'base_url' : self . base_url }
headers = { 'Content-Type' : 'application/json' }
r = self . gbdx_connection . post ( url , headers = headers , data = json . dumps ( postdata ) )
r . raise_for_status ( )
results = r . json ( ) [ 'results' ]
return results


signature:
def get_most_recent_images ( self , results , types = [ ] , sensors = [ ] , N = 1 ) :
    '''Return the most recent image
 Args :
 results : a catalog resultset , as returned from a search
 types : array of types you want . optional .
 sensors : array of sensornames . optional .
 N : number of recent images to return . defaults to 1.
 Returns :
 single catalog item , or none if not found'''

implementation:
if not len ( results ) :
    return None
# filter on type
if types :
    results = [ r for r in results if r [ 'type' ] in types ]
# filter on sensor
if sensors :
    results = [ r for r in results if r [ 'properties' ] . get ( 'sensorPlatformName' ) in sensors ]
# sort by date :
# sorted ( results , key = results . _ _ getitem _ _ ( ' properties ' ) . get ( ' timestamp ' ) )
newlist = sorted ( results , key = lambda k : k [ 'properties' ] . get ( 'timestamp' ) , reverse = True )
return newlist [ : N ]


signature:
def get_bytes_from_blob ( val ) -> bytes :
    """不同数据库从blob拿出的数据有所差别 , 有的是memoryview有的是bytes"""

implementation:
if isinstance ( val , bytes ) :
    return val
elif isinstance ( val , memoryview ) :
    return val . tobytes ( )
else :
    raise TypeError ( 'invalid type for get bytes' )


signature:
def pagination_calc ( items_count , page_size , cur_page = 1 , nearby = 2 ) :
    """: param nearby :
 : param items _ count : count of all items
 : param page _ size : size of one page
 : param cur _ page : current page number , accept string digit
 : return : num of pages , an iterator"""

implementation:
if type ( cur_page ) == str : # noinspection PyUnresolvedReferences
    cur_page = int ( cur_page ) if cur_page . isdigit ( ) else 1
elif type ( cur_page ) == int :
    if cur_page <= 0 :
        cur_page = 1
else :
    cur_page = 1
page_count = 1 if page_size == - 1 else int ( math . ceil ( items_count / page_size ) )
items_length = nearby * 2 + 1
# if first page in page items , first _ page is None ,
# it means the " go to first page " button should not be available .
first_page = None
last_page = None
prev_page = cur_page - 1 if cur_page != 1 else None
next_page = cur_page + 1 if cur_page != page_count else None
if page_count <= items_length :
    items = range ( 1 , page_count + 1 )
elif cur_page <= nearby : # start of items
    items = range ( 1 , items_length + 1 )
    last_page = True
elif cur_page >= page_count - nearby : # end of items
    items = range ( page_count - items_length + 1 , page_count + 1 )
    first_page = True
else :
    items = range ( cur_page - nearby , cur_page + nearby + 1 )
    first_page , last_page = True , True
if first_page :
    first_page = 1
if last_page :
    last_page = page_count
return { 'cur_page' : cur_page , 'prev_page' : prev_page , 'next_page' : next_page , 'first_page' : first_page , 'last_page' : last_page , 'page_numbers' : list ( items ) , 'info' : { 'page_size' : page_size , 'page_count' : page_count , 'items_count' : items_count , } }


signature:
def add_record_check ( self , actions , table , func ) : # emitted after query
# table : ' table _ name '
# column : ( ' table _ name ' , ' column _ name ' )
    assert isinstance ( table , str ) , '`table` must be table name'
    for i in actions :
        assert i not in ( A . QUERY , A . CREATE ) , "meaningless action check with record: [%s]" % i
    self . record_checks . append ( [ table , actions , func ] )
    """def func ( ability , user , action , record : DataRecord , available _ columns : list ) :
 pass"""


signature:
def _parse_permission ( self , obj ) :
    """从 obj 中取出权限
 : param obj :
 : return : [ A . QUERY , A . WRITE , . . . ]"""

implementation:
if isinstance ( obj , str ) :
    if obj == '*' :
        return A . ALL
    elif obj in A . ALL :
        return obj ,
    else :
        logger . warning ( 'Invalid permission action: %s' , obj )
elif isinstance ( obj , ( list , tuple ) ) :
    for i in obj :
        if i not in A . ALL :
            logger . warning ( 'Invalid permission action: %s' , i )
    return obj
elif isinstance ( obj , dict ) :
    return self . _parse_permission ( obj . get ( '*' ) )


signature:
def can_with_columns ( self , user , action , table , columns ) :
    """根据权限进行列过滤
 注意一点 , 只要有一个条件能够通过权限检测 , 那么过滤后还会有剩余条件 , 最终就不会报错 。
 如果全部条件都不能过检测 , 就会爆出权限错误了 。
 : param user :
 : param action : 行为
 : param table : 表名
 : param columns : 列名列表
 : return : 可用列的列表"""

implementation:
# TODO : 此过程可以加缓存
# 全局
global_data = self . rules . get ( '*' )
global_actions = self . _parse_permission ( global_data )
if global_actions and action in global_actions :
    available = list ( columns )
else :
    available = [ ]
# table
table_data = self . rules . get ( table )
table_actions = self . _parse_permission ( table_data )
if table_actions and action in table_actions :
    available = list ( columns )
# column
if type ( table_data ) == dict : # 这意味着有详细的列权限设定 , 不然类型是 list
    for column in columns :
        column_actions = self . _parse_permission ( table_data . get ( column ) )
        if column_actions is not None :
            if action in column_actions : # 有权限 , 试图加入列表
                if column not in available :
                    available . append ( column )
            else : # 无权限 , 从列表剔除
                if column in available :
                    available . remove ( column )
for check in self . common_checks :
    if check [ 0 ] == table and action in check [ 1 ] :
        ret = check [ - 1 ] ( self , user , action , available )
        if isinstance ( ret , ( tuple , set , list ) ) : # 返回列表则进行值覆盖
            available = list ( ret )
        elif ret == '*' : # 返回 * 加上所有可用列
            available = list ( columns )
        elif ret is False : # 返回 false 清空
            available = [ ]
        if not available :
            break
return available


signature:
def can_with_record ( self , user , action , record : DataRecord , * , available = None ) :
    """进行基于 Record 的权限判定 , 返回可用列 。
 : param user :
 : param action :
 : param record :
 : param available : 限定检查范围
 : return : 可用列"""

implementation:
assert action not in ( A . QUERY , A . CREATE ) , "meaningless action check with record: [%s]" % action
# 先行匹配规则适用范围
rules = [ ]
for rule in self . record_checks :
    if record . table == rule [ 0 ] and action in rule [ 1 ] :
        rules . append ( rule )
# 逐个过检查
if available is None :
    available = self . can_with_columns ( user , action , record . table , record . keys ( ) )
else :
    available = list ( available )
bak = available . copy ( )
for rule in rules :
    ret = rule [ - 1 ] ( self , user , action , record , available )
    if isinstance ( ret , ( tuple , set , list ) ) :
        available = list ( ret )
    elif ret == '*' :
        available = list ( bak )
    elif not ret :
        available = [ ]
return available


signature:
def use ( cls , name , method : [ str , Set , List ] , url = None ) :
    """interface helper function"""

implementation:
if not isinstance ( method , ( str , list , set , tuple ) ) :
    raise BaseException ( 'Invalid type of method: %s' % type ( method ) . __name__ )
if isinstance ( method , str ) :
    method = { method }
# TODO : check methods available
cls . _interface [ name ] = [ { 'method' : method , 'url' : url } ]


signature:
async def get_ip ( self ) -> Union [ IPv4Address , IPv6Address ] :
    """get ip address of client
 : return :"""

implementation:
xff = await self . get_x_forwarded_for ( )
if xff :
    return xff [ 0 ]
ip_addr = self . _request . transport . get_extra_info ( 'peername' ) [ 0 ]
return ip_address ( ip_addr )


signature:
def finish ( self , code , data = NotImplemented ) :
    """Set response as { ' code ' : xxx , ' data ' : xxx }
 : param code :
 : param data :
 : return :"""

implementation:
if data is NotImplemented :
    data = RETCODE . txt_cn . get ( code , None )
self . ret_val = { 'code' : code , 'data' : data }
# for access in inhreads method
self . response = web . json_response ( self . ret_val , dumps = json_ex_dumps )
logger . debug ( 'finish: %s' % self . ret_val )
self . _finish_end ( )


signature:
def finish_raw ( self , body : bytes , status : int = 200 , content_type : Optional [ str ] = None ) :
    """Set raw response
 : param body :
 : param status :
 : param content _ type :
 : return :"""

implementation:
self . ret_val = body
self . response = web . Response ( body = body , status = status , content_type = content_type )
logger . debug ( 'finish: raw body(%d bytes)' % len ( body ) )
self . _finish_end ( )


signature:
def add_soft_foreign_key ( cls , column , table_name , alias = None ) :
    """the column stores foreign table ' s primary key but isn ' t a foreign key ( to avoid constraint )
 warning : if the table not exists , will crash when query with loadfk
 : param column : table ' s column
 : param table _ name : foreign table name
 : param alias : table name ' s alias . Default is as same as table name .
 : return : True , None"""

implementation:
if column in cls . fields :
    table = SQLForeignKey ( table_name , column , cls . fields [ column ] , True )
    if alias :
        if alias in cls . foreign_keys_table_alias :
            logger . warning ( "This alias of table is already exists, overwriting: %s.%s to %s" % ( cls . __name__ , column , table_name ) )
        cls . foreign_keys_table_alias [ alias ] = table
    if column not in cls . foreign_keys :
        cls . foreign_keys [ column ] = [ table ]
    else :
        if not alias :
            logger . warning ( "The soft foreign key will not work, an alias required: %s.%s to %r" % ( cls . __name__ , column , table_name ) )
        cls . foreign_keys [ column ] . append ( table )
    return True


signature:
def current_request_role ( self ) -> [ int , str ] :
    """Current role requested by client .
 : return :"""

implementation:
role_val = self . headers . get ( 'Role' )
return int ( role_val ) if role_val and role_val . isdigit ( ) else role_val


signature:
async def load_fk ( self , info : SQLQueryInfo , records : Iterable [ DataRecord ] ) -> Union [ List , Iterable ] :
    """: param info :
 : param records : the data got from database and filtered from permission
 : return :"""

implementation:
# if not items , items is probably [ ] , so return itself .
# if not items : return items
# 1 . get tables ' instances
# table _ map = { }
# for column in info [ ' loadfk ' ] . keys ( ) :
# tbl _ name = self . foreign _ keys [ column ] [ 0]
# table _ map [ column ] = self . app . tables [ tbl _ name ]
# 2 . get query parameters
async def check ( data , records ) :
    for column , fkvalues_lst in data . items ( ) :
        for fkvalues in fkvalues_lst :
            pks = [ ]
            all_ni = True
            vcls = self . app . tables [ fkvalues [ 'table' ] ]
            for i in records :
                val = i . get ( column , NotImplemented )
                if val != NotImplemented :
                    all_ni = False
                pks . append ( val )
            if all_ni :
                logger . debug ( "load foreign key failed, do you have read permission to the column %r?" % column )
                continue
            # 3 . query foreign keys
            v = vcls ( self . app , self . _request )
            # fake view
            await v . _prepare ( )
            info2 = SQLQueryInfo ( )
            info2 . set_select ( ALL_COLUMNS )
            info2 . add_condition ( PRIMARY_KEY , SQL_OP . IN , pks )
            info2 . bind ( v )
            # ability = vcls . permission . request _ role ( self . current _ user , fkvalues [ ' role ' ] )
            # info2 . check _ query _ permission _ full ( self . current _ user , fktable , ability )
            try :
                fk_records , count = await v . _sql . select_page ( info2 , size = - 1 )
            except RecordNotFound : # 外键没有找到值 , 也许全部都是null , 这很常见
                continue
            # if not fk _ records : continue
            await v . check_records_permission ( info2 , fk_records )
            fk_dict = { }
            for i in fk_records : # 主键 : 数据
                fk_dict [ i [ vcls . primary_key ] ] = i
            column_to_set = fkvalues . get ( 'as' , column ) or column
            for _ , record in enumerate ( records ) :
                k = record . get ( column , NotImplemented )
                if k in fk_dict :
                    record [ column_to_set ] = fk_dict [ k ]
            if fkvalues [ 'loadfk' ] :
                await check ( fkvalues [ 'loadfk' ] , fk_records )
await check ( info . loadfk , records )
return records


signature:
async def _call_handle ( self , func , * args ) :
    """call and check result of handle _ query / read / insert / update"""

implementation:
await async_call ( func , * args )
if self . is_finished :
    raise FinishQuitException ( )


signature:
async def after_update ( self , raw_post : Dict , values : SQLValuesToWrite , old_records : List [ DataRecord ] , records : List [ DataRecord ] ) :
    """: param old _ records :
 : param raw _ post :
 : param values :
 : param records :
 : return :"""


signature:
def roles ( self ) :
    """BaseUser . roles 的实现 , 返回用户可用角色
 : return :"""

implementation:
ret = { None }
if self . state == POST_STATE . DEL :
    return ret
ret . add ( 'user' )
return ret


signature:
def gen_password_and_salt ( cls , password_text ) :
    """生成加密后的密码和盐"""

implementation:
salt = os . urandom ( 32 )
dk = hashlib . pbkdf2_hmac ( config . PASSWORD_HASH_FUNC_NAME , password_text . encode ( 'utf-8' ) , salt , config . PASSWORD_HASH_ITERATIONS , )
return { 'password' : dk , 'salt' : salt }


signature:
def gen_token ( cls ) :
    """生成 access _ token"""

implementation:
token = os . urandom ( 16 )
token_time = int ( time . time ( ) )
return { 'token' : token , 'token_time' : token_time }


signature:
def set_password ( self , new_password ) :
    """设置密码"""

implementation:
info = self . gen_password_and_salt ( new_password )
self . password = info [ 'password' ]
self . salt = info [ 'salt' ]
self . save ( )


signature:
def _auth_base ( self , password_text ) :
    """已获取了用户对象 , 进行密码校验
 : param password _ text :
 : return :"""

implementation:
dk = hashlib . pbkdf2_hmac ( config . PASSWORD_HASH_FUNC_NAME , password_text . encode ( 'utf-8' ) , get_bytes_from_blob ( self . salt ) , config . PASSWORD_HASH_ITERATIONS )
if self . password == dk :
    return self


signature:
async def get_session ( cls , view ) :
    """Every request have a session instance
 : param view :
 : return :"""

implementation:
session = cls ( view )
session . key = await session . get_key ( )
session . _data = await session . load ( ) or { }
return session


signature:
async def select_page ( self , info : SQLQueryInfo , size = 1 , page = 1 ) -> Tuple [ Tuple [ DataRecord , ... ] , int ] :
    """Select from database
 : param info :
 : param size : - 1 means infinite
 : param page :
 : param need _ count : if True , get count as second return value , otherwise - 1
 : return : records . count"""

implementation: raise NotImplementedError ( )


signature:
async def update ( self , records : Iterable [ DataRecord ] , values : SQLValuesToWrite , returning = False ) -> Union [ int , Iterable [ DataRecord ] ] :
    """: param records :
 : param values :
 : param returning :
 : return : return count if returning is False , otherwise records"""


signature:
async def insert ( self , values_lst : Iterable [ SQLValuesToWrite ] , returning = False ) -> Union [ int , List [ DataRecord ] ] :
    """: param values _ lst :
 : param returning :
 : return : return count if returning is False , otherwise records"""


signature:
def parse_order ( text ) :
    """: param text : order = id . desc , xxx . asc
 : return : [
 [ < column > , asc | desc | default ] ,
 [ < column2 > , asc | desc | default ] ,"""

implementation:
orders = [ ]
for i in map ( str . strip , text . split ( ',' ) ) :
    items = i . split ( '.' , 2 )
    if len ( items ) == 1 :
        column , order = items [ 0 ] , 'default'
    elif len ( items ) == 2 :
        column , order = items
    else :
        raise InvalidParams ( "Invalid order syntax" )
    order = order . lower ( )
    if order not in ( 'asc' , 'desc' , 'default' ) :
        raise InvalidParams ( 'Invalid order mode: %s' % order )
    if order != 'default' :
        orders . append ( SQLQueryOrder ( column , order ) )
return orders


signature:
def parse_select ( cls , text : str ) -> Set :
    """get columns from select text
 : param text : col1 , col2
 : return : ALL _ COLUMNS or [ ' col1 ' , ' col2 ' ]"""

implementation:
if text == '*' :
    return ALL_COLUMNS
    # None means ALL
selected_columns = set ( filter ( lambda x : x , map ( str . strip , text . split ( ',' ) ) ) )
if not selected_columns :
    raise InvalidParams ( "No column(s) selected" )
return selected_columns


signature:
def parse_load_fk ( cls , data : Dict [ str , List [ Dict [ str , object ] ] ] ) -> Dict [ str , List [ Dict [ str , object ] ] ] :
    """: param data : {
 < column > : role ,
 < column2 > : role ,
 < column > : {
 ' role ' : role ,
 ' loadfk ' : { . . . } ,
 : return : {
 < column > : {
 ' role ' : role ,
 < column3 > : {
 ' role ' : role ,
 ' loadfk ' : { . . . } ,"""

implementation:
default_value_dict = { 'role' : None , 'as' : None , 'table' : None , 'loadfk' : None }
def value_normalize_dict ( value ) :
    def check ( k , v ) :
        if k == 'role' :
            return isinstance ( v , str )
        if k == 'as' :
            return isinstance ( v , str )
        if k == 'table' :
            return isinstance ( v , str )
        if k == 'loadfk' :
            return isinstance ( v , dict )
    valid = { k : v for k , v in value . items ( ) if check ( k , v ) }
    if not valid :
        return default_value_dict . copy ( )
    if 'loadfk' in valid and valid [ 'loadfk' ] :
        valid [ 'loadfk' ] = cls . parse_load_fk ( valid [ 'loadfk' ] )
    for k , v in default_value_dict . items ( ) :
        valid . setdefault ( k , v )
    return valid
def value_normalize ( value , no_list = True ) :
    if value is None :
        return default_value_dict . copy ( )
    elif not no_list and isinstance ( value , List ) : # < column > : [ value1 , value2 , . . . ]
        return list ( map ( value_normalize , value ) )
    elif isinstance ( value , str ) : # < column > : role
        val = default_value_dict . copy ( )
        val [ 'role' ] = value
        return val
    elif isinstance ( value , Dict ) : # { ' role ' : < str > , ' as ' : < str > , . . . }
        return value_normalize_dict ( value )
    else :
        raise InvalidParams ( 'Invalid syntax for "loadfk": %s' % value )
# 对全部项进行检查
new_data = { }
if not isinstance ( data , dict ) :
    raise InvalidParams ( 'Invalid syntax for "loadfk": %s' % data )
for k , v in data . items ( ) :
    nv = value_normalize ( v , False )
    new_data [ k ] = nv if isinstance ( nv , List ) else [ nv ]
return new_data


signature:
def add_condition ( self , field_name , op , value ) :
    """Add a query condition and validate it .
 raise ParamsException if failed .
 self . view required
 : param field _ name :
 : param op :
 : param value :
 : return : None"""

implementation:
if not isinstance ( op , SQL_OP ) :
    if op not in SQL_OP . txt2op :
        raise SQLOperatorInvalid ( op )
    else :
        op = SQL_OP . txt2op . get ( op )
self . conditions . append ( [ field_name , op , value ] )


signature:
def _packb2 ( obj , ** options ) :
    """Serialize a Python object into MessagePack bytes .
 Args :
 obj : a Python object
 Kwargs :
 ext _ handlers ( dict ) : dictionary of Ext handlers , mapping a custom type
 to a callable that packs an instance of the type
 into an Ext object
 force _ float _ precision ( str ) : " single " to force packing floats as
 IEEE - 754 single - precision floats ,
 " double " to force packing floats as
 IEEE - 754 double - precision floats .
 Returns :
 A ' str ' containing serialized MessagePack bytes .
 Raises :
 UnsupportedType ( PackException ) :
 Object type not supported for packing .
 Example :
 > > > umsgpack . packb ( { u " compact " : True , u " schema " : 0 } )
 ' \x82 \xa7 compact \xc3 \xa6 schema \x00 '"""

implementation:
fp = io . BytesIO ( )
_pack2 ( obj , fp , ** options )
return fp . getvalue ( )


signature:
def _packb3 ( obj , ** options ) :
    """Serialize a Python object into MessagePack bytes .
 Args :
 obj : a Python object
 Kwargs :
 ext _ handlers ( dict ) : dictionary of Ext handlers , mapping a custom type
 to a callable that packs an instance of the type
 into an Ext object
 force _ float _ precision ( str ) : " single " to force packing floats as
 IEEE - 754 single - precision floats ,
 " double " to force packing floats as
 IEEE - 754 double - precision floats .
 Returns :
 A ' bytes ' containing serialized MessagePack bytes .
 Raises :
 UnsupportedType ( PackException ) :
 Object type not supported for packing .
 Example :
 > > > umsgpack . packb ( { u " compact " : True , u " schema " : 0 } )
 b ' \x82 \xa7 compact \xc3 \xa6 schema \x00 '"""

implementation:
fp = io . BytesIO ( )
_pack3 ( obj , fp , ** options )
return fp . getvalue ( )


signature:
def _unpackb2 ( s , ** options ) :
    """Deserialize MessagePack bytes into a Python object .
 Args :
 s : a ' str ' or ' bytearray ' containing serialized MessagePack bytes
 Kwargs :
 ext _ handlers ( dict ) : dictionary of Ext handlers , mapping integer Ext
 type to a callable that unpacks an instance of
 Ext into an object
 use _ ordered _ dict ( bool ) : unpack maps into OrderedDict , instead of
 unordered dict ( default False )
 allow _ invalid _ utf8 ( bool ) : unpack invalid strings into instances of
 InvalidString , for access to the bytes
 ( default False )
 Returns :
 A Python object .
 Raises :
 TypeError :
 Packed data type is neither ' str ' nor ' bytearray ' .
 InsufficientDataException ( UnpackException ) :
 Insufficient data to unpack the serialized object .
 InvalidStringException ( UnpackException ) :
 Invalid UTF - 8 string encountered during unpacking .
 UnsupportedTimestampException ( UnpackException ) :
 Unsupported timestamp format encountered during unpacking .
 ReservedCodeException ( UnpackException ) :
 Reserved code encountered during unpacking .
 UnhashableKeyException ( UnpackException ) :
 Unhashable key encountered during map unpacking .
 The serialized map cannot be deserialized into a Python dictionary .
 DuplicateKeyException ( UnpackException ) :
 Duplicate key encountered during map unpacking .
 Example :
 > > > umsgpack . unpackb ( b ' \x82 \xa7 compact \xc3 \xa6 schema \x00 ' )
 { u ' compact ' : True , u ' schema ' : 0}"""

implementation:
if not isinstance ( s , ( str , bytearray ) ) :
    raise TypeError ( "packed data must be type 'str' or 'bytearray'" )
return _unpack ( io . BytesIO ( s ) , options )


signature:
def _unpackb3 ( s , ** options ) :
    """Deserialize MessagePack bytes into a Python object .
 Args :
 s : a ' bytes ' or ' bytearray ' containing serialized MessagePack bytes
 Kwargs :
 ext _ handlers ( dict ) : dictionary of Ext handlers , mapping integer Ext
 type to a callable that unpacks an instance of
 Ext into an object
 use _ ordered _ dict ( bool ) : unpack maps into OrderedDict , instead of
 unordered dict ( default False )
 allow _ invalid _ utf8 ( bool ) : unpack invalid strings into instances of
 InvalidString , for access to the bytes
 ( default False )
 Returns :
 A Python object .
 Raises :
 TypeError :
 Packed data type is neither ' bytes ' nor ' bytearray ' .
 InsufficientDataException ( UnpackException ) :
 Insufficient data to unpack the serialized object .
 InvalidStringException ( UnpackException ) :
 Invalid UTF - 8 string encountered during unpacking .
 UnsupportedTimestampException ( UnpackException ) :
 Unsupported timestamp format encountered during unpacking .
 ReservedCodeException ( UnpackException ) :
 Reserved code encountered during unpacking .
 UnhashableKeyException ( UnpackException ) :
 Unhashable key encountered during map unpacking .
 The serialized map cannot be deserialized into a Python dictionary .
 DuplicateKeyException ( UnpackException ) :
 Duplicate key encountered during map unpacking .
 Example :
 > > > umsgpack . unpackb ( b ' \x82 \xa7 compact \xc3 \xa6 schema \x00 ' )
 { ' compact ' : True , ' schema ' : 0}"""

implementation:
if not isinstance ( s , ( bytes , bytearray ) ) :
    raise TypeError ( "packed data must be type 'bytes' or 'bytearray'" )
return _unpack ( io . BytesIO ( s ) , options )


signature:
def view_bind ( app , cls_url , view_cls : Type [ 'BaseView' ] ) :
    """将 API 绑定到 web 服务上
 : param view _ cls :
 : param app :
 : param cls _ url :
 : return :"""

implementation:
if view_cls . _no_route :
    return
cls_url = cls_url or view_cls . __class__ . __name__ . lower ( )
def add_route ( name , route_info , beacon_info ) :
    for method in route_info [ 'method' ] :
        async def beacon ( request ) :
            pass
        route_key = route_info [ 'url' ] if route_info [ 'url' ] else name
        app . _raw_app . router . add_route ( method , urljoin ( '/api' , cls_url , route_key ) , beacon )
        app . route . _beacons [ beacon ] = beacon_info
# noinspection PyProtectedMember
for name , route_info_lst in view_cls . _interface . items ( ) :
    for route_info in route_info_lst :
        real_handler = getattr ( view_cls , name , None )
        if real_handler is None :
            continue
            # TODO : delete
        assert real_handler is not None , "handler must be exists"
        handler_name = '%s.%s' % ( view_cls . __name__ , real_handler . __name__ )
        assert iscoroutinefunction ( real_handler ) , "Add 'async' before interface function %r" % handler_name
        beacon_info = { 'view' : view_cls , 'name' : name , 'handler' : real_handler , 'route_info' : route_info }
        add_route ( name , route_info , beacon_info )


signature:
def add_static ( self , prefix , path , ** kwargs ) :
    """: param prefix : URL prefix
 : param path : file directory
 : param kwargs :
 : return :"""

implementation: self . statics . append ( ( prefix , path , kwargs ) , )


implementation:
data = json . loads ( data )
for i in ( 'tables' , 'columns' , 'conditions' ) :
    if i not in data :
        raise QueryException ( "query: %s not found" % i )
tables = data [ 'tables' ]
columns = data [ 'columns' ]
conditions = data [ 'conditions' ]
def parse_stmt ( s , expr_cls , all_op , multi_items_op ) :
    if len ( s ) == 0 :
        return [ ]
    if s [ 0 ] in all_op :
        if s [ 0 ] in multi_items_op :
            values = [ ]
            for i in s [ 1 : ] :
                values . append ( parse_stmt ( i , expr_cls , all_op , multi_items_op ) )
            return expr_cls ( None , s [ 0 ] , None , values = values )
        else :
            if len ( s ) == 5 : # t1 . c1 = = t2 . c2
                lhs = Column ( s [ 2 ] , table = s [ 1 ] )
                rhs = Column ( s [ 4 ] , table = s [ 3 ] )
                if ( s [ 1 ] not in tables ) or ( s [ 3 ] not in tables ) :
                    raise QueryException ( 'Bad query' )
                return expr_cls ( lhs , s [ 0 ] , rhs )
            else : # t1 . c1 = = val
                lhs = Column ( s [ 2 ] , table = s [ 1 ] )
                if s [ 1 ] not in tables :
                    raise QueryException ( 'Bad query' )
                return expr_cls ( lhs , s [ 0 ] , s [ 3 ] )
    else :
        raise QueryException ( 'Bad query' )
query_op = ( '+' , '-' , '*' , '/' )
query_columns = [ ]
for i in columns :
    if len ( i ) == 2 :
        query_columns . append ( Column ( i [ 1 ] , table = i [ 0 ] ) )
    else :
        query_columns . append ( parse_stmt ( i , QueryExpression , query_op , query_op ) )
wheres = parse_stmt ( conditions , ConditionExpression , _operator_map , ( 'and' , 'or' , ) )
return { 'tables' : tables , 'columns' : query_columns , 'wheres' : wheres , }


signature:
def validate ( method ) :
    """Config option name value validator decorator ."""

implementation:
# Name error template
name_error = 'configuration option "{}" is not supported'
@ functools . wraps ( method )
def validator ( self , name , * args ) :
    if name not in self . allowed_opts :
        raise ValueError ( name_error . format ( name ) )
    return method ( self , name , * args )
return validator


signature:
def run ( self , ctx ) :
    """Runs the current phase ."""

implementation:
# Reverse engine assertion if needed
if ctx . reverse :
    self . engine . reverse ( )
if self . engine . empty :
    raise AssertionError ( 'grappa: no assertions to run' )
try : # Run assertion in series and return error , if present
    return self . run_assertions ( ctx )
except Exception as _err : # Handle legit grappa internval errors
    if getattr ( _err , '__legit__' , False ) :
        raise _err
    # Otherwise render it
    return self . render_error ( ctx , _err )


signature:
def observe ( matcher ) :
    """Internal decorator to trigger operator hooks before / after
 matcher execution ."""

implementation:
@ functools . wraps ( matcher )
def observer ( self , subject , * expected , ** kw ) : # Trigger before hook , if present
    if hasattr ( self , 'before' ) :
        self . before ( subject , * expected , ** kw )
    # Trigger matcher method
    result = matcher ( self , subject , * expected , ** kw )
    # After error hook
    if result is not True and hasattr ( self , 'after_error' ) :
        self . after_error ( result , subject , * expected , ** kw )
    # After success hook
    if result is True and hasattr ( self , 'after_success' ) :
        self . after_success ( subject , * expected , ** kw )
    # Enable diff comparison on error , if needed
    if not hasattr ( self , 'show_diff' ) :
        self . show_diff = all ( [ isinstance ( subject , six . string_types ) , all ( [ isinstance ( x , six . string_types ) for x in expected ] ) , ] )
    return result
return observer


signature:
def run_matcher ( self , subject , * expected , ** kw ) :
    """Runs the operator matcher test function ."""

implementation:
# Update assertion expectation
self . expected = expected
_args = ( subject , )
if self . kind == OperatorTypes . MATCHER :
    _args += expected
try :
    result = self . match ( * _args , ** kw )
except Exception as error :
    return self . _make_error ( error = error )
reasons = [ ]
if isinstance ( result , tuple ) :
    result , reasons = result
if result is False and self . ctx . negate :
    return True
if result is True and not self . ctx . negate :
    return True
return self . _make_error ( reasons = reasons )


signature:
def run ( self , * args , ** kw ) :
    """Runs the current operator with the subject arguments to test .
 This method is implemented by matchers only ."""

implementation:
log . debug ( '[operator] run "{}" with arguments: {}' . format ( self . __class__ . __name__ , args ) )
if self . kind == OperatorTypes . ATTRIBUTE :
    return self . match ( self . ctx )
else :
    return self . run_matcher ( * args , ** kw )


signature:
def operator ( name = None , operators = None , aliases = None , kind = None ) :
    """Registers a new operator function in the test engine .
 Arguments :
 * args : variadic arguments .
 * * kw : variadic keyword arguments .
 Returns :
 function"""

implementation:
def delegator ( assertion , subject , expected , * args , ** kw ) :
    return assertion . test ( subject , expected , * args , ** kw )
def decorator ( fn ) :
    operator = Operator ( fn = fn , aliases = aliases , kind = kind )
    _name = name if isinstance ( name , six . string_types ) else fn . __name__
    operator . operators = ( _name , )
    _operators = operators
    if isinstance ( _operators , list ) :
        _operators = tuple ( _operators )
    if isinstance ( _operators , tuple ) :
        operator . operators += _operators
    # Register operator
    Engine . register ( operator )
    return functools . partial ( delegator , operator )
return decorator ( name ) if inspect . isfunction ( name ) else decorator


signature:
def attribute ( * args , ** kw ) :
    """Registers a new attribute only operator function in the test engine .
 Arguments :
 * args : variadic arguments .
 * * kw : variadic keyword arguments .
 Returns :
 function"""

implementation: return operator ( kind = Operator . Type . ATTRIBUTE , * args , ** kw )


signature:
def use ( plugin ) :
    """Register plugin in grappa .
 ` plugin ` argument can be a function or a object that implement ` register `
 method , which should accept one argument : ` grappa . Engine ` instance .
 Arguments :
 plugin ( function | module ) : grappa plugin object to register .
 Raises :
 ValueError : if ` plugin ` is not a valid interface .
 Example : :
 import grappa
 class MyOperator ( grappa . Operator ) :
 pass
 def my _ plugin ( engine ) :
 engine . register ( MyOperator )
 grappa . use ( my _ plugin )"""

implementation:
log . debug ( 'register new plugin: {}' . format ( plugin ) )
if inspect . isfunction ( plugin ) :
    return plugin ( Engine )
if plugin and hasattr ( plugin , 'register' ) :
    return plugin . register ( Engine )
raise ValueError ( 'invalid plugin: must be a function or ' 'implement register() method' )


signature:
def load ( ) :
    """Loads the built - in operators into the global test engine ."""

implementation:
for operator in operators :
    module , symbols = operator [ 0 ] , operator [ 1 : ]
    path = 'grappa.operators.{}' . format ( module )
    # Dynamically import modules
    operator = __import__ ( path , None , None , symbols )
    # Register operators in the test engine
    for symbol in symbols :
        Engine . register ( getattr ( operator , symbol ) )


signature:
def register_operators ( * operators ) :
    """Registers one or multiple operators in the test engine ."""

implementation:
def validate ( operator ) :
    if isoperator ( operator ) :
        return True
    raise NotImplementedError ( 'invalid operator: {}' . format ( operator ) )
def register ( operator ) : # Register operator by DSL keywords
    for name in operator . operators : # Check valid operators
        if name in Engine . operators :
            raise ValueError ( 'operator name "{}" from {} is already ' 'in use by other operator' . format ( name , operator . __name__ ) )
        # Register operator by name
        Engine . operators [ name ] = operator
# Validates and registers operators
[ register ( operator ) for operator in operators if validate ( operator ) ]


signature:
def find_address_file ( self ) :
    """Finds the OMXPlayer DBus connection
 Assumes there is an alive OMXPlayer process .
 : return :"""

implementation:
possible_address_files = [ ]
while not possible_address_files : # filter is used here as glob doesn ' t support regexp : (
    isnt_pid_file = lambda path : not path . endswith ( '.pid' )
    possible_address_files = list ( filter ( isnt_pid_file , glob ( '/tmp/omxplayerdbus.*' ) ) )
    possible_address_files . sort ( key = lambda path : os . path . getmtime ( path ) )
    time . sleep ( 0.05 )
self . path = possible_address_files [ - 1 ]


signature:
def load ( self , source , pause = False ) :
    """Loads a new source ( as a file ) from ` ` source ` ` ( a file path or URL )
 by killing the current ` ` omxplayer ` ` process and forking a new one .
 Args :
 source ( string ) : Path to the file to play or URL"""

implementation:
self . _source = source
self . _load_source ( source )
if pause :
    time . sleep ( 0.5 )
    # Wait for the DBus interface to be initialised
    self . pause ( )


signature:
def set_volume ( self , volume ) :
    """Args :
 float : volume in the interval [ 0 , 10]"""

implementation:
# 0 isn ' t handled correctly so we have to set it to a very small value to achieve the same purpose
if volume == 0 :
    volume = 1e-10
return self . _player_interface_property ( 'Volume' , dbus . Double ( volume ) )


signature:
def set_rate ( self , rate ) :
    """Set the playback rate of the video as a multiple of the default playback speed
 Examples :
 > > > player . set _ rate ( 2)
 # Will play twice as fast as normal speed
 > > > player . set _ rate ( 0.5)
 # Will play half speed"""

implementation:
self . _rate = self . _player_interface_property ( 'Rate' , dbus . Double ( rate ) )
return self . _rate


signature:
def pause ( self ) :
    """Pause playback"""

implementation:
self . _player_interface . Pause ( )
self . _is_playing = False
self . pauseEvent ( self )


signature:
def play_pause ( self ) :
    """Pause playback if currently playing , otherwise start playing if currently paused ."""

implementation:
self . _player_interface . PlayPause ( )
self . _is_playing = not self . _is_playing
if self . _is_playing :
    self . playEvent ( self )
else :
    self . pauseEvent ( self )


signature:
def seek ( self , relative_position ) :
    """Seek the video by ` relative _ position ` seconds
 Args :
 relative _ position ( float ) : The position in seconds to seek to ."""

implementation:
self . _player_interface . Seek ( Int64 ( 1000.0 * 1000 * relative_position ) )
self . seekEvent ( self , relative_position )


signature:
def set_position ( self , position ) :
    """Set the video to playback position to ` position ` seconds from the start of the video
 Args :
 position ( float ) : The position in seconds ."""

implementation:
self . _player_interface . SetPosition ( ObjectPath ( "/not/used" ) , Int64 ( position * 1000.0 * 1000 ) )
self . positionEvent ( self , position )


signature:
def set_video_pos ( self , x1 , y1 , x2 , y2 ) :
    """Set the video position on the screen
 Args :
 x1 ( int ) : Top left x coordinate ( px )
 y1 ( int ) : Top left y coordinate ( px )
 x2 ( int ) : Bottom right x coordinate ( px )
 y2 ( int ) : Bottom right y coordinate ( px )"""

implementation:
position = "%s %s %s %s" % ( str ( x1 ) , str ( y1 ) , str ( x2 ) , str ( y2 ) )
self . _player_interface . VideoPos ( ObjectPath ( '/not/used' ) , String ( position ) )


signature:
def video_pos ( self ) :
    """Returns :
 ( int , int , int , int ) : Video spatial position ( x1 , y1 , x2 , y2 ) where ( x1 , y1 ) is top left ,
 and ( x2 , y2 ) is bottom right . All values in px ."""

implementation:
position_string = self . _player_interface . VideoPos ( ObjectPath ( '/not/used' ) )
return list ( map ( int , position_string . split ( " " ) ) )


signature:
def set_video_crop ( self , x1 , y1 , x2 , y2 ) :
    """Args :
 x1 ( int ) : Top left x coordinate ( px )
 y1 ( int ) : Top left y coordinate ( px )
 x2 ( int ) : Bottom right x coordinate ( px )
 y2 ( int ) : Bottom right y coordinate ( px )"""

implementation:
crop = "%s %s %s %s" % ( str ( x1 ) , str ( y1 ) , str ( x2 ) , str ( y2 ) )
self . _player_interface . SetVideoCropPos ( ObjectPath ( '/not/used' ) , String ( crop ) )


signature:
def is_playing ( self ) :
    """Returns :
 bool : Whether the player is playing"""

implementation:
self . _is_playing = ( self . playback_status ( ) == "Playing" )
logger . info ( "Playing?: %s" % self . _is_playing )
return self . _is_playing


signature:
def play_sync ( self ) :
    """Play the video and block whilst the video is playing"""

implementation:
self . play ( )
logger . info ( "Playing synchronously" )
try :
    time . sleep ( 0.05 )
    logger . debug ( "Wait for playing to start" )
    while self . is_playing ( ) :
        time . sleep ( 0.05 )
except DBusException :
    logger . error ( "Cannot play synchronously any longer as DBus calls timed out." )


signature:
def play ( self ) :
    """Play the video asynchronously returning control immediately to the calling code"""

implementation:
if not self . is_playing ( ) :
    self . play_pause ( )
    self . _is_playing = True
    self . playEvent ( self )


signature:
def quit ( self ) :
    """Quit the player , blocking until the process has died"""

implementation:
if self . _process is None :
    logger . debug ( 'Quit was called after self._process had already been released' )
    return
try :
    logger . debug ( 'Quitting OMXPlayer' )
    process_group_id = os . getpgid ( self . _process . pid )
    os . killpg ( process_group_id , signal . SIGTERM )
    logger . debug ( 'SIGTERM Sent to pid: %s' % process_group_id )
    self . _process_monitor . join ( )
except OSError :
    logger . error ( 'Could not find the process to kill' )
self . _process = None


signature:
def render_to_response ( self , context , ** response_kwargs ) :
    """Returns a response with a template depending if the request is ajax
 or not and it renders with the given context ."""

implementation:
if self . request . is_ajax ( ) :
    template = self . page_template
else :
    template = self . get_template_names ( )
return self . response_class ( request = self . request , template = template , context = context , ** response_kwargs )


signature:
def translate_value ( document_field , form_value ) :
    """Given a document _ field and a form _ value this will translate the value
 to the correct result for mongo to use ."""

implementation:
value = form_value
if isinstance ( document_field , ReferenceField ) :
    value = document_field . document_type . objects . get ( id = form_value ) if form_value else None
return value


signature:
def trim_field_key ( document , field_key ) :
    """Returns the smallest delimited version of field _ key that
 is an attribute on document .
 return ( key , left _ over _ array )"""

implementation:
trimming = True
left_over_key_values = [ ]
current_key = field_key
while trimming and current_key :
    if hasattr ( document , current_key ) :
        trimming = False
    else :
        key_array = current_key . split ( "_" )
        left_over_key_values . append ( key_array . pop ( ) )
        current_key = u"_" . join ( key_array )
left_over_key_values . reverse ( )
return current_key , left_over_key_values


signature:
def has_edit_permission ( self , request ) :
    """Can edit this object"""

implementation: return request . user . is_authenticated and request . user . is_active and request . user . is_staff


signature:
def has_add_permission ( self , request ) :
    """Can add this object"""


signature:
def has_delete_permission ( self , request ) :
    """Can delete this object"""

implementation: return request . user . is_authenticated and request . user . is_active and request . user . is_superuser


signature:
def get_form_field_dict ( self , model_dict ) :
    """Takes a model dictionary representation and creates a dictionary
 keyed by form field . Each value is a keyed 4 tuple of :
 ( widget , mode _ field _ instance , model _ field _ type , field _ key )"""

implementation:
return_dict = OrderedDict ( )
# Workaround : mongoengine doesn ' t preserve form fields ordering from metaclass _ _ new _ _
if hasattr ( self . model , 'Meta' ) and hasattr ( self . model . Meta , 'form_fields_ordering' ) :
    field_order_list = tuple ( form_field for form_field in self . model . Meta . form_fields_ordering if form_field in model_dict . iterkeys ( ) )
    order_dict = OrderedDict . fromkeys ( field_order_list )
    return_dict = order_dict
for field_key , field_dict in sorted ( model_dict . items ( ) ) :
    if not field_key . startswith ( "_" ) :
        widget = field_dict . get ( '_widget' , None )
        if widget is None :
            return_dict [ field_key ] = self . get_form_field_dict ( field_dict )
            return_dict [ field_key ] . update ( { '_field_type' : field_dict . get ( '_field_type' , None ) } )
        else :
            return_dict [ field_key ] = FieldTuple ( widget , field_dict . get ( '_document_field' , None ) , field_dict . get ( '_field_type' , None ) , field_dict . get ( '_key' , None ) )
return return_dict


signature:
def set_form_fields ( self , form_field_dict , parent_key = None , field_type = None ) :
    """Set the form fields for every key in the form _ field _ dict .
 Params :
 form _ field _ dict - - a dictionary created by get _ form _ field _ dict
 parent _ key - - the key for the previous key in the recursive call
 field _ type - - used to determine what kind of field we are setting"""

implementation:
for form_key , field_value in form_field_dict . items ( ) :
    form_key = make_key ( parent_key , form_key ) if parent_key is not None else form_key
    if isinstance ( field_value , tuple ) :
        set_list_class = False
        base_key = form_key
        # Style list fields
        if ListField in ( field_value . field_type , field_type ) : # Nested lists / embedded docs need special care to get
        # styles to work out nicely .
            if parent_key is None or ListField == field_value . field_type :
                if field_type != EmbeddedDocumentField :
                    field_value . widget . attrs [ 'class' ] += ' listField {0}' . format ( form_key )
                set_list_class = True
            else :
                field_value . widget . attrs [ 'class' ] += ' listField'
            # Compute number value for list key
            list_keys = [ field_key for field_key in self . form . fields . keys ( ) if has_digit ( field_key ) ]
            key_int = 0
            while form_key in list_keys :
                key_int += 1
            form_key = make_key ( form_key , key_int )
        if parent_key is not None : # Get the base key for our embedded field class
            valid_base_keys = [ model_key for model_key in self . model_map_dict . keys ( ) if not model_key . startswith ( "_" ) ]
            while base_key not in valid_base_keys and base_key :
                base_key = make_key ( base_key , exclude_last_string = True )
            # We need to remove the trailing number from the key
            # so that grouping will occur on the front end when we have a list .
            embedded_key_class = None
            if set_list_class :
                field_value . widget . attrs [ 'class' ] += " listField" . format ( base_key )
                embedded_key_class = make_key ( field_key , exclude_last_string = True )
            field_value . widget . attrs [ 'class' ] += " embeddedField"
            # Setting the embedded key correctly allows to visually nest the
            # embedded documents on the front end .
            if base_key == parent_key :
                field_value . widget . attrs [ 'class' ] += ' {0}' . format ( base_key )
            else :
                field_value . widget . attrs [ 'class' ] += ' {0} {1}' . format ( base_key , parent_key )
            if embedded_key_class is not None :
                field_value . widget . attrs [ 'class' ] += ' {0}' . format ( embedded_key_class )
        default_value = self . get_field_value ( form_key )
        # Style embedded documents
        if isinstance ( default_value , list ) and len ( default_value ) > 0 :
            key_index = int ( form_key . split ( "_" ) [ - 1 ] )
            new_base_key = make_key ( form_key , exclude_last_string = True )
            for list_value in default_value : # Note , this is copied every time so each widget gets a different class
                list_widget = deepcopy ( field_value . widget )
                new_key = make_key ( new_base_key , six . text_type ( key_index ) )
                list_widget . attrs [ 'class' ] += " {0}" . format ( make_key ( base_key , key_index ) )
                self . set_form_field ( list_widget , field_value . document_field , new_key , list_value )
                key_index += 1
        else :
            self . set_form_field ( field_value . widget , field_value . document_field , form_key , default_value )
    elif isinstance ( field_value , dict ) :
        self . set_form_fields ( field_value , form_key , field_value . get ( "_field_type" , None ) )


signature:
def set_form_field ( self , widget , model_field , field_key , default_value ) :
    """Parmams :
 widget - - the widget to use for displyaing the model _ field
 model _ field - - the field on the model to create a form field with
 field _ key - - the name for the field on the form
 default _ value - - the value to give for the field
 Default : None"""

implementation:
# Empty lists cause issues on form validation
if default_value == [ ] :
    default_value = None
if widget and isinstance ( widget , forms . widgets . Select ) :
    self . form . fields [ field_key ] = forms . ChoiceField ( label = model_field . name , required = model_field . required , widget = widget )
else :
    field_class = get_form_field_class ( model_field )
    self . form . fields [ field_key ] = field_class ( label = model_field . name , required = model_field . required , widget = widget )
if default_value is not None :
    if isinstance ( default_value , Document ) : # Probably a reference field , therefore , add id
        self . form . fields [ field_key ] . initial = getattr ( default_value , 'id' , None )
    else :
        self . form . fields [ field_key ] . initial = default_value
else :
    self . form . fields [ field_key ] . initial = getattr ( model_field , 'default' , None )
if isinstance ( model_field , ReferenceField ) :
    self . form . fields [ field_key ] . choices = [ ( six . text_type ( x . id ) , get_document_unicode ( x ) ) for x in model_field . document_type . objects . all ( ) ]
    # Adding in blank choice so a reference field can be deleted by selecting blank
    self . form . fields [ field_key ] . choices . insert ( 0 , ( "" , "" ) )
elif model_field . choices :
    self . form . fields [ field_key ] . choices = model_field . choices
for key , form_attr in CHECK_ATTRS . items ( ) :
    if hasattr ( model_field , key ) :
        value = getattr ( model_field , key )
        setattr ( self . form . fields [ field_key ] , key , value )


signature:
def get_field_value ( self , field_key ) :
    """Given field _ key will return value held at self . model _ instance . If
 model _ instance has not been provided will return None ."""

implementation:
def get_value ( document , field_key ) : # Short circuit the function if we do not have a document
    if document is None :
        return None
    current_key , new_key_array = trim_field_key ( document , field_key )
    key_array_digit = int ( new_key_array [ - 1 ] ) if new_key_array and has_digit ( new_key_array ) else None
    new_key = make_key ( new_key_array )
    if key_array_digit is not None and len ( new_key_array ) > 0 : # Handleing list fields
        if len ( new_key_array ) == 1 :
            return_data = document . _data . get ( current_key , [ ] )
        elif isinstance ( document , BaseList ) :
            return_list = [ ]
            if len ( document ) > 0 :
                return_list = [ get_value ( doc , new_key ) for doc in document ]
            return_data = return_list
        else :
            return_data = get_value ( getattr ( document , current_key ) , new_key )
    elif len ( new_key_array ) > 0 :
        return_data = get_value ( document . _data . get ( current_key ) , new_key )
    else : # Handeling all other fields and id
        try : # Added try except otherwise we get " TypeError : getattr ( ) : attribute name must be string " error from mongoengine / base / datastructures . py
            return_data = ( document . _data . get ( None , None ) if current_key == "id" else document . _data . get ( current_key , None ) )
        except :
            return_data = document . _data . get ( current_key , None )
    return return_data
if self . is_initialized :
    return get_value ( self . model_instance , field_key )
else :
    return None


signature:
def has_digit ( string_or_list , sep = "_" ) :
    """Given a string or a list will return true if the last word or
 element is a digit . sep is used when a string is given to know
 what separates one word from another ."""

implementation:
if isinstance ( string_or_list , ( tuple , list ) ) :
    list_length = len ( string_or_list )
    if list_length :
        return six . text_type ( string_or_list [ - 1 ] ) . isdigit ( )
    else :
        return False
else :
    return has_digit ( string_or_list . split ( sep ) )


signature:
def make_key ( * args , ** kwargs ) :
    """Given any number of lists and strings will join them in order as one
 string separated by the sep kwarg . sep defaults to u " _ " .
 Add exclude _ last _ string = True as a kwarg to exclude the last item in a
 given string after being split by sep . Note if you only have one word
 in your string you can end up getting an empty string .
 Example uses :
 > > > from mongonaut . forms . form _ utils import make _ key
 > > > make _ key ( ' hi ' , ' my ' , ' firend ' )
 > > > u ' hi _ my _ firend '
 > > > make _ key ( ' hi ' , ' my ' , ' firend ' , sep = ' i ' )
 > > > ' hiimyifirend '
 > > > make _ key ( ' hi ' , ' my ' , ' firend ' , [ ' this ' , ' be ' , ' what ' ] , sep = ' i ' )
 > > > ' hiimyifirendithisibeiwhat '
 > > > make _ key ( ' hi ' , ' my ' , ' firend ' , [ ' this ' , ' be ' , ' what ' ] )
 > > > u ' hi _ my _ firend _ this _ be _ what '"""

implementation:
sep = kwargs . get ( 'sep' , u"_" )
exclude_last_string = kwargs . get ( 'exclude_last_string' , False )
string_array = [ ]
for arg in args :
    if isinstance ( arg , list ) :
        string_array . append ( six . text_type ( sep . join ( arg ) ) )
    else :
        if exclude_last_string :
            new_key_array = arg . split ( sep ) [ : - 1 ]
            if len ( new_key_array ) > 0 :
                string_array . append ( make_key ( new_key_array ) )
        else :
            string_array . append ( six . text_type ( arg ) )
return sep . join ( string_array )


signature:
def set_fields ( self ) :
    """Sets existing data to form fields ."""

implementation:
# Get dictionary map of current model
if self . is_initialized :
    self . model_map_dict = self . create_document_dictionary ( self . model_instance )
else :
    self . model_map_dict = self . create_document_dictionary ( self . model )
form_field_dict = self . get_form_field_dict ( self . model_map_dict )
self . set_form_fields ( form_field_dict )


signature:
def set_post_data ( self ) :
    """Need to set form data so that validation on all post data occurs and
 places newly entered form data on the form object ."""

implementation:
self . form . data = self . post_data_dict
# Specifically adding list field keys to the form so they are included
# in form . cleaned _ data after the call to is _ valid
for field_key , field in self . form . fields . items ( ) :
    if has_digit ( field_key ) : # We have a list field .
        base_key = make_key ( field_key , exclude_last_string = True )
        # Add new key value with field to form fields so validation
        # will work correctly
        for key in self . post_data_dict . keys ( ) :
            if base_key in key :
                self . form . fields . update ( { key : field } )


signature:
def get_form ( self ) :
    """Generate the form for view ."""

implementation:
self . set_fields ( )
if self . post_data_dict is not None :
    self . set_post_data ( )
return self . form


signature:
def create_doc_dict ( self , document , doc_key = None , owner_document = None ) :
    """Generate a dictionary representation of the document . ( no recursion )
 DO NOT CALL DIRECTLY"""

implementation:
# Get doc field for top level documents
if owner_document :
    doc_field = owner_document . _fields . get ( doc_key , None ) if doc_key else None
else :
    doc_field = document . _fields . get ( doc_key , None ) if doc_key else None
# Generate the base fields for the document
doc_dict = { "_document" : document if owner_document is None else owner_document , "_key" : document . __class__ . __name__ . lower ( ) if doc_key is None else doc_key , "_document_field" : doc_field }
if not isinstance ( document , TopLevelDocumentMetaclass ) and doc_key :
    doc_dict . update ( { "_field_type" : EmbeddedDocumentField } )
for key , field in document . _fields . items ( ) :
    doc_dict [ key ] = field
return doc_dict


signature:
def create_list_dict ( self , document , list_field , doc_key ) :
    """Genereates a dictionary representation of the list field . Document
 should be the document the list _ field comes from .
 DO NOT CALL DIRECTLY"""

implementation:
list_dict = { "_document" : document }
if isinstance ( list_field . field , EmbeddedDocumentField ) :
    list_dict . update ( self . create_document_dictionary ( document = list_field . field . document_type_obj , owner_document = document ) )
# Set the list _ dict after it may have been updated
list_dict . update ( { "_document_field" : list_field . field , "_key" : doc_key , "_field_type" : ListField , "_widget" : get_widget ( list_field . field ) , "_value" : getattr ( document , doc_key , None ) } )
return list_dict


signature:
def create_document_dictionary ( self , document , document_key = None , owner_document = None ) :
    """Given document generates a dictionary representation of the document .
 Includes the widget for each for each field in the document ."""

implementation:
doc_dict = self . create_doc_dict ( document , document_key , owner_document )
for doc_key , doc_field in doc_dict . items ( ) : # Base fields should not be evaluated
    if doc_key . startswith ( "_" ) :
        continue
    if isinstance ( doc_field , ListField ) :
        doc_dict [ doc_key ] = self . create_list_dict ( document , doc_field , doc_key )
    elif isinstance ( doc_field , EmbeddedDocumentField ) :
        doc_dict [ doc_key ] = self . create_document_dictionary ( doc_dict [ doc_key ] . document_type_obj , doc_key )
    else :
        doc_dict [ doc_key ] = { "_document" : document , "_key" : doc_key , "_document_field" : doc_field , "_widget" : get_widget ( doc_dict [ doc_key ] , getattr ( doc_field , 'disabled' , False ) ) }
return doc_dict


signature:
def get_widget ( model_field , disabled = False ) :
    """Choose which widget to display for a field ."""

implementation:
attrs = get_attrs ( model_field , disabled )
if hasattr ( model_field , "max_length" ) and not model_field . max_length :
    return forms . Textarea ( attrs = attrs )
elif isinstance ( model_field , DateTimeField ) :
    return forms . DateTimeInput ( attrs = attrs )
elif isinstance ( model_field , BooleanField ) :
    return forms . CheckboxInput ( attrs = attrs )
elif isinstance ( model_field , ReferenceField ) or model_field . choices :
    return forms . Select ( attrs = attrs )
elif ( isinstance ( model_field , ListField ) or isinstance ( model_field , EmbeddedDocumentField ) or isinstance ( model_field , GeoPointField ) ) :
    return None
else :
    return forms . TextInput ( attrs = attrs )


signature:
def get_attrs ( model_field , disabled = False ) :
    """Set attributes on the display widget ."""

implementation:
attrs = { }
attrs [ 'class' ] = 'span6 xlarge'
if disabled or isinstance ( model_field , ObjectIdField ) :
    attrs [ 'class' ] += ' disabled'
    attrs [ 'readonly' ] = 'readonly'
return attrs


signature:
def get_form_field_class ( model_field ) :
    """Gets the default form field for a mongoenigne field ."""

implementation:
FIELD_MAPPING = { IntField : forms . IntegerField , StringField : forms . CharField , FloatField : forms . FloatField , BooleanField : forms . BooleanField , DateTimeField : forms . DateTimeField , DecimalField : forms . DecimalField , URLField : forms . URLField , EmailField : forms . EmailField }
return FIELD_MAPPING . get ( model_field . __class__ , forms . CharField )


signature:
def get_document_value ( document , key ) :
    '''Returns the display value of a field for a particular MongoDB document .'''

implementation:
value = getattr ( document , key )
if isinstance ( value , ObjectId ) :
    return value
if isinstance ( document . _fields . get ( key ) , URLField ) :
    return mark_safe ( """{1}""" . format ( value , value ) )
if isinstance ( value , Document ) :
    app_label = value . __module__ . replace ( ".models" , "" )
    document_name = value . _class_name
    url = reverse ( "document_detail" , kwargs = { 'app_label' : app_label , 'document_name' : document_name , 'id' : value . id } )
    return mark_safe ( """{1}""" . format ( url , value ) )
return value


signature:
def get_qset ( self , queryset , q ) :
    """Performs filtering against the default queryset returned by
 mongoengine ."""

implementation:
if self . mongoadmin . search_fields and q :
    params = { }
    for field in self . mongoadmin . search_fields :
        if field == 'id' : # check to make sure this is a valid ID , otherwise we just continue
            if is_valid_object_id ( q ) :
                return queryset . filter ( pk = q )
            continue
        search_key = "{field}__icontains" . format ( field = field )
        params [ search_key ] = q
    queryset = queryset . filter ( ** params )
return queryset


signature:
def get_queryset ( self ) :
    """Replicates Django CBV ` get _ queryset ( ) ` method , but for MongoEngine ."""

implementation:
if hasattr ( self , "queryset" ) and self . queryset :
    return self . queryset
self . set_mongonaut_base ( )
self . set_mongoadmin ( )
self . document = getattr ( self . models , self . document_name )
queryset = self . document . objects . all ( )
if self . mongoadmin . ordering :
    queryset = queryset . order_by ( * self . mongoadmin . ordering )
# search . move this to get _ queryset
# search . move this to get _ queryset
q = self . request . GET . get ( 'q' )
queryset = self . get_qset ( queryset , q )
# # # Start pagination
# # # Note :
# # # Can ' t use Paginator in Django because mongoengine querysets are
# # # not the same as Django ORM querysets and it broke .
# Make sure page request is an int . If not , deliver first page .
try :
    self . page = int ( self . request . GET . get ( 'page' , '1' ) )
except ValueError :
    self . page = 1
obj_count = queryset . count ( )
self . total_pages = math . ceil ( obj_count / self . documents_per_page )
if self . page > self . total_pages :
    self . page = self . total_pages
if self . page < 1 :
    self . page = 1
start = ( self . page - 1 ) * self . documents_per_page
end = self . page * self . documents_per_page
queryset = queryset [ start : end ] if obj_count else queryset
self . queryset = queryset
return queryset


signature:
def get_initial ( self ) :
    """Used during adding / editing of data ."""

implementation:
self . query = self . get_queryset ( )
mongo_ids = { 'mongo_id' : [ str ( x . id ) for x in self . query ] }
return mongo_ids


signature:
def get_context_data ( self , ** kwargs ) :
    """Injects data into the context to replicate CBV ListView ."""

implementation:
context = super ( DocumentListView , self ) . get_context_data ( ** kwargs )
context = self . set_permissions_in_context ( context )
if not context [ 'has_view_permission' ] :
    return HttpResponseForbidden ( "You do not have permissions to view this content." )
context [ 'object_list' ] = self . get_queryset ( )
context [ 'document' ] = self . document
context [ 'app_label' ] = self . app_label
context [ 'document_name' ] = self . document_name
context [ 'request' ] = self . request
# pagination bits
context [ 'page' ] = self . page
context [ 'documents_per_page' ] = self . documents_per_page
if self . page > 1 :
    previous_page_number = self . page - 1
else :
    previous_page_number = None
if self . page < self . total_pages :
    next_page_number = self . page + 1
else :
    next_page_number = None
context [ 'previous_page_number' ] = previous_page_number
context [ 'has_previous_page' ] = previous_page_number is not None
context [ 'next_page_number' ] = next_page_number
context [ 'has_next_page' ] = next_page_number is not None
context [ 'total_pages' ] = self . total_pages
# Part of upcoming list view form functionality
if self . queryset . count ( ) :
    context [ 'keys' ] = [ 'id' , ]
    # Show those items for which we ' ve got list _ fields on the mongoadmin
    for key in [ x for x in self . mongoadmin . list_fields if x != 'id' and x in self . document . _fields . keys ( ) ] : # TODO - Figure out why this EmbeddedDocumentField and ListField breaks this view
    # Note - This is the challenge part , right ? : )
        if isinstance ( self . document . _fields [ key ] , EmbeddedDocumentField ) :
            continue
        if isinstance ( self . document . _fields [ key ] , ListField ) :
            continue
        context [ 'keys' ] . append ( key )
if self . mongoadmin . search_fields :
    context [ 'search_field' ] = True
return context


signature:
def post ( self , request , * args , ** kwargs ) :
    """Creates new mongoengine records ."""

implementation:
# TODO - make sure to check the rights of the poster
# self . get _ queryset ( ) # TODO - write something that grabs the document class better
form_class = self . get_form_class ( )
form = self . get_form ( form_class )
mongo_ids = self . get_initial ( ) [ 'mongo_id' ]
for form_mongo_id in form . data . getlist ( 'mongo_id' ) :
    for mongo_id in mongo_ids :
        if form_mongo_id == mongo_id :
            self . document . objects . get ( pk = mongo_id ) . delete ( )
return self . form_invalid ( form )


signature:
def get_context_data ( self , ** kwargs ) :
    """TODO - possibly inherit this from DocumentEditFormView . This is same thing minus :
 self . ident = self . kwargs . get ( ' id ' )
 self . document = self . document _ type . objects . get ( pk = self . ident )"""

implementation:
context = super ( DocumentAddFormView , self ) . get_context_data ( ** kwargs )
self . set_mongoadmin ( )
context = self . set_permissions_in_context ( context )
self . document_type = getattr ( self . models , self . document_name )
context [ 'app_label' ] = self . app_label
context [ 'document_name' ] = self . document_name
context [ 'form_action' ] = reverse ( 'document_detail_add_form' , args = [ self . kwargs . get ( 'app_label' ) , self . kwargs . get ( 'document_name' ) ] )
return context


signature:
def get_mongoadmins ( self ) :
    """Returns a list of all mongoadmin implementations for the site"""

implementation:
apps = [ ]
for app_name in settings . INSTALLED_APPS :
    mongoadmin = "{0}.mongoadmin" . format ( app_name )
    try :
        module = import_module ( mongoadmin )
    except ImportError as e :
        if str ( e ) . startswith ( "No module named" ) :
            continue
        raise e
    app_store = AppStore ( module )
    apps . append ( dict ( app_name = app_name , obj = app_store ) )
return apps


signature:
def set_mongonaut_base ( self ) :
    """Sets a number of commonly used attributes"""

implementation:
if hasattr ( self , "app_label" ) : # prevents us from calling this multiple times
    return None
self . app_label = self . kwargs . get ( 'app_label' )
self . document_name = self . kwargs . get ( 'document_name' )
# TODO Allow this to be assigned via url variable
self . models_name = self . kwargs . get ( 'models_name' , 'models' )
# import the models file
self . model_name = "{0}.{1}" . format ( self . app_label , self . models_name )
self . models = import_module ( self . model_name )