signature:
def add_link ( cls , attr , title = '' , display = '' ) :
    """Adds a ` ` list _ display ` ` attribute that appears as a link to the
 django admin change page for the type of object being shown . Supports
 double underscore attribute name dereferencing .
 : param attr :
 Name of the attribute to dereference from the corresponding
 object , i . e . what will be lined to . This name supports double
 underscore object link referencing for ` ` models . ForeignKey ` `
 members .
 : param title :
 Title for the column of the django admin table . If not given it
 defaults to a capitalized version of ` ` attr ` `
 : param display :
 What to display as the text for the link being shown . If not
 given it defaults to the string representation of the object for
 the row : ` ` str ( obj ) ` ` . This parameter supports django
 templating , the context for which contains a dictionary key named
 " obj " with the value being the object for the row .
 Example usage :
 . . code - block : : python
 # - - - - admin . py file - - - -
 base = fancy _ modeladmin ( ' id ' )
 base . add _ link ( ' author ' , ' Our Authors ' ,
 ' { { obj . name } } ( id = { { obj . id } } ) ' )
 @ admin . register ( Book )
 class BookAdmin ( base ) :
 pass
 The django admin change page for the Book class would have a column
 for " id " and another titled " Our Authors " . The " Our Authors " column
 would have a link for each Author object referenced by " book . author " .
 The link would go to the Author django admin change listing . The
 display of the link would be the name of the author with the id in
 brakcets , e . g . " Douglas Adams ( id = 42 ) " """

implementation:
global klass_count
klass_count += 1
fn_name = 'dyn_fn_%d' % klass_count
cls . list_display . append ( fn_name )
if not title :
    title = attr . capitalize ( )
# python scoping is a bit weird with default values , if it isn ' t
# referenced the inner function won ' t see it , so assign it for use
_display = display
def _link ( self , obj ) :
    field_obj = admin_obj_attr ( obj , attr )
    if not field_obj :
        return ''
    text = _obj_display ( field_obj , _display )
    return admin_obj_link ( field_obj , text )
_link . short_description = title
_link . allow_tags = True
_link . admin_order_field = attr
setattr ( cls , fn_name , _link )


signature:
def add_object ( cls , attr , title = '' , display = '' ) :
    """Adds a ` ` list _ display ` ` attribute showing an object . Supports
 double underscore attribute name dereferencing .
 : param attr :
 Name of the attribute to dereference from the corresponding
 object , i . e . what will be lined to . This name supports double
 underscore object link referencing for ` ` models . ForeignKey ` `
 members .
 : param title :
 Title for the column of the django admin table . If not given it
 defaults to a capitalized version of ` ` attr ` `
 : param display :
 What to display as the text for the link being shown . If not
 given it defaults to the string representation of the object for
 the row : ` ` str ( obj ) ` ` . This parameter supports django templating ,
 the context for which contains a dictionary key named " obj " with
 the value being the object for the row ."""

implementation:
global klass_count
klass_count += 1
fn_name = 'dyn_fn_%d' % klass_count
cls . list_display . append ( fn_name )
if not title :
    title = attr . capitalize ( )
# python scoping is a bit weird with default values , if it isn ' t
# referenced the inner function won ' t see it , so assign it for use
_display = display
def _ref ( self , obj ) :
    field_obj = admin_obj_attr ( obj , attr )
    if not field_obj :
        return ''
    return _obj_display ( field_obj , _display )
_ref . short_description = title
_ref . allow_tags = True
_ref . admin_order_field = attr
setattr ( cls , fn_name , _ref )


signature:
def add_formatted_field ( cls , field , format_string , title = '' ) :
    """Adds a ` ` list _ display ` ` attribute showing a field in the object
 using a python % formatted string .
 : param field :
 Name of the field in the object .
 : param format _ string :
 A old - style ( to remain python 2 . x compatible ) % string formatter
 with a single variable reference . The named ` ` field ` ` attribute
 will be passed to the formatter using the " % " operator .
 : param title :
 Title for the column of the django admin table . If not given it
 defaults to a capitalized version of ` ` field ` `"""

implementation:
global klass_count
klass_count += 1
fn_name = 'dyn_fn_%d' % klass_count
cls . list_display . append ( fn_name )
if not title :
    title = field . capitalize ( )
# python scoping is a bit weird with default values , if it isn ' t
# referenced the inner function won ' t see it , so assign it for use
_format_string = format_string
def _ref ( self , obj ) :
    return _format_string % getattr ( obj , field )
_ref . short_description = title
_ref . allow_tags = True
_ref . admin_order_field = field
setattr ( cls , fn_name , _ref )


signature:
def post_required ( method_or_options = [ ] ) :
    """View decorator that enforces that the method was called using POST .
 This decorator can be called with or without parameters . As it is
 expected to wrap a view , the first argument of the method being wrapped is
 expected to be a ` ` request ` ` object .
 . . code - block : : python
 @ post _ required
 def some _ view ( request ) :
 pass
 @ post _ required ( [ ' firstname ' , ' lastname ' ] )
 def some _ view ( request ) :
 pass
 The optional parameter contains a single list which specifies the names of
 the expected fields in the POST dictionary . The list is not exclusive ,
 you can pass in fields that are not checked by the decorator .
 : param options :
 List of the names of expected POST keys ."""

implementation:
def decorator ( method ) : # handle wrapping or wrapping with arguments ; if no arguments ( and no
# calling parenthesis ) then method _ or _ options will be a list ,
# otherwise it will be the wrapped function
    expected_fields = [ ]
    if not callable ( method_or_options ) : # not callable means wrapping with arguments
        expected_fields = method_or_options
    @ wraps ( method )
    def wrapper ( * args , ** kwargs ) :
        request = args [ 0 ]
        if request . method != 'POST' :
            logger . error ( 'POST required for this url' )
            raise Http404 ( 'only POST allowed for this url' )
        missing = [ ]
        for field in expected_fields :
            if field not in request . POST :
                missing . append ( field )
        if missing :
            s = 'Expected fields missing in POST: %s' % missing
            logger . error ( s )
            raise Http404 ( s )
        # everything verified , run the view
        return method ( * args , ** kwargs )
    return wrapper
if callable ( method_or_options ) : # callable means decorated method without options , call our decorator
    return decorator ( method_or_options )
return decorator


signature:
def json_post_required ( * decorator_args ) :
    """View decorator that enforces that the method was called using POST and
 contains a field containing a JSON dictionary . This method should
 only be used to wrap views and assumes the first argument of the method
 being wrapped is a ` ` request ` ` object .
 . . code - block : : python
 @ json _ post _ required ( ' data ' , ' json _ data ' )
 def some _ view ( request ) :
 username = request . json _ data [ ' username ' ]
 : param field :
 The name of the POST field that contains a JSON dictionary
 : param request _ name :
 [ optional ] Name of the parameter on the request to put the
 deserialized JSON data . If not given the field name is used"""

implementation:
def decorator ( method ) :
    @ wraps ( method )
    def wrapper ( * args , ** kwargs ) :
        field = decorator_args [ 0 ]
        if len ( decorator_args ) == 2 :
            request_name = decorator_args [ 1 ]
        else :
            request_name = field
        request = args [ 0 ]
        if request . method != 'POST' :
            logger . error ( 'POST required for this url' )
            raise Http404 ( 'only POST allowed for this url' )
        if field not in request . POST :
            s = 'Expected field named %s in POST' % field
            logger . error ( s )
            raise Http404 ( s )
        # deserialize the JSON and put it in the request
        setattr ( request , request_name , json . loads ( request . POST [ field ] ) )
        # everything verified , run the view
        return method ( * args , ** kwargs )
    return wrapper
return decorator


signature:
def hist ( x , bins = 10 , labels = None , aspect = "auto" , plot = True , ax = None , range = None ) :
    """Creates a histogram of data * x * with a * bins * , * labels * = : code : ` [ title , xlabel , ylabel ] ` ."""

implementation:
h , edge = _np . histogram ( x , bins = bins , range = range )
mids = edge + ( edge [ 1 ] - edge [ 0 ] ) / 2
mids = mids [ : - 1 ]
if plot :
    if ax is None :
        _plt . hist ( x , bins = bins , range = range )
    else :
        ax . hist ( x , bins = bins , range = range )
    if labels is not None :
        _addlabel ( labels [ 0 ] , labels [ 1 ] , labels [ 2 ] )
return h , mids


signature:
def sigma ( self ) :
    """Spot size of matched beam : math : ` \\ left ( \\ frac { 2 E \\ varepsilon _ 0 } { n _ p e ^ 2 } \\ right ) ^ { 1/4 } \\ sqrt { \\ epsilon } `"""

implementation: return _np . power ( 2 * _sltr . GeV2joule ( self . E ) * _spc . epsilon_0 / ( self . plasma . n_p * _np . power ( _spc . elementary_charge , 2 ) ) , 0.25 ) * _np . sqrt ( self . emit )


signature:
def sigma_prime ( self ) :
    """Divergence of matched beam"""

implementation: return _np . sqrt ( self . emit / self . beta ( self . E ) )


signature:
def n_p ( self ) :
    """The plasma density in SI units ."""

implementation: return 2 * _sltr . GeV2joule ( self . E ) * _spc . epsilon_0 / ( self . beta * _spc . elementary_charge ) ** 2


signature:
def plasma ( self , species = _pt . hydrogen ) :
    """The matched : class : ` Plasma ` ."""

implementation: return _Plasma ( self . n_p , species = species )


signature:
def main ( target , label ) :
    """Semver tag triggered deployment helper"""

implementation:
check_environment ( target , label )
click . secho ( 'Fetching tags from the upstream ...' )
handler = TagHandler ( git . list_tags ( ) )
print_information ( handler , label )
tag = handler . yield_tag ( target , label )
confirm ( tag )


signature:
def check_environment ( target , label ) :
    """Performs some environment checks prior to the program ' s execution"""

implementation:
if not git . exists ( ) :
    click . secho ( 'You must have git installed to use yld.' , fg = 'red' )
    sys . exit ( 1 )
if not os . path . isdir ( '.git' ) :
    click . secho ( 'You must cd into a git repository to use yld.' , fg = 'red' )
    sys . exit ( 1 )
if not git . is_committed ( ) :
    click . secho ( 'You must commit or stash your work before proceeding.' , fg = 'red' )
    sys . exit ( 1 )
if target is None and label is None :
    click . secho ( 'You must specify either a target or a label.' , fg = 'red' )
    sys . exit ( 1 )


signature:
def print_information ( handler , label ) :
    """Prints latest tag ' s information"""

implementation:
click . echo ( '=> Latest stable: {tag}' . format ( tag = click . style ( str ( handler . latest_stable or 'N/A' ) , fg = 'yellow' if handler . latest_stable else 'magenta' ) ) )
if label is not None :
    latest_revision = handler . latest_revision ( label )
    click . echo ( '=> Latest relative revision ({label}): {tag}' . format ( label = click . style ( label , fg = 'blue' ) , tag = click . style ( str ( latest_revision or 'N/A' ) , fg = 'yellow' if latest_revision else 'magenta' ) ) )


signature:
def confirm ( tag ) :
    """Prompts user before proceeding"""

implementation:
click . echo ( )
if click . confirm ( 'Do you want to create the tag {tag}?' . format ( tag = click . style ( str ( tag ) , fg = 'yellow' ) ) , default = True , abort = True ) :
    git . create_tag ( tag )
if click . confirm ( 'Do you want to push the tag {tag} into the upstream?' . format ( tag = click . style ( str ( tag ) , fg = 'yellow' ) ) , default = True ) :
    git . push_tag ( tag )
    click . echo ( 'Done!' )
else :
    git . delete_tag ( tag )
    click . echo ( 'Aborted!' )


signature:
def imshow ( X , ax = None , add_cbar = True , rescale_fig = True , ** kwargs ) :
    """Plots an array * X * such that the first coordinate is the * x * coordinate and the second coordinate is the * y * coordinate , with the origin at the bottom left corner .
 Optional argument * ax * allows an existing axes to be used .
 * \ * \ * kwargs * are passed on to : meth : ` matplotlib . axes . Axes . imshow ` .
 . . versionadded : : 1.3
 Returns
 fig , ax , im :
 if axes aren ' t specified .
 im :
 if axes are specified ."""

implementation: return _plot_array ( X , plottype = _IMSHOW , ax = ax , add_cbar = add_cbar , rescale_fig = rescale_fig , ** kwargs )


signature:
def scaled_figsize ( X , figsize = None , h_pad = None , v_pad = None ) :
    """Given an array * X * , determine a good size for the figure to be by shrinking it to fit within * figsize * . If not specified , shrinks to fit the figsize specified by the current : attr : ` matplotlib . rcParams ` .
 . . versionadded : : 1.3"""

implementation:
if figsize is None :
    figsize = _mpl . rcParams [ 'figure.figsize' ]
# Find the height and width
width , height = _np . shape ( X )
ratio = width / height
# Find how to rescale the figure
if ratio > figsize [ 0 ] / figsize [ 1 ] :
    figsize [ 1 ] = figsize [ 0 ] / ratio
else :
    figsize [ 0 ] = figsize [ 1 ] * ratio
return figsize


signature:
def get ( f , key , default = None ) :
    """Gets an array from datasets .
 . . versionadded : : 1.4"""

implementation:
if key in f . keys ( ) :
    val = f [ key ] . value
    if default is None :
        return val
    else :
        if _np . shape ( val ) == _np . shape ( default ) :
            return val
return default


signature:
def get_state ( self ) :
    """Get the current directory state"""

implementation: return [ os . path . join ( dp , f ) for dp , _ , fn in os . walk ( self . dir ) for f in fn ]


signature:
def tick ( self ) :
    """Add one tick to progress bar"""

implementation:
self . current += 1
if self . current == self . factor :
    sys . stdout . write ( '+' )
    sys . stdout . flush ( )
    self . current = 0


signature:
def push ( self , k ) :
    """Push k to the top of the list
 > > > l = DLL ( )
 > > > l . push ( 1)
 > > > l . push ( 2)
 [2 , 1]
 > > > l . push ( 3)
 [3 , 2 , 1]"""

implementation:
if not self . _first : # first item
    self . _first = self . _last = node = DLL . Node ( k )
elif self . _first . value == k : # it ' s already at the top
    return
else :
    try :
        self . delete ( k )
        # in case we have it already
    except KeyError :
        pass
    self . _first = node = self . _first . insert_before ( k )
self . _index [ k ] = node
self . _size += 1


signature:
def deletenode ( self , node ) :
    """> > > l = DLL ( )
 > > > l . push ( 1)
 > > > l . size ( )
 > > > l . deletenode ( l . _ first )
 > > > l . size ( )
 > > > l . _ index
 > > > l . _ first"""

implementation:
if self . _last == node :
    self . _last = node . previous
if self . _first == node :
    self . _first = node . next
node . pop ( )
del self . _index [ node . value ]
self . _size -= 1


implementation:
k = self . _last . value
self . deletenode ( self . _last )
return k


signature:
def increment ( cls , name ) :
    """Call this method to increment the named counter . This is atomic on
 the database .
 : param name :
 Name for a previously created ` ` Counter ` ` object"""

implementation:
with transaction . atomic ( ) :
    counter = Counter . objects . select_for_update ( ) . get ( name = name )
    counter . value += 1
    counter . save ( )
return counter . value


signature:
def pcolor_axes ( array , px_to_units = px_to_units ) :
    """Return axes : code : ` x , y ` for * array * to be used with : func : ` matplotlib . pyplot . color ` .
 * px _ to _ units * is a function to convert pixels to units . By default , returns pixels ."""

implementation:
# Coords need to be + 1 larger than array
x_size = array . shape [ 0 ] + 1
y_size = array . shape [ 1 ] + 1
x = _np . empty ( ( x_size , y_size ) )
y = _np . empty ( ( x_size , y_size ) )
for i in range ( x_size ) :
    for j in range ( y_size ) :
        x [ i , j ] , y [ i , j ] = px_to_units ( i - 0.5 , j - 0.5 )
return x , y


signature:
def nb0 ( self ) :
    """On - axis beam density : math : ` n _ { b , 0 } ` ."""

implementation: return self . N_e / ( ( 2 * _np . pi ) ** ( 3 / 2 ) * self . sig_r ** 2 * self . sig_xi )


signature:
def lambda_large ( self , r0 ) :
    """The wavelength for large ( : math : ` r _ 0 < \\ sigma _ r ` ) oscillations ."""

implementation: return 2 * _np . sqrt ( 2 * _np . pi / self . k ) * r0


signature:
def r_small ( self , x , r0 ) :
    """Approximate trajectory function for small ( : math : ` r _ 0 < \\ sigma _ r ` ) oscillations ."""

implementation: return r0 * _np . cos ( _np . sqrt ( self . k_small ) * x )


signature:
def r_large ( self , x , r0 ) :
    """Approximate trajectory function for large ( : math : ` r _ 0 > \\ sigma _ r ` ) oscillations ."""

implementation: return r0 * _np . cos ( x * self . omega_big ( r0 ) )


signature:
def k ( self ) :
    """Driving force term : : math : ` r ' ' = - k \\ left ( \\ frac { 1 - e ^ { - r ^ 2/2{ \\ sigma _ r } ^ 2 } } { r } \\ right ) `"""

implementation:
try :
    return self . _k
except AttributeError :
    self . _k = e ** 2 * self . N_e / ( ( 2 * _np . pi ) ** ( 5 / 2 ) * e0 * self . m * c ** 2 * self . sig_xi )
    return self . _k


signature:
def print_loading ( self , wait , message ) :
    """print loading message on screen
 . . note : :
 loading message only write to ` sys . stdout `
 : param int wait : seconds to wait
 : param str message : message to print
 : return : None"""

implementation:
tags = [ '\\' , '|' , '/' , '-' ]
for i in range ( wait ) :
    time . sleep ( 0.25 )
    sys . stdout . write ( "%(message)s... %(tag)s\r" % { 'message' : message , 'tag' : tags [ i % 4 ] } )
    sys . stdout . flush ( )
    pass
sys . stdout . write ( "%s... Done...\n" % message )
sys . stdout . flush ( )
pass


signature:
def warn_message ( self , message , fh = None , prefix = "[warn]:" , suffix = "..." ) :
    """print warn type message ,
 if file handle is ` sys . stdout ` , print color message
 : param str message : message to print
 : param file fh : file handle , default is ` sys . stdout `
 : param str prefix : message prefix , default is ` [ warn ] `
 : param str suffix : message suffix , default is ` . . . `
 : return : None"""

implementation:
msg = prefix + message + suffix
fh = fh or sys . stdout
if fh is sys . stdout :
    termcolor . cprint ( msg , color = "yellow" )
else :
    fh . write ( msg )
pass


signature:
def error_message ( self , message , fh = None , prefix = "[error]:" , suffix = "..." ) :
    """print error type message
 if file handle is ` sys . stderr ` , print color message
 : param str message : message to print
 : param file fh : file handle , default is ` sys . stdout `
 : param str prefix : message prefix , default is ` [ error ] `
 : param str suffix : message suffix , default is ' . . . '
 : return : None"""

implementation:
msg = prefix + message + suffix
fh = fh or sys . stderr
if fh is sys . stderr :
    termcolor . cprint ( msg , color = "red" )
else :
    fh . write ( msg )
pass


signature:
def system ( self , cmd , fake_code = False ) :
    """a built - in wrapper make dry - run easier .
 you should use this instead use ` os . system `
 . . note : :
 to use it , you need add ' - - dry - run ' option in
 your argparser options
 : param str cmd : command to execute
 : param bool fake _ code : only display command
 when is True , default is False
 : return :"""

implementation:
try :
    if self . options . dry_run :
        def fake_system ( cmd ) :
            self . print_message ( cmd )
            return fake_code
        return fake_system ( cmd )
except AttributeError :
    self . logger . warnning ( "fake mode enabled," "but you don't set '--dry-run' option " "in your argparser options" )
    pass
return os . system ( cmd )


signature:
def load_resource ( path , root = '' ) :
    """. . warning : :
 Experiment feature .
 BE CAREFUL ! WE MAY REMOVE THIS FEATURE !
 load resource file which in package .
 this method is used to load file easier in different environment .
 e . g :
 consume we have a file named ` resource . io ` in package ` cliez . conf ` ,
 and we want to load it .
 the easiest way may like this :
 . . code - block : : python
 open ( ' . . / conf / resource . io ' ) . read ( )
 An obvious problem is ` . . ` is relative path .
 it will cause an error .
 ` load _ resource ` is designed for solve this problem .
 The following code are equivalent :
 . . code - block : : python
 a = Component ( )
 a . load _ resource ( ' resource . io ' , root = ' cliez / base ' )
 a . load _ resource ( ' base / resource . io ' , root = ' cliez ' )
 a . load _ resource ( ' / base / resource . io ' , root = ' cliez ' )
 a . load _ resource ( ' cliez / base / resource . io ' )
 a . load _ resource ( _ _ file _ _ . rsplit ( ' / ' , 2 ) [ 0 ] +
 ' / cliez / base / resource . io ' )
 . . note : :
 The document charset * MUST BE * utf - 8
 : param str path : file path
 : param str root : root path
 : return : str"""

implementation:
if root :
    full_path = root + '/' + path . strip ( '/' )
else :
    full_path = path
buf = ''
try :
    buf = open ( full_path ) . read ( )
except IOError :
    pkg , path = full_path . split ( '/' , 1 )
    try :
        import pkg_resources
        buf = pkg_resources . resource_string ( pkg , path )
        # compatible python3 and only support utf - 8
        if type ( buf ) != str :
            buf = buf . decode ( 'utf-8' )
            pass
    except AttributeError : # load resource feature not work in python2
        pass
return buf


signature:
def load_description ( name , root = '' ) :
    """. . warning : :
 Experiment feature .
 BE CAREFUL ! WE MAY REMOVE THIS FEATURE !
 Load resource file as description ,
 if resource file not exist , will return empty string .
 : param str path : name resource path
 : param str root : same as ` load _ resource ` root
 : return : ` str `"""

implementation:
desc = ''
try :
    desc = Component . load_resource ( name , root = root )
except ( IOError , ImportError ) :
    pass
return desc


signature:
def name ( self ) :
    '''Returns the name of the Firebase . If a Firebase instance points to
 ' https : / / my _ firebase . firebaseio . com / users ' its name would be ' users ' '''

implementation:
i = self . __url . rfind ( '/' )
if self . __url [ : i ] == 'https:/' :
    return "/"
return self . __url [ i + 1 : ]


signature:
def url_correct ( self , point , auth = None , export = None ) :
    '''Returns a Corrected URL to be used for a Request
 as per the REST API .'''

implementation:
newUrl = self . __url + point + '.json'
if auth or export :
    newUrl += "?"
if auth :
    newUrl += ( "auth=" + auth )
if export :
    if not newUrl . endswith ( '?' ) :
        newUrl += "&"
    newUrl += "format=export"
return newUrl


signature:
def __read ( path ) :
    '''Reads a File with contents in correct JSON format .
 Returns the data as Python objects .
 path - ( string ) path to the file'''

implementation:
try :
    with open ( path , 'r' ) as data_file :
        data = data_file . read ( )
        data = json . loads ( data )
        return data
except IOError as err :
    pass
except Exception as err : # Invalid JSON formatted files
    pass


signature:
def __write ( path , data , mode = "w" ) :
    '''Writes to a File . Returns the data written .
 path - ( string ) path to the file to write to .
 data - ( json ) data from a request .
 mode - ( string ) mode to open the file in . Default to ' w ' . Overwrites .'''

implementation:
with open ( path , mode ) as data_file :
    data = json . dumps ( data , indent = 4 )
    data_file . write ( data )
    return data


signature:
def amust ( self , args , argv ) :
    '''Requires the User to provide a certain parameter
 for the method to function properly .
 Else , an Exception is raised .
 args - ( tuple ) arguments you are looking for .
 argv - ( dict ) arguments you have received and want to inspect .'''

implementation:
for arg in args :
    if str ( arg ) not in argv :
        raise KeyError ( "ArgMissing: " + str ( arg ) + " not passed" )


signature:
def catch_error ( response ) :
    '''Checks for Errors in a Response .
 401 or 403 - Security Rules Violation .
 404 or 417 - Firebase NOT Found .
 response - ( Request . Response ) - response from a request .'''

implementation:
status = response . status_code
if status == 401 or status == 403 :
    raise EnvironmentError ( "Forbidden" )
elif status == 417 or status == 404 :
    raise EnvironmentError ( "NotFound" )


signature:
def get_sync ( self , ** kwargs ) :
    '''GET : gets data from the Firebase .
 Requires the ' point ' parameter as a keyworded argument .'''

implementation:
self . amust ( ( "point" , ) , kwargs )
response = requests . get ( self . url_correct ( kwargs [ "point" ] , kwargs . get ( "auth" , self . __auth ) ) )
self . catch_error ( response )
return response . content


signature:
def put_sync ( self , ** kwargs ) :
    '''PUT : puts data into the Firebase .
 Requires the ' point ' parameter as a keyworded argument .'''

implementation:
self . amust ( ( "point" , "data" ) , kwargs )
response = requests . put ( self . url_correct ( kwargs [ "point" ] , kwargs . get ( "auth" , self . __auth ) ) , data = json . dumps ( kwargs [ "data" ] ) )
self . catch_error ( response )
return response . content


signature:
def post_sync ( self , ** kwargs ) :
    '''POST : post data to a Firebase .
 Requires the ' point ' parameter as a keyworded argument .
 Note : Firebase will give it a randomized " key "
 and the data will be the " value " . Thus a key - value pair'''

implementation:
self . amust ( ( "point" , "data" ) , kwargs )
response = requests . post ( self . url_correct ( kwargs [ "point" ] , kwargs . get ( "auth" , self . __auth ) ) , data = json . dumps ( kwargs [ "data" ] ) )
self . catch_error ( response )
return response . content


signature:
def update_sync ( self , ** kwargs ) :
    '''UPDATE : updates data in the Firebase .
 Requires the ' point ' parameter as a keyworded argument .'''

implementation:
self . amust ( ( "point" , "data" ) , kwargs )
# Sending the ' PATCH ' request
response = requests . patch ( self . url_correct ( kwargs [ "point" ] , kwargs . get ( "auth" , self . __auth ) ) , data = json . dumps ( kwargs [ "data" ] ) )
self . catch_error ( response )
return response . content


signature:
def delete_sync ( self , ** kwargs ) :
    '''DELETE : delete data in the Firebase .
 Requires the ' point ' parameter as a keyworded argument .'''

implementation:
self . amust ( ( "point" , ) , kwargs )
response = requests . delete ( self . url_correct ( kwargs [ "point" ] , kwargs . get ( "auth" , self . __auth ) ) )
self . catch_error ( response )
return response . content


signature:
def export_sync ( self , ** kwargs ) :
    '''EXPORT : exports data from the Firebase into a File , if given .
 Requires the ' point ' parameter as a keyworded argument .'''

implementation:
self . amust ( ( "point" , ) , kwargs )
response = requests . get ( self . url_correct ( kwargs [ "point" ] , kwargs . get ( "auth" , self . __auth ) , True ) )
self . catch_error ( response )
path = kwargs . get ( "path" , None )
if path :
    self . __write ( path , response . content , kwargs . get ( "mode" , "w" ) )
return response . content


signature:
def superable ( cls ) :
    '''Provide . _ _ super in python 2 . x classes without having to specify the current
 class name each time super is used ( DRY principle ) .'''

implementation:
name = cls . __name__
super_name = '_%s__super' % ( name , )
setattr ( cls , super_name , super ( cls ) )
return cls


signature:
def main ( ) :
    """Main method .
 This method holds what you want to execute when
 the script is run on command line ."""

implementation:
args = get_arguments ( )
setup_logging ( args )
version_path = os . path . abspath ( os . path . join ( os . path . dirname ( __file__ ) , '..' , '..' , '.VERSION' ) )
try :
    version_text = open ( version_path ) . read ( ) . strip ( )
except Exception :
    print ( 'Could not open or read the .VERSION file' )
    sys . exit ( 1 )
try :
    semver . parse ( version_text )
except ValueError :
    print ( ( 'The .VERSION file contains an invalid ' 'version: "{}"' ) . format ( version_text ) )
    sys . exit ( 1 )
new_version = version_text
if args . version :
    try :
        if semver . parse ( args . version ) :
            new_version = args . version
    except Exception :
        print ( 'Could not parse "{}" as a version' . format ( args . version ) )
        sys . exit ( 1 )
elif args . bump_major :
    new_version = semver . bump_major ( version_text )
elif args . bump_minor :
    new_version = semver . bump_minor ( version_text )
elif args . bump_patch :
    new_version = semver . bump_patch ( version_text )
try :
    with open ( version_path , 'w' ) as version_file :
        version_file . write ( new_version )
except Exception :
    print ( 'Could not write the .VERSION file' )
    sys . exit ( 1 )
print ( new_version )


signature:
def valid_url ( name ) :
    '''Validates a Firebase URL . A valid URL must exhibit :
 - Protocol must be HTTPS
 - Domain Must be firebaseio . com
 - firebasename can only contain :
 - hyphen , small and capital letters
 - firebasename can not have leading and trailing hyphens
 - childnames can not contain :
 - period , dollar sign , square brackets , pound
 - ASCII Control Chars from \x00 - \x1F and \x7F'''

implementation:
try :
    pattern = r'^https://([a-zA-Z0-9][a-zA-Z0-9\-]+[a-zA-Z0-9]).firebaseio.com(/[^\#\$\[\]\.\x00-\x1F\x7F]*)'
    result = re . search ( pattern , name + '/' ) . group ( )
    result = result [ : len ( result ) - 1 ]
    if result == name :
        return name
    else :
        raise ValueError ( "InvalidURL" )
except :
    raise ValueError ( "InvalidURL" )


signature:
def pickle ( obj , filepath ) :
    """Pickle and compress ."""

implementation:
arr = pkl . dumps ( obj , - 1 )
with open ( filepath , 'wb' ) as f :
    s = 0
    while s < len ( arr ) :
        e = min ( s + blosc . MAX_BUFFERSIZE , len ( arr ) )
        carr = blosc . compress ( arr [ s : e ] , typesize = 8 )
        f . write ( carr )
        s = e


signature:
def unpickle ( filepath ) :
    """Decompress and unpickle ."""

implementation:
arr = [ ]
with open ( filepath , 'rb' ) as f :
    carr = f . read ( blosc . MAX_BUFFERSIZE )
    while len ( carr ) > 0 :
        arr . append ( blosc . decompress ( carr ) )
        carr = f . read ( blosc . MAX_BUFFERSIZE )
return pkl . loads ( b"" . join ( arr ) )


signature:
def contact ( request ) :
    """Displays the contact form and sends the email"""

implementation:
form = ContactForm ( request . POST or None )
if form . is_valid ( ) :
    subject = form . cleaned_data [ 'subject' ]
    message = form . cleaned_data [ 'message' ]
    sender = form . cleaned_data [ 'sender' ]
    cc_myself = form . cleaned_data [ 'cc_myself' ]
    recipients = settings . CONTACTFORM_RECIPIENTS
    if cc_myself :
        recipients . append ( sender )
    send_mail ( getattr ( settings , "CONTACTFORM_SUBJECT_PREFIX" , '' ) + subject , message , sender , recipients )
    return render ( request , 'contactform/thanks.html' )
return render ( request , 'contactform/contact.html' , { 'form' : form } )


signature:
def load_gitconfig ( self ) :
    """try use gitconfig info .
 author , email etc ."""

implementation:
gitconfig_path = os . path . expanduser ( '~/.gitconfig' )
if os . path . exists ( gitconfig_path ) :
    parser = Parser ( )
    parser . read ( gitconfig_path )
    parser . sections ( )
    return parser
pass


signature:
def add_arguments ( cls ) :
    """Init project ."""

implementation: return [ ( ( '--yes' , ) , dict ( action = 'store_true' , help = 'clean .git repo' ) ) , ( ( '--variable' , '-s' ) , dict ( nargs = '+' , help = 'set extra variable,format is name:value' ) ) , ( ( '--skip-builtin' , ) , dict ( action = 'store_true' , help = 'skip replace builtin variable' ) ) , ]


signature:
def set_signal ( self ) :
    """设置信号处理
 默认直接中断 , 考虑到一些场景用户需要等待线程结束 .
 可在此自定义操作
 : return :"""

implementation:
def signal_handle ( signum , frame ) :
    self . error ( "User interrupt.kill threads." )
    sys . exit ( - 1 )
    pass
signal . signal ( signal . SIGINT , signal_handle )
pass


signature:
def check_exclusive_mode ( self ) :
    """检查是否是独占模式
 参数顺序必须一致 , 也就是说如果参数顺序不一致 , 则判定为是两个不同的进程
 这么设计是考虑到 :
 - 一般而言 , 排他模式的服务启动都是crontab等脚本来完成的 , 不存在顺序变更的可能
 - 这在调试的时候可以帮助我们不需要结束原有进程就可以继续调试
 : return :"""

implementation:
if self . options . exclusive_mode :
    import psutil
    current_pid = os . getpid ( )
    current = psutil . Process ( current_pid ) . cmdline ( )
    for pid in psutil . pids ( ) :
        p = psutil . Process ( pid )
        try :
            if current_pid != pid and current == p . cmdline ( ) :
                self . error_message ( "process exist. pid:{}" . format ( p . pid ) )
                sys . exit ( - 1 )
                pass
        except psutil . ZombieProcess : # 僵尸进程 , 无视
            pass
        except psutil . AccessDenied : # 不具备用户权限
            pass
        pass
    pass
pass


signature:
def run ( self , options ) :
    """In general , you don ' t need to overwrite this method .
 : param options :
 : return :"""

implementation:
self . set_signal ( )
self . check_exclusive_mode ( )
slot = self . Handle ( self )
# start thread
i = 0
while i < options . threads :
    t = threading . Thread ( target = self . worker , args = [ slot ] )
    # only set daemon when once is False
    if options . once is True or options . no_daemon is True :
        t . daemon = False
    else :
        t . daemon = True
    t . start ( )
    i += 1
# waiting thread
if options . once is False :
    while True :
        if threading . active_count ( ) > 1 :
            sleep ( 1 )
        else :
            if threading . current_thread ( ) . name == "MainThread" :
                sys . exit ( 0 )
pass


signature:
def serial_ports ( ) :
    '''Returns
 pandas . DataFrame
 Table of serial ports that match the USB vendor ID and product ID for
 the ` Teensy 3.2 ` _ board .
 . . Teensy 3.2 : https : / / www . pjrc . com / store / teensy32 . html'''

implementation:
df_comports = sd . comports ( )
# Match COM ports with USB vendor ID and product IDs for [ Teensy 3.2
# device ] [ 1 ] .
# [1 ] : https : / / www . pjrc . com / store / teensy32 . html
df_teensy_comports = df_comports . loc [ df_comports . hardware_id . str . contains ( 'VID:PID=16c0:0483' , case = False ) ]
return df_teensy_comports


signature:
def combine_filenames ( filenames , max_length = 40 ) :
    """Return a new filename to use as the combined file name for a
 bunch of files , based on the SHA of their contents .
 A precondition is that they all have the same file extension
 Given that the list of files can have different paths , we aim to use the
 most common path .
 Example :
 / somewhere / else / foo . js
 / somewhere / bar . js
 / somewhere / different / too / foobar . js
 The result will be
 / somewhere / 148713695b4a4b9083e506086f061f9c . js
 Another thing to note , if the filenames have timestamps in them , combine
 them all and use the highest timestamp ."""

implementation:
# Get the SHA for each file , then sha all the shas .
path = None
names = [ ]
extension = None
timestamps = [ ]
shas = [ ]
filenames . sort ( )
concat_names = "_" . join ( filenames )
if concat_names in COMBINED_FILENAMES_GENERATED :
    return COMBINED_FILENAMES_GENERATED [ concat_names ]
for filename in filenames :
    name = os . path . basename ( filename )
    if not extension :
        extension = os . path . splitext ( name ) [ 1 ]
    elif os . path . splitext ( name ) [ 1 ] != extension :
        raise ValueError ( "Can't combine multiple file extensions" )
    for base in MEDIA_ROOTS :
        try :
            shas . append ( md5 ( os . path . join ( base , filename ) ) )
            break
        except IOError :
            pass
    if path is None :
        path = os . path . dirname ( filename )
    else :
        if len ( os . path . dirname ( filename ) ) < len ( path ) :
            path = os . path . dirname ( filename )
m = hashlib . md5 ( )
m . update ( "," . join ( shas ) )
new_filename = "%s-inkmd" % m . hexdigest ( )
new_filename = new_filename [ : max_length ]
new_filename += extension
COMBINED_FILENAMES_GENERATED [ concat_names ] = new_filename
return os . path . join ( path , new_filename )


signature:
def apply_orientation ( im ) :
    """Extract the oritentation EXIF tag from the image , which should be a PIL Image instance ,
 and if there is an orientation tag that would rotate the image , apply that rotation to
 the Image instance given to do an in - place rotation .
 : param Image im : Image instance to inspect
 : return : A possibly transposed image instance"""

implementation:
try :
    kOrientationEXIFTag = 0x0112
    if hasattr ( im , '_getexif' ) : # only present in JPEGs
        e = im . _getexif ( )
        # returns None if no EXIF data
        if e is not None : # log . info ( ' EXIF data found : % r ' , e )
            orientation = e [ kOrientationEXIFTag ]
            f = orientation_funcs [ orientation ]
            return f ( im )
except : # We ' d be here with an invalid orientation value or some random error ?
    pass
    # log . exception ( " Error applying EXIF Orientation tag " )
return im


signature:
def write ( ) :
    """Start a new piece"""

implementation:
click . echo ( "Fantastic. Let's get started. " )
title = click . prompt ( "What's the title?" )
# Make sure that title doesn ' t exist .
url = slugify ( title )
url = click . prompt ( "What's the URL?" , default = url )
# Make sure that title doesn ' t exist .
click . echo ( "Got it. Creating %s..." % url )
scaffold_piece ( title , url )


signature:
def scaffold ( ) :
    """Start a new site ."""

implementation:
click . echo ( "A whole new site? Awesome." )
title = click . prompt ( "What's the title?" )
url = click . prompt ( "Great. What's url? http://" )
# Make sure that title doesn ' t exist .
click . echo ( "Got it. Creating %s..." % url )


signature:
def publish ( ) :
    """Publish the site"""

implementation:
try :
    build_site ( dev_mode = False , clean = True )
    click . echo ( 'Deploying the site...' )
    # call ( " firebase deploy " , shell = True )
    call ( "rsync -avz -e ssh --progress %s/ %s" % ( BUILD_DIR , CONFIG [ "scp_target" ] , ) , shell = True )
    if "cloudflare" in CONFIG and "purge" in CONFIG [ "cloudflare" ] and CONFIG [ "cloudflare" ] [ "purge" ] :
        do_purge ( )
except ( KeyboardInterrupt , SystemExit ) :
    raise
    sys . exit ( 1 )


signature:
def promote ( ) :
    """Schedule all the social media posts ."""

implementation:
if "BUFFER_ACCESS_TOKEN" not in os . environ :
    warn ( "Missing BUFFER_ACCESS_TOKEN." )
    echo ( "To publish to social medial, you'll need an access token for buffer." )
    echo ( "The simplest way to get one is to create a new app here: https://buffer.com/developers/apps" )
    echo ( "The token you want is the 'Access Token'" )
    echo ( "Once you have it, make it available to ink by putting it in the environment." )
# GET https : / / api . bufferapp . com / 1 / profiles . json
echo ( "Verifying available profiles on buffer" )
profiles = buffer_get ( "/1/profiles.json" )
for p in profiles :
    supported_profile = False
    if p [ "formatted_service" ] . lower ( ) == "facebook" or p [ "formatted_service" ] . lower ( ) == "facebook page" :
        facebook_profiles . append ( p )
        supported_profile = True
    elif p [ "formatted_service" ] . lower ( ) == "twitter" :
        twitter_profiles . append ( p )
        supported_profile = True
    if supported_profile :
        click . secho ( u"✓  %s: %s" % ( p [ "formatted_service" ] , p [ "formatted_username" ] ) , fg = "green" )
echo ( "Checking publication status..." )
site_json_filename = os . path . join ( ROOT_DIR , BUILD_DIR , "static" , "private.json" )
with open ( site_json_filename , "r" ) as site_json :
    site = load ( site_json )
echo ( 'Reviewing social posts...' )
posts = { }
unpublished_posts = [ ]
for dirpath , dirnames , filenames in os . walk ( os . path . join ( ROOT_DIR , "posts" ) , topdown = False ) :
    for filename in filenames :
        if "piece.md" in filename :
            if exists ( dirpath , "social.yml" ) and exists ( dirpath , "meta.yml" ) :
                with open ( os . path . join ( dirpath , "social.yml" ) ) as f :
                    social = load ( f )
                with open ( os . path . join ( dirpath , "meta.yml" ) ) as f :
                    meta = load ( f )
                if "url" in meta :
                    site_json_entry = None
                    for sp in site [ "posts" ] :
                        if meta [ "url" ] == sp [ "url" ] :
                            site_json_entry = sp
                            break
                    posts [ meta [ "url" ] ] = { "meta" : meta , "social" : social , "dirpath" : dirpath , "site" : site_json_entry , }
                    if "published" not in social or social [ "published" ] is not True :
                        unpublished_posts . append ( meta [ "url" ] )
                else :
                    warn ( "No url found for %s" % dirpath . replace ( ROOT_DIR ) )
automark_set = False
automark = None
for u in unpublished_posts :
    post = posts [ u ]
    if "posts" in post [ "social" ] and post [ "social" ] [ "posts" ] and len ( post [ "social" ] [ "posts" ] ) > 0 :
        facebook_posts = [ ]
        twitter_posts = [ ]
        mark_as_published = False
        has_valid_post = False
        for p in post [ "social" ] [ "posts" ] :
            try :
                if len ( p . keys ( ) ) != 1 :
                    error ( "Something's formatted wrong in %s's social.yml" % u )
                    break
                if p . keys ( ) [ 0 ] == "facebook" :
                    facebook_posts . append ( p [ "facebook" ] )
                    if post_in_future ( p [ "facebook" ] , post ) :
                        has_valid_post = True
                elif p . keys ( ) [ 0 ] == "twitter" :
                    if post_in_future ( p [ "twitter" ] , post ) :
                        has_valid_post = True
                    twitter_posts . append ( p [ "twitter" ] )
                else :
                    warn ( "Unknown post type: %s.  Skipping." % p . keys ( ) [ 0 ] )
            except :
                error ( "Error parsing social.yml for \"%s\"" % post [ "meta" ] [ "title" ] )
                import traceback
                traceback . print_exc ( )
        if not has_valid_post :
            if automark :
                mark_as_published = True
            else :
                warn ( '"%s" hasn\'t been published, but all posts are in the past.' % post [ "meta" ] [ "title" ] )
                if click . confirm ( "Mark as published?" ) :
                    mark_as_published = True
                    if not automark_set :
                        if click . confirm ( "Mark all other similar posts as published?" ) :
                            automark = True
                        automark_set = True
        else :
            echo ( '\n"%s" hasn\'t been published to social media.' % post [ "meta" ] [ "title" ] )
            if len ( facebook_posts ) > 0 :
                echo ( "  Facebook:" )
                for p in facebook_posts :
                    if ( len ( p [ "content" ] ) > 40 ) :
                        truncated_content = "%s..." % p [ "content" ] [ : 40 ]
                    else :
                        truncated_content = p [ "content" ]
                    if post_in_future ( p , post ) :
                        echo ( "   - %s:  \"%s\"" % ( publish_datetime ( p , post ) . strftime ( "%c" ) , truncated_content , ) )
                    else :
                        warn ( "   - %s:  \"%s\" skipping (past)" % ( publish_datetime ( p , post ) . strftime ( "%c" ) , truncated_content , ) )
            echo ( "  Twitter:" )
            if len ( twitter_posts ) > 0 :
                for p in twitter_posts :
                    if ( len ( p [ "content" ] ) > 40 ) :
                        truncated_content = "%s..." % p [ "content" ] [ : 40 ]
                    else :
                        truncated_content = p [ "content" ]
                    if post_in_future ( p , post ) :
                        echo ( "   - %s:  \"%s\"" % ( publish_datetime ( p , post ) . strftime ( "%c" ) , truncated_content , ) )
                    else :
                        warn ( "   - %s:  \"%s\" skipping (past)" % ( publish_datetime ( p , post ) . strftime ( "%c" ) , truncated_content , ) )
            if click . confirm ( click . style ( "  Publish now?" , fg = "green" ) ) :
                mark_as_published = True
                echo ( "  Publishing..." )
                for p in facebook_posts :
                    if post_in_future ( p , post ) :
                        publish_facebook ( p , post )
                        if ( len ( p [ "content" ] ) > 40 ) :
                            truncated_content = "%s..." % p [ "content" ] [ : 40 ]
                        else :
                            truncated_content = p [ "content" ]
                        click . secho ( u"   ✓ Facebook %s:  \"%s\"" % ( publish_datetime ( p , post ) . strftime ( "%c" ) , truncated_content , ) , fg = "green" )
                for p in twitter_posts :
                    if post_in_future ( p , post ) :
                        publish_twitter ( p , post )
                        if ( len ( p [ "content" ] ) > 40 ) :
                            truncated_content = "%s..." % p [ "content" ] [ : 40 ]
                        else :
                            truncated_content = p [ "content" ]
                        click . secho ( u"   ✓ Twitter %s:  \"%s\"" % ( publish_datetime ( p , post ) . strftime ( "%c" ) , truncated_content , ) , fg = "green" )
                echo ( "  Published." )
        # Save as published .
        if mark_as_published or automark :
            post [ "social" ] [ "published" ] = True
            with open ( os . path . join ( post [ "dirpath" ] , "social.yml" ) , "w" ) as f :
                dump ( post [ "social" ] , f , default_flow_style = False , width = 1000 )
if click . confirm ( "Publish your entire backlog to buffer?" ) :
    print ( "dope" )


signature:
def get_branches ( self ) :
    """Returns a list of the branches"""

implementation: return [ self . _sanitize ( branch ) for branch in self . _git . branch ( color = "never" ) . splitlines ( ) ]


signature:
def get_current_branch ( self ) :
    """Returns the currently active branch"""

implementation: return next ( ( self . _sanitize ( branch ) for branch in self . _git . branch ( color = "never" ) . splitlines ( ) if branch . startswith ( '*' ) ) , None )


signature:
def create_patch ( self , from_tag , to_tag ) :
    """Create a patch between tags"""

implementation: return str ( self . _git . diff ( '{}..{}' . format ( from_tag , to_tag ) , _tty_out = False ) )


signature:
def one ( func , n = 0 ) :
    """Create a callable that applies ` ` func ` ` to a value in a sequence .
 If the value is not a sequence or is an empty sequence then ` ` None ` ` is
 returned .
 : type func : ` callable `
 : param func : Callable to be applied to each result .
 : type n : ` int `
 : param n : Index of the value to apply ` ` func ` ` to ."""

implementation:
def _one ( result ) :
    if _isSequenceTypeNotText ( result ) and len ( result ) > n :
        return func ( result [ n ] )
    return None
return maybe ( _one )


signature:
def many ( func ) :
    """Create a callable that applies ` ` func ` ` to every value in a sequence .
 If the value is not a sequence then an empty list is returned .
 : type func : ` callable `
 : param func : Callable to be applied to the first result ."""

implementation:
def _many ( result ) :
    if _isSequenceTypeNotText ( result ) :
        return map ( func , result )
    return [ ]
return maybe ( _many , default = [ ] )


signature:
def Text ( value , encoding = None ) :
    """Parse a value as text .
 : type value : ` unicode ` or ` bytes `
 : param value : Text value to parse
 : type encoding : ` bytes `
 : param encoding : Encoding to treat ` ` bytes ` ` values as , defaults to
 ` ` utf - 8 ` ` .
 : rtype : ` unicode `
 : return : Parsed text or ` ` None ` ` if ` ` value ` ` is neither ` bytes ` nor
 ` unicode ` ."""

implementation:
if encoding is None :
    encoding = 'utf-8'
if isinstance ( value , bytes ) :
    return value . decode ( encoding )
elif isinstance ( value , unicode ) :
    return value
return None


signature:
def Integer ( value , base = 10 , encoding = None ) :
    """Parse a value as an integer .
 : type value : ` unicode ` or ` bytes `
 : param value : Text value to parse
 : type base : ` unicode ` or ` bytes `
 : param base : Base to assume ` ` value ` ` is specified in .
 : type encoding : ` bytes `
 : param encoding : Encoding to treat ` ` bytes ` ` values as , defaults to
 ` ` utf - 8 ` ` .
 : rtype : ` int `
 : return : Parsed integer or ` ` None ` ` if ` ` value ` ` could not be parsed as an
 integer ."""

implementation:
try :
    return int ( Text ( value , encoding ) , base )
except ( TypeError , ValueError ) :
    return None


signature:
def Boolean ( value , true = ( u'yes' , u'1' , u'true' ) , false = ( u'no' , u'0' , u'false' ) , encoding = None ) :
    """Parse a value as a boolean .
 : type value : ` unicode ` or ` bytes `
 : param value : Text value to parse .
 : type true : ` tuple ` of ` unicode `
 : param true : Values to compare , ignoring case , for ` ` True ` ` values .
 : type false : ` tuple ` of ` unicode `
 : param false : Values to compare , ignoring case , for ` ` False ` ` values .
 : type encoding : ` bytes `
 : param encoding : Encoding to treat ` bytes ` values as , defaults to
 ` ` utf - 8 ` ` .
 : rtype : ` bool `
 : return : Parsed boolean or ` ` None ` ` if ` ` value ` ` did not match ` ` true ` ` or
 ` ` false ` ` values ."""

implementation:
value = Text ( value , encoding )
if value is not None :
    value = value . lower ( ) . strip ( )
if value in true :
    return True
elif value in false :
    return False
return None


signature:
def Delimited ( value , parser = Text , delimiter = u',' , encoding = None ) :
    """Parse a value as a delimited list .
 : type value : ` unicode ` or ` bytes `
 : param value : Text value to parse .
 : type parser : ` callable ` taking a ` unicode ` parameter
 : param parser : Callable to map over the delimited text values .
 : type delimiter : ` unicode `
 : param delimiter : Delimiter text .
 : type encoding : ` bytes `
 : param encoding : Encoding to treat ` bytes ` values as , defaults to
 ` ` utf - 8 ` ` .
 : rtype : ` list `
 : return : List of parsed values ."""

implementation:
value = Text ( value , encoding )
if value is None or value == u'' :
    return [ ]
return map ( parser , value . split ( delimiter ) )


signature:
def Timestamp ( value , _divisor = 1. , tz = UTC , encoding = None ) :
    """Parse a value as a POSIX timestamp in seconds .
 : type value : ` unicode ` or ` bytes `
 : param value : Text value to parse , which should be the number of seconds
 since the epoch .
 : type _ divisor : ` float `
 : param _ divisor : Number to divide the value by .
 : type tz : ` tzinfo `
 : param tz : Timezone , defaults to UTC .
 : type encoding : ` bytes `
 : param encoding : Encoding to treat ` bytes ` values as , defaults to
 ` ` utf - 8 ` ` .
 : rtype : ` datetime . datetime `
 : return : Parsed datetime or ` ` None ` ` if ` ` value ` ` could not be parsed ."""

implementation:
value = Float ( value , encoding )
if value is not None :
    value = value / _divisor
    return datetime . fromtimestamp ( value , tz )
return None


signature:
def parse ( expected , query ) :
    """Parse query parameters .
 : type expected : ` dict ` mapping ` bytes ` to ` callable `
 : param expected : Mapping of query argument names to argument parsing
 callables .
 : type query : ` dict ` mapping ` bytes ` to ` list ` of ` bytes `
 : param query : Mapping of query argument names to lists of argument values ,
 this is the form that Twisted Web ' s ` IRequest . args
 < twisted : twisted . web . iweb . IRequest . args > ` value takes .
 : rtype : ` dict ` mapping ` bytes ` to ` object `
 : return : Mapping of query argument names to parsed argument values ."""

implementation: return dict ( ( key , parser ( query . get ( key , [ ] ) ) ) for key , parser in expected . items ( ) )


signature:
def put ( self , metrics ) :
    """Put metrics to cloudwatch . Metric shoult be instance or list of
 instances of CloudWatchMetric"""

implementation:
if type ( metrics ) == list :
    for metric in metrics :
        self . c . put_metric_data ( ** metric )
else :
    self . c . put_metric_data ( ** metrics )


signature:
def _renderResource ( resource , request ) :
    """Render a given resource .
 See ` IResource . render < twisted : twisted . web . resource . IResource . render > ` ."""

implementation:
meth = getattr ( resource , 'render_' + nativeString ( request . method ) , None )
if meth is None :
    try :
        allowedMethods = resource . allowedMethods
    except AttributeError :
        allowedMethods = _computeAllowedMethods ( resource )
    raise UnsupportedMethod ( allowedMethods )
return meth ( request )


signature:
def _adaptToResource ( self , result ) :
    """Adapt a result to ` IResource ` .
 Several adaptions are tried they are , in order : ` ` None ` ` ,
 ` IRenderable < twisted : twisted . web . iweb . IRenderable > ` , ` IResource
 < twisted : twisted . web . resource . IResource > ` , and ` URLPath
 < twisted : twisted . python . urlpath . URLPath > ` . Anything else is returned as
 is .
 A ` URLPath < twisted : twisted . python . urlpath . URLPath > ` is treated as
 a redirect ."""

implementation:
if result is None :
    return NotFound ( )
spinneretResource = ISpinneretResource ( result , None )
if spinneretResource is not None :
    return SpinneretResource ( spinneretResource )
renderable = IRenderable ( result , None )
if renderable is not None :
    return _RenderableResource ( renderable )
resource = IResource ( result , None )
if resource is not None :
    return resource
if isinstance ( result , URLPath ) :
    return Redirect ( str ( result ) )
return result


signature:
def _handleRenderResult ( self , request , result ) :
    """Handle the result from ` IResource . render ` .
 If the result is a ` Deferred ` then return ` NOT _ DONE _ YET ` and add
 a callback to write the result to the request when it arrives ."""

implementation:
def _requestFinished ( result , cancel ) :
    cancel ( )
    return result
if not isinstance ( result , Deferred ) :
    result = succeed ( result )
def _whenDone ( result ) :
    render = getattr ( result , 'render' , lambda request : result )
    renderResult = render ( request )
    if renderResult != NOT_DONE_YET :
        request . write ( renderResult )
        request . finish ( )
    return result
request . notifyFinish ( ) . addBoth ( _requestFinished , result . cancel )
result . addCallback ( self . _adaptToResource )
result . addCallback ( _whenDone )
result . addErrback ( request . processingFailed )
return NOT_DONE_YET


signature:
def _negotiateHandler ( self , request ) :
    """Negotiate a handler based on the content types acceptable to the
 client .
 : rtype : 2 - ` tuple ` of ` twisted . web . iweb . IResource ` and ` bytes `
 : return : Pair of a resource and the content type ."""

implementation:
accept = _parseAccept ( request . requestHeaders . getRawHeaders ( 'Accept' ) )
for contentType in accept . keys ( ) :
    handler = self . _acceptHandlers . get ( contentType . lower ( ) )
    if handler is not None :
        return handler , handler . contentType
if self . _fallback :
    handler = self . _handlers [ 0 ]
    return handler , handler . contentType
return NotAcceptable ( ) , None


signature:
def _parseAccept ( headers ) :
    """Parse and sort an ` ` Accept ` ` header .
 The header is sorted according to the ` ` q ` ` parameter for each header value .
 @ rtype : ` OrderedDict ` mapping ` bytes ` to ` dict `
 @ return : Mapping of media types to header parameters ."""

implementation:
def sort ( value ) :
    return float ( value [ 1 ] . get ( 'q' , 1 ) )
return OrderedDict ( sorted ( _splitHeaders ( headers ) , key = sort , reverse = True ) )


signature:
def _splitHeaders ( headers ) :
    """Split an HTTP header whose components are separated with commas .
 Each component is then split on semicolons and the component arguments
 converted into a ` dict ` .
 @ return : ` list ` of 2 - ` tuple ` of ` bytes ` , ` dict `
 @ return : List of header arguments and mapping of component argument names
 to values ."""

implementation: return [ cgi . parse_header ( value ) for value in chain . from_iterable ( s . split ( ',' ) for s in headers if s ) ]


signature:
def contentEncoding ( requestHeaders , encoding = None ) :
    """Extract an encoding from a ` ` Content - Type ` ` header .
 @ type requestHeaders : ` twisted . web . http _ headers . Headers `
 @ param requestHeaders : Request headers .
 @ type encoding : ` bytes `
 @ param encoding : Default encoding to assume if the ` ` Content - Type ` `
 header is lacking one . Defaults to ` ` UTF - 8 ` ` .
 @ rtype : ` bytes `
 @ return : Content encoding ."""

implementation:
if encoding is None :
    encoding = b'utf-8'
headers = _splitHeaders ( requestHeaders . getRawHeaders ( b'Content-Type' , [ ] ) )
if headers :
    return headers [ 0 ] [ 1 ] . get ( b'charset' , encoding )
return encoding


signature:
def maybe ( f , default = None ) :
    """Create a nil - safe callable decorator .
 If the wrapped callable receives ` ` None ` ` as its argument , it will return
 ` ` None ` ` immediately ."""

implementation:
@ wraps ( f )
def _maybe ( x , * a , ** kw ) :
    if x is None :
        return default
    return f ( x , * a , ** kw )
return _maybe


signature:
def settings ( path = None , with_path = None ) :
    """Get or set ` Settings . _ wrapped `
 : param str path : a python module file ,
 if user set it , write config to ` Settings . _ wrapped `
 : param str with _ path : search path
 : return : A instance of ` Settings `"""

implementation:
if path :
    Settings . bind ( path , with_path = with_path )
return Settings . _wrapped


signature:
def bind ( mod_path , with_path = None ) :
    """bind user variable to ` _ wrapped `
 . . note : :
 you don ' t need call this method by yourself .
 program will call it in ` cliez . parser . parse `
 . . expection : :
 if path is not correct , will cause an ` ImportError `
 : param str mod _ path : module path , * use dot style , ' mod . mod1 ' *
 : param str with _ path : add path to ` sys . path ` ,
 if path is file , use its parent .
 : return : A instance of ` Settings `"""

implementation:
if with_path :
    if os . path . isdir ( with_path ) :
        sys . path . insert ( 0 , with_path )
    else :
        sys . path . insert ( 0 , with_path . rsplit ( '/' , 2 ) [ 0 ] )
    pass
# raise ` ImportError ` mod _ path if not exist
mod = importlib . import_module ( mod_path )
settings = Settings ( )
for v in dir ( mod ) :
    if v [ 0 ] == '_' or type ( getattr ( mod , v ) ) . __name__ == 'module' :
        continue
    setattr ( settings , v , getattr ( mod , v ) )
    pass
Settings . _path = mod_path
Settings . _wrapped = settings
return settings


signature:
def get_version ( ) :
    """Get the version from version module without importing more than
 necessary ."""

implementation:
version_module_path = os . path . join ( os . path . dirname ( __file__ ) , "txspinneret" , "_version.py" )
# The version module contains a variable called _ _ version _ _
with open ( version_module_path ) as version_module :
    exec ( version_module . read ( ) )
return locals ( ) [ "__version__" ]


signature:
def send ( self , use_open_peers = True , queue = True , ** kw ) :
    """send a transaction immediately . Failed transactions are picked up by the TxBroadcaster
 : param ip : specific peer IP to send tx to
 : param port : port of specific peer
 : param use _ open _ peers : use Arky ' s broadcast method"""

implementation:
if not use_open_peers :
    ip = kw . get ( 'ip' )
    port = kw . get ( 'port' )
    peer = 'http://{}:{}' . format ( ip , port )
    res = arky . rest . POST . peer . transactions ( peer = peer , transactions = [ self . tx . tx ] )
else :
    res = arky . core . sendPayload ( self . tx . tx )
if self . tx . success != '0.0%' :
    self . tx . error = None
    self . tx . success = True
else :
    self . tx . error = res [ 'messages' ]
    self . tx . success = False
self . tx . tries += 1
self . tx . res = res
if queue :
    self . tx . send = True
self . __save ( )
return res


signature:
def check_confirmations_or_resend ( self , use_open_peers = False , ** kw ) :
    """check if a tx is confirmed , else resend it .
 : param use _ open _ peers : select random peers fro api / peers endpoint"""

implementation:
if self . confirmations ( ) == 0 :
    self . send ( use_open_peers , ** kw )


signature:
def command_list ( ) :
    """Get sub - command list
 . . note : :
 Don ' t use logger handle this function errors .
 Because the error should be a code error , not runtime error .
 : return : ` list ` matched sub - parser"""

implementation:
from cliez . conf import COMPONENT_ROOT
root = COMPONENT_ROOT
if root is None :
    sys . stderr . write ( "cliez.conf.COMPONENT_ROOT not set.\n" )
    sys . exit ( 2 )
    pass
if not os . path . exists ( root ) :
    sys . stderr . write ( "please set a valid path for `cliez.conf.COMPONENT_ROOT`\n" )
    sys . exit ( 2 )
    pass
try :
    path = os . listdir ( os . path . join ( root , 'components' ) )
    return [ f [ : - 3 ] for f in path if f . endswith ( '.py' ) and f != '__init__.py' ]
except FileNotFoundError :
    return [ ]


signature:
def append_arguments ( klass , sub_parsers , default_epilog , general_arguments ) :
    """Add class options to argparser options .
 : param cliez . component . Component klass : subclass of Component
 : param Namespace sub _ parsers :
 : param str default _ epilog : default _ epilog
 : param list general _ arguments : global options , defined by user
 : return : Namespace subparser"""

implementation:
entry_name = hump_to_underscore ( klass . __name__ ) . replace ( '_component' , '' )
# set sub command document
epilog = default_epilog if default_epilog else 'This tool generate by `cliez` ' 'https://www.github.com/wangwenpei/cliez'
sub_parser = sub_parsers . add_parser ( entry_name , help = klass . __doc__ , epilog = epilog )
sub_parser . description = klass . add_arguments . __doc__
# add slot arguments
if hasattr ( klass , 'add_slot_args' ) :
    slot_args = klass . add_slot_args ( ) or [ ]
    for v in slot_args :
        sub_parser . add_argument ( * v [ 0 ] , ** v [ 1 ] )
    sub_parser . description = klass . add_slot_args . __doc__
    pass
user_arguments = klass . add_arguments ( ) or [ ]
for v in user_arguments :
    sub_parser . add_argument ( * v [ 0 ] , ** v [ 1 ] )
if not klass . exclude_global_option :
    for v in general_arguments :
        sub_parser . add_argument ( * v [ 0 ] , ** v [ 1 ] )
return sub_parser


signature:
def parse ( parser , argv = None , settings_key = 'settings' , no_args_func = None ) :
    """parser cliez app
 : param argparse . ArgumentParser parser : an instance
 of argparse . ArgumentParser
 : param argv : argument list , default is ` sys . argv `
 : type argv : list or tuple
 : param str settings : settings option name ,
 default is settings .
 : param object no _ args _ func : a callable object . if no sub - parser matched ,
 parser will call it .
 : return : an instance of ` cliez . component . Component ` or its subclass"""

implementation:
argv = argv or sys . argv
commands = command_list ( )
if type ( argv ) not in [ list , tuple ] :
    raise TypeError ( "argv only can be list or tuple" )
# match sub - parser
if len ( argv ) >= 2 and argv [ 1 ] in commands :
    sub_parsers = parser . add_subparsers ( )
    class_name = argv [ 1 ] . capitalize ( ) + 'Component'
    from cliez . conf import ( COMPONENT_ROOT , LOGGING_CONFIG , EPILOG , GENERAL_ARGUMENTS )
    sys . path . insert ( 0 , os . path . dirname ( COMPONENT_ROOT ) )
    mod = importlib . import_module ( '{}.components.{}' . format ( os . path . basename ( COMPONENT_ROOT ) , argv [ 1 ] ) )
    # dynamic load component
    klass = getattr ( mod , class_name )
    sub_parser = append_arguments ( klass , sub_parsers , EPILOG , GENERAL_ARGUMENTS )
    options = parser . parse_args ( argv [ 1 : ] )
    settings = Settings . bind ( getattr ( options , settings_key ) ) if settings_key and hasattr ( options , settings_key ) else None
    obj = klass ( parser , sub_parser , options , settings )
    # init logger
    logger_level = logging . CRITICAL
    if hasattr ( options , 'verbose' ) :
        if options . verbose == 1 :
            logger_level = logging . ERROR
        elif options . verbose == 2 :
            logger_level = logging . WARNING
        elif options . verbose == 3 :
            logger_level = logging . INFO
            obj . logger . setLevel ( logging . INFO )
        pass
    if hasattr ( options , 'debug' ) and options . debug :
        logger_level = logging . DEBUG
        # http lib use a strange way to logging
        try :
            import http . client as http_client
            http_client . HTTPConnection . debuglevel = 1
        except Exception : # do nothing
            pass
        pass
    loggers = LOGGING_CONFIG [ 'loggers' ]
    for k , v in loggers . items ( ) :
        v . setdefault ( 'level' , logger_level )
        if logger_level in [ logging . INFO , logging . DEBUG ] :
            v [ 'handlers' ] = [ 'stdout' ]
        pass
    logging_config . dictConfig ( LOGGING_CONFIG )
    # this may not necessary
    # obj . logger . setLevel ( logger _ level )
    obj . run ( options )
    # return object to make unit test easy
    return obj
# print all sub commands when user set .
if not parser . description and len ( commands ) :
    sub_parsers = parser . add_subparsers ( )
    [ sub_parsers . add_parser ( v ) for v in commands ]
    pass
pass
options = parser . parse_args ( argv [ 1 : ] )
if no_args_func and callable ( no_args_func ) :
    return no_args_func ( options )
else :
    parser . _print_message ( "nothing to do...\n" )
pass


signature:
def include_file ( filename , global_vars = None , local_vars = None ) :
    """. . deprecated 2.1 : :
 Don ' t use this any more .
 It ' s not pythonic .
 include file like php include .
 include is very useful when we need to split large config file"""

implementation:
if global_vars is None :
    global_vars = sys . _getframe ( 1 ) . f_globals
if local_vars is None :
    local_vars = sys . _getframe ( 1 ) . f_locals
with open ( filename , 'r' ) as f :
    code = compile ( f . read ( ) , os . path . basename ( filename ) , 'exec' )
    exec ( code , global_vars , local_vars )
    pass


signature:
def hump_to_underscore ( name ) :
    """Convert Hump style to underscore
 : param name : Hump Character
 : return : str"""

implementation:
new_name = ''
pos = 0
for c in name :
    if pos == 0 :
        new_name = c . lower ( )
    elif 65 <= ord ( c ) <= 90 :
        new_name += '_' + c . lower ( )
        pass
    else :
        new_name += c
    pos += 1
    pass
return new_name


signature:
def get_fuel_prices ( self ) -> GetFuelPricesResponse :
    """Fetches fuel prices for all stations ."""

implementation:
response = requests . get ( '{}/prices' . format ( API_URL_BASE ) , headers = self . _get_headers ( ) , timeout = self . _timeout , )
if not response . ok :
    raise FuelCheckError . create ( response )
return GetFuelPricesResponse . deserialize ( response . json ( ) )


signature:
def get_fuel_prices_for_station ( self , station : int ) -> List [ Price ] :
    """Gets the fuel prices for a specific fuel station ."""

implementation:
response = requests . get ( '{}/prices/station/{}' . format ( API_URL_BASE , station ) , headers = self . _get_headers ( ) , timeout = self . _timeout , )
if not response . ok :
    raise FuelCheckError . create ( response )
data = response . json ( )
return [ Price . deserialize ( data ) for data in data [ 'prices' ] ]


signature:
def get_fuel_prices_within_radius ( self , latitude : float , longitude : float , radius : int , fuel_type : str , brands : Optional [ List [ str ] ] = None ) -> List [ StationPrice ] :
    """Gets all the fuel prices within the specified radius ."""

implementation:
if brands is None :
    brands = [ ]
response = requests . post ( '{}/prices/nearby' . format ( API_URL_BASE ) , json = { 'fueltype' : fuel_type , 'latitude' : latitude , 'longitude' : longitude , 'radius' : radius , 'brand' : brands , } , headers = self . _get_headers ( ) , timeout = self . _timeout , )
if not response . ok :
    raise FuelCheckError . create ( response )
data = response . json ( )
stations = { station [ 'code' ] : Station . deserialize ( station ) for station in data [ 'stations' ] }
station_prices = [ ]
# type : List [ StationPrice ]
for serialized_price in data [ 'prices' ] :
    price = Price . deserialize ( serialized_price )
    station_prices . append ( StationPrice ( price = price , station = stations [ price . station_code ] ) )
return station_prices


signature:
def get_fuel_price_trends ( self , latitude : float , longitude : float , fuel_types : List [ str ] ) -> PriceTrends :
    """Gets the fuel price trends for the given location and fuel types ."""

implementation:
response = requests . post ( '{}/prices/trends/' . format ( API_URL_BASE ) , json = { 'location' : { 'latitude' : latitude , 'longitude' : longitude , } , 'fueltypes' : [ { 'code' : type } for type in fuel_types ] , } , headers = self . _get_headers ( ) , timeout = self . _timeout , )
if not response . ok :
    raise FuelCheckError . create ( response )
data = response . json ( )
return PriceTrends ( variances = [ Variance . deserialize ( variance ) for variance in data [ 'Variances' ] ] , average_prices = [ AveragePrice . deserialize ( avg_price ) for avg_price in data [ 'AveragePrices' ] ] )


signature:
def get_reference_data ( self , modified_since : Optional [ datetime . datetime ] = None ) -> GetReferenceDataResponse :
    """Fetches API reference data .
 : param modified _ since : The response will be empty if no
 changes have been made to the reference data since this
 timestamp , otherwise all reference data will be returned ."""

implementation:
if modified_since is None :
    modified_since = datetime . datetime ( year = 2010 , month = 1 , day = 1 )
response = requests . get ( '{}/lovs' . format ( API_URL_BASE ) , headers = { 'if-modified-since' : self . _format_dt ( modified_since ) , ** self . _get_headers ( ) , } , timeout = self . _timeout , )
if not response . ok :
    raise FuelCheckError . create ( response )
# return response . text
return GetReferenceDataResponse . deserialize ( response . json ( ) )


signature:
def parse_require ( self , env , keys , defaults = { } ) :
    """check and get require config
 : param dict env : user config node
 : param list keys : check keys
 . . note : :
 option and key name must be same .
 : param dict defaults : default value for keys
 : return : dict . env with verified .
 . . exception : :
 will raise ` ValueError ` when some key missed ."""

implementation:
for k in keys :
    env [ k ] = getattr ( self . options , k ) or env . get ( k , None )
    if env [ k ] is None :
        self . error ( "config syntax error," "please set `%s` in your env: %s" % ( k , env ) )
    pass
for k , v in defaults . items ( ) :
    env . setdefault ( k , v )
return env


signature:
def pre ( self , command , output_dir , vars ) :
    """Called before template is applied ."""

implementation:
# import pdb ; pdb . set _ trace ( )
vars [ 'license_name' ] = 'Apache'
vars [ 'year' ] = time . strftime ( '%Y' , time . localtime ( ) )


signature:
def Text ( name , encoding = None ) :
    """Match a route parameter .
 ` Any ` is a synonym for ` Text ` .
 : type name : ` bytes `
 : param name : Route parameter name .
 : type encoding : ` bytes `
 : param encoding : Default encoding to assume if the ` ` Content - Type ` `
 header is lacking one .
 : return : ` ` callable ` ` suitable for use with ` route ` or ` subroute ` ."""

implementation:
def _match ( request , value ) :
    return name , query . Text ( value , encoding = contentEncoding ( request . requestHeaders , encoding ) )
return _match


signature:
def Integer ( name , base = 10 , encoding = None ) :
    """Match an integer route parameter .
 : type name : ` bytes `
 : param name : Route parameter name .
 : type base : ` int `
 : param base : Base to interpret the value in .
 : type encoding : ` bytes `
 : param encoding : Default encoding to assume if the ` ` Content - Type ` `
 header is lacking one .
 : return : ` ` callable ` ` suitable for use with ` route ` or ` subroute ` ."""

implementation:
def _match ( request , value ) :
    return name , query . Integer ( value , base = base , encoding = contentEncoding ( request . requestHeaders , encoding ) )
return _match


signature:
def _matchRoute ( components , request , segments , partialMatching ) :
    """Match a request path against our path components .
 The path components are always matched relative to their parent is in the
 resource hierarchy , in other words it is only possible to match URIs nested
 more deeply than the parent resource .
 : type components : ` ` iterable ` ` of ` bytes ` or ` callable `
 : param components : Iterable of path components , to match against the
 request , either static strings or dynamic parameters . As a convenience ,
 a single ` bytes ` component containing ` ` / ` ` may be given instead of
 manually separating the components . If no components are given the null
 route is matched , this is the case where ` ` segments ` ` is empty .
 : type segments : ` ` sequence ` ` of ` bytes `
 : param segments : Sequence of path segments , from the request , to match
 against .
 : type partialMatching : ` bool `
 : param partialMatching : Allow partial matching against the request path ?
 : rtype : 2 - ` tuple ` of ` dict ` keyed on ` bytes ` and ` list ` of ` bytes `
 : return : Pair of parameter results , mapping parameter names to processed
 values , and a list of the remaining request path segments . If there is
 no route match the result will be ` ` None ` ` and the original request path
 segments ."""

implementation:
if len ( components ) == 1 and isinstance ( components [ 0 ] , bytes ) :
    components = components [ 0 ]
    if components [ : 1 ] == '/' :
        components = components [ 1 : ]
    components = components . split ( '/' )
results = OrderedDict ( )
NO_MATCH = None , segments
remaining = list ( segments )
# Handle the null route .
if len ( segments ) == len ( components ) == 0 :
    return results , remaining
for us , them in izip_longest ( components , segments ) :
    if us is None :
        if partialMatching : # We ' ve matched all of our components , there might be more
        # segments for something else to process .
            break
        else :
            return NO_MATCH
    elif them is None : # We ' ve run out of path segments to match , so this route can ' t be
    # the matching one .
        return NO_MATCH
    if callable ( us ) :
        name , match = us ( request , them )
        if match is None :
            return NO_MATCH
        results [ name ] = match
    elif us != them :
        return NO_MATCH
    remaining . pop ( 0 )
return results , remaining


signature:
def routedResource ( f , routerAttribute = 'router' ) :
    """Decorate a router - producing callable to instead produce a resource .
 This simply produces a new callable that invokes the original callable , and
 calls ` ` resource ` ` on the ` ` routerAttribute ` ` .
 If the router producer has multiple routers the attribute can be altered to
 choose the appropriate one , for example :
 . . code - block : : python
 class _ ComplexRouter ( object ) :
 router = Router ( )
 privateRouter = Router ( )
 @ router . route ( ' / ' )
 def publicRoot ( self , request , params ) :
 return SomethingPublic ( . . . )
 @ privateRouter . route ( ' / ' )
 def privateRoot ( self , request , params ) :
 return SomethingPrivate ( . . . )
 PublicResource = routedResource ( _ ComplexRouter )
 PrivateResource = routedResource ( _ ComplexRouter , ' privateRouter ' )
 : type f : ` ` callable ` `
 : param f : Callable producing an object with a ` Router ` attribute , for
 example , a type .
 : type routerAttribute : ` str `
 : param routerAttribute : Name of the ` Router ` attribute on the result of
 calling ` ` f ` ` .
 : rtype : ` callable `
 : return : Callable producing an ` IResource ` ."""

implementation: return wraps ( f ) ( lambda * a , ** kw : getattr ( f ( * a , ** kw ) , routerAttribute ) . resource ( ) )


signature:
def _forObject ( self , obj ) :
    """Create a new ` Router ` instance , with it ' s own set of routes , for
 ` ` obj ` ` ."""

implementation:
router = type ( self ) ( )
router . _routes = list ( self . _routes )
router . _self = obj
return router


signature:
def _addRoute ( self , f , matcher ) :
    """Add a route handler and matcher to the collection of possible routes ."""

implementation: self . _routes . append ( ( f . func_name , f , matcher ) )


signature:
def route ( self , * components ) :
    """See ` txspinneret . route . route ` .
 This decorator can be stacked with itself to specify multiple routes
 with a single handler ."""

implementation:
def _factory ( f ) :
    self . _addRoute ( f , route ( * components ) )
    return f
return _factory


signature:
def subroute ( self , * components ) :
    """See ` txspinneret . route . subroute ` .
 This decorator can be stacked with itself to specify multiple routes
 with a single handler ."""

implementation:
def _factory ( f ) :
    self . _addRoute ( f , subroute ( * components ) )
    return f
return _factory


signature:
def _tempfile ( filename ) :
    """Create a NamedTemporaryFile instance to be passed to atomic _ writer"""

implementation: return tempfile . NamedTemporaryFile ( mode = 'w' , dir = os . path . dirname ( filename ) , prefix = os . path . basename ( filename ) , suffix = os . fsencode ( '.tmp' ) , delete = False )


signature:
def atomic_write ( filename ) :
    """Open a NamedTemoraryFile handle in a context manager"""

implementation:
f = _tempfile ( os . fsencode ( filename ) )
try :
    yield f
finally :
    f . close ( )
    # replace the original file with the new temp file ( atomic on success )
    os . replace ( f . name , filename )


signature:
def get_item ( filename , uuid ) :
    """Read entry from JSON file"""

implementation:
with open ( os . fsencode ( str ( filename ) ) , "r" ) as f :
    data = json . load ( f )
    results = [ i for i in data if i [ "uuid" ] == str ( uuid ) ]
    if results :
        return results
    return None


signature:
def set_item ( filename , item ) :
    """Save entry to JSON file"""

implementation:
with atomic_write ( os . fsencode ( str ( filename ) ) ) as temp_file :
    with open ( os . fsencode ( str ( filename ) ) ) as products_file : # load the JSON data into memory
        products_data = json . load ( products_file )
    # check if UUID already exists
    uuid_list = [ i for i in filter ( lambda z : z [ "uuid" ] == str ( item [ "uuid" ] ) , products_data ) ]
    if len ( uuid_list ) == 0 : # add the new item to the JSON file
        products_data . append ( item )
        # save the new JSON to the temp file
        json . dump ( products_data , temp_file )
        return True
    return None


signature:
def update_item ( filename , item , uuid ) :
    """Update entry by UUID in the JSON file"""

implementation:
with atomic_write ( os . fsencode ( str ( filename ) ) ) as temp_file :
    with open ( os . fsencode ( str ( filename ) ) ) as products_file : # load the JSON data into memory
        products_data = json . load ( products_file )
    # apply modifications to the JSON data wrt UUID
    # TODO : handle this in a neat way
    if 'products' in products_data [ - 1 ] : # handle orders object
        [ products_data [ i ] [ "products" ] [ 0 ] . update ( item ) for ( i , j ) in enumerate ( products_data ) if j [ "uuid" ] == str ( uuid ) ]
    else : # handle products object
        [ products_data [ i ] . update ( item ) for ( i , j ) in enumerate ( products_data ) if j [ "uuid" ] == str ( uuid ) ]
    # save the modified JSON data into the temp file
    json . dump ( products_data , temp_file )
    return True


signature:
def run ( self , options ) :
    """. . todo : :
 check network connection
 : param Namespace options : parse result from argparse
 : return :"""

implementation:
self . logger . debug ( "debug enabled..." )
depends = [ 'git' ]
nil_tools = [ ]
self . logger . info ( "depends list: %s" , depends )
for v in depends :
    real_path = shutil . which ( v )
    if real_path :
        self . print_message ( "Found {}:{}..." "    {}" . format ( v , real_path , termcolor . colored ( '[OK]' , color = 'blue' ) ) )
    else :
        nil_tools . append ( v )
        self . error_message ( 'Missing tool:`{}`...    {}' . format ( v , '[ERR]' ) , prefix = '' , suffix = '' )
    pass
if nil_tools :
    self . print_message ( '' )
    self . error ( "please install missing tools..." )
else :
    self . print_message ( "\nNo error found," "you can use cliez in right way." )
    self . logger . debug ( "check finished..." )
    pass
pass


signature:
def experiment_data ( self , commit = None , must_contain_results = False ) :
    """: param commit : the commit that all the experiments should have happened or None to include all
 : type commit : str
 : param must _ contain _ results : include only tags that contain results
 : type must _ contain _ results : bool
 : return : all the experiment data
 : rtype : dict"""

implementation:
results = { }
for tag in self . __repository . tags :
    if not tag . name . startswith ( self . __tag_prefix ) :
        continue
    data = json . loads ( tag . tag . message )
    if "results" not in data and must_contain_results :
        continue
    if commit is not None and tag . tag . object . hexsha != name_to_object ( self . __repository , commit ) . hexsha :
        continue
    results [ tag . name ] = data
return results


signature:
def delete ( self , experiment_name ) :
    """Delete an experiment by removing the associated tag .
 : param experiment _ name : the name of the experiment to be deleted
 : type experiment _ name : str
 : rtype bool
 : return if deleting succeeded"""

implementation:
if not experiment_name . startswith ( self . __tag_prefix ) :
    target_tag = self . __tag_prefix + experiment_name
else :
    target_tag = experiment_name
if target_tag not in [ t . name for t in self . __repository . tags ] :
    return False
self . __repository . delete_tag ( target_tag )
return target_tag not in [ t . name for t in self . __repository . tags ]


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

implementation:
plugin = Register ( )
if plugin . args . option == 'command' :
    plugin . command_handle ( )
else :
    plugin . unknown ( "Unknown actions." )


signature:
def command_handle ( self ) :
    """Get the number of the shell command ."""

implementation:
self . __results = self . execute ( self . args . command )
self . close ( )
self . logger . debug ( "results: {}" . format ( self . __results ) )
if not self . __results :
    self . unknown ( "{} return nothing." . format ( self . args . command ) )
if len ( self . __results ) != 1 :
    self . unknown ( "{} return more than one number." . format ( self . args . command ) )
self . __result = int ( self . __results [ 0 ] )
self . logger . debug ( "result: {}" . format ( self . __result ) )
if not isinstance ( self . __result , ( int , long ) ) :
    self . unknown ( "{} didn't return single number." . format ( self . args . command ) )
status = self . ok
# Compare the vlaue .
if self . __result > self . args . warning :
    status = self . warning
if self . __result > self . args . critical :
    status = self . critical
# Output
self . shortoutput = "{0} return {1}." . format ( self . args . command , self . __result )
[ self . longoutput . append ( line ) for line in self . __results if self . __results ]
self . perfdata . append ( "{command}={result};{warn};{crit};0;" . format ( crit = self . args . critical , warn = self . args . warning , result = self . __result , command = self . args . command ) )
# Return status with message to Nagios .
status ( self . output ( long_output_limit = None ) )
self . logger . debug ( "Return status and exit to Nagios." )


signature:
def execute ( self , command , timeout = None ) :
    """Execute a shell command ."""

implementation:
try :
    self . channel = self . ssh . get_transport ( ) . open_session ( )
except paramiko . SSHException as e :
    self . unknown ( "Create channel error: %s" % e )
try :
    self . channel . settimeout ( self . args . timeout if not timeout else timeout )
except socket . timeout as e :
    self . unknown ( "Settimeout for channel error: %s" % e )
try :
    self . logger . debug ( "command: {}" . format ( command ) )
    self . channel . exec_command ( command )
except paramiko . SSHException as e :
    self . unknown ( "Execute command error: %s" % e )
try :
    self . stdin = self . channel . makefile ( 'wb' , - 1 )
    self . stderr = map ( string . strip , self . channel . makefile_stderr ( 'rb' , - 1 ) . readlines ( ) )
    self . stdout = map ( string . strip , self . channel . makefile ( 'rb' , - 1 ) . readlines ( ) )
except Exception as e :
    self . unknown ( "Get result error: %s" % e )
try :
    self . status = self . channel . recv_exit_status ( )
except paramiko . SSHException as e :
    self . unknown ( "Get return code error: %s" % e )
else :
    if self . status != 0 :
        self . unknown ( "Return code: %d , stderr: %s" % ( self . status , self . errors ) )
    else :
        return self . stdout
finally :
    self . logger . debug ( "Execute command finish." )


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

implementation:
try :
    self . ssh . close ( )
    self . logger . debug ( "close connect succeed." )
except paramiko . SSHException as e :
    self . unknown ( "close connect error: %s" % e )


signature:
def slinky ( filename , seconds_available , bucket_name , aws_key , aws_secret ) :
    """Simple program that creates an temp S3 link ."""

implementation:
if not os . environ . get ( 'AWS_ACCESS_KEY_ID' ) and os . environ . get ( 'AWS_SECRET_ACCESS_KEY' ) :
    print 'Need to set environment variables for AWS access and create a slinky bucket.'
    exit ( )
print create_temp_s3_link ( filename , seconds_available , bucket_name )


signature:
def check_readable ( self , timeout ) :
    """Poll ` ` self . stdout ` ` and return True if it is readable .
 : param float timeout : seconds to wait I / O
 : return : True if readable , else False
 : rtype : boolean"""

implementation:
rlist , wlist , xlist = select . select ( [ self . _stdout ] , [ ] , [ ] , timeout )
return bool ( len ( rlist ) )


signature:
def get_indices_list ( s : Any ) -> List [ str ] :
    """Retrieve a list of characters and escape codes where each escape
 code uses only one index . The indexes will not match up with the
 indexes in the original string ."""

implementation:
indices = get_indices ( s )
return [ indices [ i ] for i in sorted ( indices , key = int ) ]


signature:
def strip_codes ( s : Any ) -> str :
    """Strip all color codes from a string .
 Returns empty string for " falsey " inputs ."""

implementation: return codepat . sub ( '' , str ( s ) if ( s or ( s == 0 ) ) else '' )


signature:
def init_build ( self , asset , builder ) :
    """Called when builder group collect files
 Resolves absolute url if relative passed
 : type asset : static _ bundle . builders . Asset
 : type builder : static _ bundle . builders . StandardBuilder"""

implementation:
if not self . abs_path :
    rel_path = utils . prepare_path ( self . rel_bundle_path )
    self . abs_bundle_path = utils . prepare_path ( [ builder . config . input_dir , rel_path ] )
    self . abs_path = True
self . input_dir = builder . config . input_dir


signature:
def add_file ( self , * args ) :
    """Add single file or list of files to bundle
 : type : file _ path : str | unicode"""

implementation:
for file_path in args :
    self . files . append ( FilePath ( file_path , self ) )


signature:
def add_directory ( self , * args , ** kwargs ) :
    """Add directory or directories list to bundle
 : param exclusions : List of excluded paths
 : type path : str | unicode
 : type exclusions : list"""

implementation:
exc = kwargs . get ( 'exclusions' , None )
for path in args :
    self . files . append ( DirectoryPath ( path , self , exclusions = exc ) )


signature:
def add_path_object ( self , * args ) :
    """Add custom path objects
 : type : path _ object : static _ bundle . paths . AbstractPath"""

implementation:
for obj in args :
    obj . bundle = self
    self . files . append ( obj )


signature:
def add_prepare_handler ( self , prepare_handlers ) :
    """Add prepare handler to bundle
 : type : prepare _ handler : static _ bundle . handlers . AbstractPrepareHandler"""

implementation:
if not isinstance ( prepare_handlers , static_bundle . BUNDLE_ITERABLE_TYPES ) :
    prepare_handlers = [ prepare_handlers ]
if self . prepare_handlers_chain is None :
    self . prepare_handlers_chain = [ ]
for handler in prepare_handlers :
    self . prepare_handlers_chain . append ( handler )


signature:
def prepare ( self ) :
    """Called when builder run collect files in builder group
 : rtype : list [ static _ bundle . files . StaticFileResult ]"""

implementation:
result_files = self . collect_files ( )
chain = self . prepare_handlers_chain
if chain is None : # default handlers
    chain = [ LessCompilerPrepareHandler ( ) ]
for prepare_handler in chain :
    result_files = prepare_handler . prepare ( result_files , self )
return result_files


implementation:
plugin = Register ( )
if plugin . args . option == 'filenumber' :
    plugin . filenumber_handle ( )
else :
    plugin . unknown ( "Unknown actions." )


signature:
def filenumber_handle ( self ) :
    """Get the number of files in the folder ."""

implementation:
self . __results = [ ]
self . __dirs = [ ]
self . __files = [ ]
self . __ftp = self . connect ( )
self . __ftp . dir ( self . args . path , self . __results . append )
self . logger . debug ( "dir results: {}" . format ( self . __results ) )
self . quit ( )
status = self . ok
for data in self . __results :
    if "<DIR>" in data :
        self . __dirs . append ( str ( data . split ( ) [ 3 ] ) )
    else :
        self . __files . append ( str ( data . split ( ) [ 2 ] ) )
self . __result = len ( self . __files )
self . logger . debug ( "result: {}" . format ( self . __result ) )
# Compare the vlaue .
if self . __result > self . args . warning :
    status = self . warning
if self . __result > self . args . critical :
    status = self . critical
# Output
self . shortoutput = "Found {0} files in {1}." . format ( self . __result , self . args . path )
[ self . longoutput . append ( line ) for line in self . __results if self . __results ]
self . perfdata . append ( "{path}={result};{warn};{crit};0;" . format ( crit = self . args . critical , warn = self . args . warning , result = self . __result , path = self . args . path ) )
self . logger . debug ( "Return status and output." )
status ( self . output ( ) )


signature:
def register_json ( self , data ) :
    """Register the contents as JSON"""

implementation:
j = json . loads ( data )
self . last_data_timestamp = datetime . datetime . utcnow ( ) . replace ( microsecond = 0 ) . isoformat ( )
try :
    for v in j : # prepare the sensor entry container
        self . data [ v [ self . id_key ] ] = { }
        # add the mandatory entries
        self . data [ v [ self . id_key ] ] [ self . id_key ] = v [ self . id_key ]
        self . data [ v [ self . id_key ] ] [ self . value_key ] = v [ self . value_key ]
        # add the optional well known entries if provided
        if self . unit_key in v :
            self . data [ v [ self . id_key ] ] [ self . unit_key ] = v [ self . unit_key ]
        if self . threshold_key in v :
            self . data [ v [ self . id_key ] ] [ self . threshold_key ] = v [ self . threshold_key ]
        # add any further entries found
        for k in self . other_keys :
            if k in v :
                self . data [ v [ self . id_key ] ] [ k ] = v [ k ]
        # add the custom sensor time
        if self . sensor_time_key in v :
            self . data [ v [ self . sensor_time_key ] ] [ self . sensor_time_key ] = v [ self . sensor_time_key ]
        # last : add the time the data was received ( overwriting any
        # not properly defined timestamp that was already there )
        self . data [ v [ self . id_key ] ] [ self . time_key ] = self . last_data_timestamp
except KeyError as e :
    print ( "The main key was not found on the serial input line: " + str ( e ) )
except ValueError as e :
    print ( "No valid JSON string received. Waiting for the next turn." )
    print ( "The error was: " + str ( e ) )


signature:
def get_text ( self ) :
    """Get the data in text form ( i . e . human readable )"""

implementation:
t = "==== " + str ( self . last_data_timestamp ) + " ====\n"
for k in self . data :
    t += k + " " + str ( self . data [ k ] [ self . value_key ] )
    u = ""
    if self . unit_key in self . data [ k ] :
        u = self . data [ k ] [ self . unit_key ]
        t += u
    if self . threshold_key in self . data [ k ] :
        if ( self . data [ k ] [ self . threshold_key ] < self . data [ k ] [ self . value_key ] ) :
            t += " !Warning: Value is over threshold: " + str ( self . data [ k ] [ self . threshold_key ] ) + "!"
        else :
            t += " (" + str ( self . data [ k ] [ self . threshold_key ] ) + u + ")"
    for l in self . other_keys :
        if l in self . data [ k ] :
            t += " " + self . data [ k ] [ l ]
    t += "\n"
return t


signature:
def get_translated_data ( self ) :
    """Translate the data with the translation table"""

implementation:
j = { }
for k in self . data :
    d = { }
    for l in self . data [ k ] :
        d [ self . translation_keys [ l ] ] = self . data [ k ] [ l ]
    j [ k ] = d
return j


signature:
def get_json ( self , prettyprint = False , translate = True ) :
    """Get the data in JSON form"""

implementation:
j = [ ]
if translate :
    d = self . get_translated_data ( )
else :
    d = self . data
for k in d :
    j . append ( d [ k ] )
if prettyprint :
    j = json . dumps ( j , indent = 2 , separators = ( ',' , ': ' ) )
else :
    j = json . dumps ( j )
return j


signature:
def get_json_tuples ( self , prettyprint = False , translate = True ) :
    """Get the data as JSON tuples"""

implementation:
j = self . get_json ( prettyprint , translate )
if len ( j ) > 2 :
    if prettyprint :
        j = j [ 1 : - 2 ] + ",\n"
    else :
        j = j [ 1 : - 1 ] + ","
else :
    j = ""
return j


signature:
def get ( self , url , params = { } ) :
    """Issues a GET request against the API , properly formatting the params
 : param url : a string , the url you are requesting
 : param params : a dict , the key - value of all the paramaters needed
 in the request
 : returns : a dict parsed of the JSON response"""

implementation:
params . update ( { 'api_key' : self . api_key } )
try :
    response = requests . get ( self . host + url , params = params )
except RequestException as e :
    response = e . args
return self . json_parse ( response . content )


signature:
def post ( self , url , params = { } , files = None ) :
    """Issues a POST request against the API , allows for multipart data uploads
 : param url : a string , the url you are requesting
 : param params : a dict , the key - value of all the parameters needed
 in the request
 : param files : a list , the list of tuples of files
 : returns : a dict parsed of the JSON response"""

implementation:
params . update ( { 'api_key' : self . api_key } )
try :
    response = requests . post ( self . host + url , data = params , files = files )
    return self . json_parse ( response . content )
except RequestException as e :
    return self . json_parse ( e . args )


signature:
def json_parse ( self , content ) :
    """Wraps and abstracts content validation and JSON parsing
 to make sure the user gets the correct response .
 : param content : The content returned from the web request to be parsed as json
 : returns : a dict of the json response"""

implementation:
try :
    data = json . loads ( content )
except ValueError , e :
    return { 'meta' : { 'status' : 500 , 'msg' : 'Server Error' } , 'response' : { "error" : "Malformed JSON or HTML was returned." } }
# We only really care about the response if we succeed
# and the error if we fail
if 'error' in data :
    return { 'meta' : { 'status' : 400 , 'msg' : 'Bad Request' } , 'response' : { "error" : data [ 'error' ] } }
elif 'result' in data :
    return data [ 'result' ]
else :
    return { }


signature:
def load_values ( self ) :
    """Go through the env var map , transferring the values to this object
 as attributes .
 : raises : RuntimeError if a required env var isn ' t defined ."""

implementation:
for config_name , evar in self . evar_defs . items ( ) :
    if evar . is_required and evar . name not in os . environ :
        raise RuntimeError ( ( "Missing required environment variable: {evar_name}\n" "{help_txt}" ) . format ( evar_name = evar . name , help_txt = evar . help_txt ) )
    # Env var is present . Transfer its value over .
    if evar . name in os . environ :
        self [ config_name ] = os . environ . get ( evar . name )
    else :
        self [ config_name ] = evar . default_val
    # Perform any validations or transformations .
    for filter in evar . filters :
        current_val = self . get ( config_name )
        new_val = filter ( current_val , evar )
        self [ config_name ] = new_val
# This is the top - level filter that is often useful for checking
# the values of related env vars ( instead of individual validation ) .
self . _filter_all ( )