signature:
def average_data ( self , ranges = [ [ None , None ] ] , percentile = None ) :
    """given a list of ranges , return single point averages for every sweep .
 Units are in seconds . Expects something like :
 ranges = [ [ 1,2 ] , [ 4,5 ] , [ 7,7.5 ] ]
 None values will be replaced with maximum / minimum bounds .
 For baseline subtraction , make a range baseline then sub it youtself .
 returns datas [ iSweep ] [ iRange ] [ AVorSD ]
 if a percentile is given , return that percentile rather than average .
 percentile = 50 is the median , but requires sorting , and is slower ."""

implementation:
ranges = copy . deepcopy ( ranges )
# TODO : make this cleaner . Why needed ?
# clean up ranges , make them indexes
for i in range ( len ( ranges ) ) :
    if ranges [ i ] [ 0 ] is None :
        ranges [ i ] [ 0 ] = 0
    else :
        ranges [ i ] [ 0 ] = int ( ranges [ i ] [ 0 ] * self . rate )
    if ranges [ i ] [ 1 ] is None :
        ranges [ i ] [ 1 ] = - 1
    else :
        ranges [ i ] [ 1 ] = int ( ranges [ i ] [ 1 ] * self . rate )
# do the math
datas = np . empty ( ( self . sweeps , len ( ranges ) , 2 ) )
# [ sweep ] [ range ] = [ Av , Er ]
for iSweep in range ( self . sweeps ) :
    self . setSweep ( iSweep )
    for iRange in range ( len ( ranges ) ) :
        I1 = ranges [ iRange ] [ 0 ]
        I2 = ranges [ iRange ] [ 1 ]
        if percentile :
            datas [ iSweep ] [ iRange ] [ 0 ] = np . percentile ( self . dataY [ I1 : I2 ] , percentile )
        else :
            datas [ iSweep ] [ iRange ] [ 0 ] = np . average ( self . dataY [ I1 : I2 ] )
        datas [ iSweep ] [ iRange ] [ 1 ] = np . std ( self . dataY [ I1 : I2 ] )
return datas


signature:
def filter_gaussian ( self , sigmaMs = 100 , applyFiltered = False , applyBaseline = False ) :
    """RETURNS filtered trace . Desn ' t filter it in place ."""

implementation:
if sigmaMs == 0 :
    return self . dataY
filtered = cm . filter_gaussian ( self . dataY , sigmaMs )
if applyBaseline :
    self . dataY = self . dataY - filtered
elif applyFiltered :
    self . dataY = filtered
else :
    return filtered


signature:
def saveThing ( self , thing , fname , overwrite = True , ext = ".pkl" ) :
    """save any object as / swhlab4 / ID _ [ fname ] . pkl"""

implementation:
if not os . path . exists ( os . path . dirname ( self . outpre ) ) :
    os . mkdir ( os . path . dirname ( self . outpre ) )
if ext and not ext in fname :
    fname += ext
fname = self . outpre + fname
if overwrite is False :
    if os . path . exists ( fname ) :
        print ( " o- not overwriting [%s]" % os . path . basename ( fname ) )
        return
time1 = cm . timethis ( )
pickle . dump ( thing , open ( fname , "wb" ) , pickle . HIGHEST_PROTOCOL )
print ( " <- saving [%s] %s (%.01f kB) took %.02f ms" % ( os . path . basename ( fname ) , str ( type ( thing ) ) , sys . getsizeof ( pickle . dumps ( thing , - 1 ) ) / 1e3 , cm . timethis ( time1 ) ) )


signature:
def loadThing ( self , fname , ext = ".pkl" ) :
    """save any object from / swhlab4 / ID _ [ fname ] . pkl"""

implementation:
if ext and not ext in fname :
    fname += ext
fname = self . outpre + fname
time1 = cm . timethis ( )
thing = pickle . load ( open ( fname , "rb" ) )
print ( " -> loading [%s] (%.01f kB) took %.02f ms" % ( os . path . basename ( fname ) , sys . getsizeof ( pickle . dumps ( thing , - 1 ) ) / 1e3 , cm . timethis ( time1 ) ) )
return thing


signature:
def deleteStuff ( self , ext = "*" , spareInfo = True , spare = [ "_info.pkl" ] ) :
    """delete / swhlab4 / ID _ *"""

implementation:
print ( " -- deleting /swhlab4/" + ext )
for fname in sorted ( glob . glob ( self . outpre + ext ) ) :
    reallyDelete = True
    for item in spare :
        if item in fname :
            reallyDelete = False
    if reallyDelete :
        os . remove ( fname )


signature:
def validate_activatable_models ( ) :
    """Raises a ValidationError for any ActivatableModel that has ForeignKeys or OneToOneFields that will
 cause cascading deletions to occur . This function also raises a ValidationError if the activatable
 model has not defined a Boolean field with the field name defined by the ACTIVATABLE _ FIELD _ NAME variable
 on the model ."""

implementation:
for model in get_activatable_models ( ) : # Verify the activatable model has an activatable boolean field
    activatable_field = next ( ( f for f in model . _meta . fields if f . __class__ == models . BooleanField and f . name == model . ACTIVATABLE_FIELD_NAME ) , None )
    if activatable_field is None :
        raise ValidationError ( ( 'Model {0} is an activatable model. It must define an activatable BooleanField that ' 'has a field name of model.ACTIVATABLE_FIELD_NAME (which defaults to is_active)' . format ( model ) ) )
    # Ensure all foreign keys and onetoone fields will not result in cascade deletions if not cascade deletable
    if not model . ALLOW_CASCADE_DELETE :
        for field in model . _meta . fields :
            if field . __class__ in ( models . ForeignKey , models . OneToOneField ) :
                if field . remote_field . on_delete == models . CASCADE :
                    raise ValidationError ( ( 'Model {0} is an activatable model. All ForeignKey and OneToOneFields ' 'must set on_delete methods to something other than CASCADE (the default). ' 'If you want to explicitely allow cascade deletes, then you must set the ' 'ALLOW_CASCADE_DELETE=True class variable on your model.' ) . format ( model ) )


signature:
def to_table ( args , vdims = [ ] ) :
    "Helper function to convet an Args object to a HoloViews Table"

implementation:
if not Table :
    return "HoloViews Table not available"
kdims = [ dim for dim in args . constant_keys + args . varying_keys if dim not in vdims ]
items = [ tuple ( [ spec [ k ] for k in kdims + vdims ] ) for spec in args . specs ]
return Table ( items , kdims = kdims , vdims = vdims )


signature:
def pprint_args ( self , pos_args , keyword_args , infix_operator = None , extra_params = { } ) :
    """Method to define the positional arguments and keyword order
 for pretty printing ."""

implementation:
if infix_operator and not ( len ( pos_args ) == 2 and keyword_args == [ ] ) :
    raise Exception ( 'Infix format requires exactly two' ' positional arguments and no keywords' )
( kwargs , _ , _ , _ ) = self . _pprint_args
self . _pprint_args = ( keyword_args + kwargs , pos_args , infix_operator , extra_params )


signature:
def _pprint ( self , cycle = False , flat = False , annotate = False , onlychanged = True , level = 1 , tab = '   ' ) :
    """Pretty printer that prints only the modified keywords and
 generates flat representations ( for repr ) and optionally
 annotates the top of the repr with a comment ."""

implementation:
( kwargs , pos_args , infix_operator , extra_params ) = self . _pprint_args
( br , indent ) = ( '' if flat else '\n' , '' if flat else tab * level )
prettify = lambda x : isinstance ( x , PrettyPrinted ) and not flat
pretty = lambda x : x . _pprint ( flat = flat , level = level + 1 ) if prettify ( x ) else repr ( x )
params = dict ( self . get_param_values ( ) )
show_lexsort = getattr ( self , '_lexorder' , None ) is not None
modified = [ k for ( k , v ) in self . get_param_values ( onlychanged = onlychanged ) ]
pkwargs = [ ( k , params [ k ] ) for k in kwargs if ( k in modified ) ] + list ( extra_params . items ( ) )
arg_list = [ ( k , params [ k ] ) for k in pos_args ] + pkwargs
lines = [ ]
if annotate : # Optional annotating comment
    len_ckeys , len_vkeys = len ( self . constant_keys ) , len ( self . varying_keys )
    info_triple = ( len ( self ) , ', %d constant key(s)' % len_ckeys if len_ckeys else '' , ', %d varying key(s)' % len_vkeys if len_vkeys else '' )
    annotation = '# == %d items%s%s ==\n' % info_triple
    lines = [ annotation ]
if show_lexsort :
    lines . append ( '(' )
if cycle :
    lines . append ( '%s(...)' % self . __class__ . __name__ )
elif infix_operator :
    level = level - 1
    triple = ( pretty ( params [ pos_args [ 0 ] ] ) , infix_operator , pretty ( params [ pos_args [ 1 ] ] ) )
    lines . append ( '%s %s %s' % triple )
else :
    lines . append ( '%s(' % self . __class__ . __name__ )
    for ( k , v ) in arg_list :
        lines . append ( '%s%s=%s' % ( br + indent , k , pretty ( v ) ) )
        lines . append ( ',' )
    lines = lines [ : - 1 ] + [ br + ( tab * ( level - 1 ) ) + ')' ]
# Remove trailing comma
if show_lexsort :
    lines . append ( ').lexsort(%s)' % ', ' . join ( repr ( el ) for el in self . _lexorder ) )
return '' . join ( lines )


signature:
def spec_formatter ( cls , spec ) :
    "Formats the elements of an argument set appropriately"

implementation: return type ( spec ) ( ( k , str ( v ) ) for ( k , v ) in spec . items ( ) )


signature:
def _collect_by_key ( self , specs ) :
    """Returns a dictionary like object with the lists of values
 collapsed by their respective key . Useful to find varying vs
 constant keys and to find how fast keys vary ."""

implementation:
# Collect ( key , value ) tuples as list of lists , flatten with chain
allkeys = itertools . chain . from_iterable ( [ [ ( k , run [ k ] ) for k in run ] for run in specs ] )
collection = defaultdict ( list )
for ( k , v ) in allkeys :
    collection [ k ] . append ( v )
return collection


signature:
def _cartesian_product ( self , first_specs , second_specs ) :
    """Takes the Cartesian product of the specifications . Result will
 contain N specifications where N = len ( first _ specs ) *
 len ( second _ specs ) and keys are merged .
 Example : [ { ' a ' : 1 } , { ' b ' : 2 } ] * [ { ' c ' : 3 } , { ' d ' : 4 } ] =
 [ { ' a ' : 1 , ' c ' : 3 } , { ' a ' : 1 , ' d ' : 4 } , { ' b ' : 2 , ' c ' : 3 } , { ' b ' : 2 , ' d ' : 4 } ]"""

implementation: return [ dict ( zip ( list ( s1 . keys ( ) ) + list ( s2 . keys ( ) ) , list ( s1 . values ( ) ) + list ( s2 . values ( ) ) ) ) for s1 in first_specs for s2 in second_specs ]


signature:
def summary ( self ) :
    """A succinct summary of the argument specifier . Unlike the repr ,
 a summary does not have to be complete but must supply the
 most relevant information about the object to the user ."""

implementation:
print ( "Items: %s" % len ( self ) )
varying_keys = ', ' . join ( '%r' % k for k in self . varying_keys )
print ( "Varying Keys: %s" % varying_keys )
items = ', ' . join ( [ '%s=%r' % ( k , v ) for ( k , v ) in self . constant_items ] )
if self . constant_items :
    print ( "Constant Items: %s" % items )


signature:
def _build_specs ( self , specs , kwargs , fp_precision ) :
    """Returns the specs , the remaining kwargs and whether or not the
 constructor was called with kwarg or explicit specs ."""

implementation:
if specs is None :
    overrides = param . ParamOverrides ( self , kwargs , allow_extra_keywords = True )
    extra_kwargs = overrides . extra_keywords ( )
    kwargs = dict ( [ ( k , v ) for ( k , v ) in kwargs . items ( ) if k not in extra_kwargs ] )
    rounded_specs = list ( self . round_floats ( [ extra_kwargs ] , fp_precision ) )
    if extra_kwargs == { } :
        return [ ] , kwargs , True
    else :
        return rounded_specs , kwargs , False
return list ( self . round_floats ( specs , fp_precision ) ) , kwargs , True


signature:
def _unique ( self , sequence , idfun = repr ) :
    """Note : repr ( ) must be implemented properly on all objects . This
 is implicitly assumed by Lancet when Python objects need to be
 formatted to string representation ."""

implementation:
seen = { }
return [ seen . setdefault ( idfun ( e ) , e ) for e in sequence if idfun ( e ) not in seen ]


signature:
def show ( self , exclude = [ ] ) :
    """Convenience method to inspect the available argument values in
 human - readable format . The ordering of keys is determined by
 how quickly they vary .
 The exclude list allows specific keys to be excluded for
 readability ( e . g . to hide long , absolute filenames ) ."""

implementation:
ordering = self . constant_keys + self . varying_keys
spec_lines = [ ', ' . join ( [ '%s=%s' % ( k , s [ k ] ) for k in ordering if ( k in s ) and ( k not in exclude ) ] ) for s in self . specs ]
print ( '\n' . join ( [ '%d: %s' % ( i , l ) for ( i , l ) in enumerate ( spec_lines ) ] ) )


signature:
def lexsort ( self , * order ) :
    """The lexical sort order is specified by a list of string
 arguments . Each string is a key name prefixed by ' + ' or ' - '
 for ascending and descending sort respectively . If the key is
 not found in the operand ' s set of varying keys , it is ignored ."""

implementation:
if order == [ ] :
    raise Exception ( "Please specify the keys for sorting, use" "'+' prefix for ascending," "'-' for descending.)" )
if not set ( el [ 1 : ] for el in order ) . issubset ( set ( self . varying_keys ) ) :
    raise Exception ( "Key(s) specified not in the set of varying keys." )
sorted_args = copy . deepcopy ( self )
specs_param = sorted_args . params ( 'specs' )
specs_param . constant = False
sorted_args . specs = self . _lexsorted_specs ( order )
specs_param . constant = True
sorted_args . _lexorder = order
return sorted_args


signature:
def _lexsorted_specs ( self , order ) :
    """A lexsort is specified using normal key string prefixed by ' + '
 ( for ascending ) or ' - ' for ( for descending ) .
 Note that in Python 2 , if a key is missing , None is returned
 ( smallest Python value ) . In Python 3 , an Exception will be
 raised regarding comparison of heterogenous types ."""

implementation:
specs = self . specs [ : ]
if not all ( el [ 0 ] in [ '+' , '-' ] for el in order ) :
    raise Exception ( "Please specify the keys for sorting, use" "'+' prefix for ascending," "'-' for descending.)" )
sort_cycles = [ ( el [ 1 : ] , True if el [ 0 ] == '+' else False ) for el in reversed ( order ) if el [ 1 : ] in self . varying_keys ]
for ( key , ascending ) in sort_cycles :
    specs = sorted ( specs , key = lambda s : s . get ( key , None ) , reverse = ( not ascending ) )
return specs


signature:
def linspace ( self , start , stop , n ) :
    """Simple replacement for numpy linspace"""

implementation:
if n == 1 :
    return [ start ]
L = [ 0.0 ] * n
nm1 = n - 1
nm1inv = 1.0 / nm1
for i in range ( n ) :
    L [ i ] = nm1inv * ( start * ( nm1 - i ) + stop * i )
return L


signature:
def extract_log ( log_path , dict_type = dict ) :
    """Parses the log file generated by a launcher and returns
 dictionary with tid keys and specification values .
 Ordering can be maintained by setting dict _ type to the
 appropriate constructor ( i . e . OrderedDict ) . Keys are converted
 from unicode to strings for kwarg use ."""

implementation:
log_path = ( log_path if os . path . isfile ( log_path ) else os . path . join ( os . getcwd ( ) , log_path ) )
with open ( log_path , 'r' ) as log :
    splits = ( line . split ( ) for line in log )
    uzipped = ( ( int ( split [ 0 ] ) , json . loads ( " " . join ( split [ 1 : ] ) ) ) for split in splits )
    szipped = [ ( i , dict ( ( str ( k ) , v ) for ( k , v ) in d . items ( ) ) ) for ( i , d ) in uzipped ]
return dict_type ( szipped )


signature:
def write_log ( log_path , data , allow_append = True ) :
    """Writes the supplied specifications to the log path . The data
 may be supplied as either as a an Args or as a list of
 dictionaries .
 By default , specifications will be appropriately appended to
 an existing log file . This can be disabled by setting
 allow _ append to False ."""

implementation:
append = os . path . isfile ( log_path )
islist = isinstance ( data , list )
if append and not allow_append :
    raise Exception ( 'Appending has been disabled' ' and file %s exists' % log_path )
if not ( islist or isinstance ( data , Args ) ) :
    raise Exception ( 'Can only write Args objects or dictionary' ' lists to log file.' )
specs = data if islist else data . specs
if not all ( isinstance ( el , dict ) for el in specs ) :
    raise Exception ( 'List elements must be dictionaries.' )
log_file = open ( log_path , 'r+' ) if append else open ( log_path , 'w' )
start = int ( log_file . readlines ( ) [ - 1 ] . split ( ) [ 0 ] ) + 1 if append else 0
ascending_indices = range ( start , start + len ( data ) )
log_str = '\n' . join ( [ '%d %s' % ( tid , json . dumps ( el ) ) for ( tid , el ) in zip ( ascending_indices , specs ) ] )
log_file . write ( "\n" + log_str if append else log_str )
log_file . close ( )


signature:
def directory ( cls , directory , root = None , extension = None , ** kwargs ) :
    """Load all the files in a given directory selecting only files
 with the given extension if specified . The given kwargs are
 passed through to the normal constructor ."""

implementation:
root = os . getcwd ( ) if root is None else root
suffix = '' if extension is None else '.' + extension . rsplit ( '.' ) [ - 1 ]
pattern = directory + os . sep + '*' + suffix
key = os . path . join ( root , directory , '*' ) . rsplit ( os . sep ) [ - 2 ]
format_parse = list ( string . Formatter ( ) . parse ( key ) )
if not all ( [ el is None for el in zip ( * format_parse ) [ 1 ] ] ) :
    raise Exception ( 'Directory cannot contain format field specifications' )
return cls ( key , pattern , root , ** kwargs )


signature:
def fields ( self ) :
    """Return the fields specified in the pattern using Python ' s
 formatting mini - language ."""

implementation:
parse = list ( string . Formatter ( ) . parse ( self . pattern ) )
return [ f for f in zip ( * parse ) [ 1 ] if f is not None ]


signature:
def _load_expansion ( self , key , root , pattern ) :
    """Loads the files that match the given pattern ."""

implementation:
path_pattern = os . path . join ( root , pattern )
expanded_paths = self . _expand_pattern ( path_pattern )
specs = [ ]
for ( path , tags ) in expanded_paths :
    filelist = [ os . path . join ( path , f ) for f in os . listdir ( path ) ] if os . path . isdir ( path ) else [ path ]
    for filepath in filelist :
        specs . append ( dict ( tags , ** { key : os . path . abspath ( filepath ) } ) )
return sorted ( specs , key = lambda s : s [ key ] )


signature:
def _expand_pattern ( self , pattern ) :
    """From the pattern decomposition , finds the absolute paths
 matching the pattern ."""

implementation:
( globpattern , regexp , fields , types ) = self . _decompose_pattern ( pattern )
filelist = glob . glob ( globpattern )
expansion = [ ]
for fname in filelist :
    if fields == [ ] :
        expansion . append ( ( fname , { } ) )
        continue
    match = re . match ( regexp , fname )
    if match is None :
        continue
    match_items = match . groupdict ( ) . items ( )
    tags = dict ( ( k , types . get ( k , str ) ( v ) ) for ( k , v ) in match_items )
    expansion . append ( ( fname , tags ) )
return expansion


signature:
def _decompose_pattern ( self , pattern ) :
    """Given a path pattern with format declaration , generates a
 four - tuple ( glob _ pattern , regexp pattern , fields , type map )"""

implementation:
sep = '~lancet~sep~'
float_codes = [ 'e' , 'E' , 'f' , 'F' , 'g' , 'G' , 'n' ]
typecodes = dict ( [ ( k , float ) for k in float_codes ] + [ ( 'b' , bin ) , ( 'd' , int ) , ( 'o' , oct ) , ( 'x' , hex ) ] )
parse = list ( string . Formatter ( ) . parse ( pattern ) )
text , fields , codes , _ = zip ( * parse )
# Finding the field types from format string
types = [ ]
for ( field , code ) in zip ( fields , codes ) :
    if code in [ '' , None ] :
        continue
    constructor = typecodes . get ( code [ - 1 ] , None )
    if constructor :
        types += [ ( field , constructor ) ]
stars = [ '' if not f else '*' for f in fields ]
globpat = '' . join ( text + star for ( text , star ) in zip ( text , stars ) )
refields = [ '' if not f else sep + ( '(?P<%s>.*?)' % f ) + sep for f in fields ]
parts = '' . join ( text + group for ( text , group ) in zip ( text , refields ) ) . split ( sep )
for i in range ( 0 , len ( parts ) , 2 ) :
    parts [ i ] = re . escape ( parts [ i ] )
regexp_pattern = '' . join ( parts ) . replace ( '\\*' , '.*' )
fields = list ( f for f in fields if f )
return globpat , regexp_pattern , fields , dict ( types )


signature:
def from_pattern ( cls , pattern , filetype = None , key = 'filename' , root = None , ignore = [ ] ) :
    """Convenience method to directly chain a pattern processed by
 FilePattern into a FileInfo instance .
 Note that if a default filetype has been set on FileInfo , the
 filetype argument may be omitted ."""

implementation:
filepattern = FilePattern ( key , pattern , root = root )
if FileInfo . filetype and filetype is None :
    filetype = FileInfo . filetype
elif filetype is None :
    raise Exception ( "The filetype argument must be supplied unless " "an appropriate default has been specified as " "FileInfo.filetype" )
return FileInfo ( filepattern , key , filetype , ignore = ignore )


signature:
def load ( self , val , ** kwargs ) :
    """Load the file contents into the supplied pandas dataframe or
 HoloViews Table . This allows a selection to be made over the
 metadata before loading the file contents ( may be slow ) ."""

implementation:
if Table and isinstance ( val , Table ) :
    return self . load_table ( val , ** kwargs )
elif DataFrame and isinstance ( val , DataFrame ) :
    return self . load_dframe ( val , ** kwargs )
else :
    raise Exception ( "Type %s not a DataFrame or Table." % type ( val ) )


signature:
def load_table ( self , table ) :
    """Load the file contents into the supplied Table using the
 specified key and filetype . The input table should have the
 filenames as values which will be replaced by the loaded
 data . If data _ key is specified , this key will be used to index
 the loaded data to retrive the specified item ."""

implementation:
items , data_keys = [ ] , None
for key , filename in table . items ( ) :
    data_dict = self . filetype . data ( filename [ 0 ] )
    current_keys = tuple ( sorted ( data_dict . keys ( ) ) )
    values = [ data_dict [ k ] for k in current_keys ]
    if data_keys is None :
        data_keys = current_keys
    elif data_keys != current_keys :
        raise Exception ( "Data keys are inconsistent" )
    items . append ( ( key , values ) )
return Table ( items , kdims = table . kdims , vdims = data_keys )


signature:
def load_dframe ( self , dframe ) :
    """Load the file contents into the supplied dataframe using the
 specified key and filetype ."""

implementation:
filename_series = dframe [ self . key ]
loaded_data = filename_series . map ( self . filetype . data )
keys = [ list ( el . keys ( ) ) for el in loaded_data . values ]
for key in set ( ) . union ( * keys ) :
    key_exists = key in dframe . columns
    if key_exists :
        self . warning ( "Appending '_data' suffix to data key %r to avoid" "overwriting existing metadata with the same name." % key )
    suffix = '_data' if key_exists else ''
    dframe [ key + suffix ] = loaded_data . map ( lambda x : x . get ( key , np . nan ) )
return dframe


signature:
def _info ( self , source , key , filetype , ignore ) :
    """Generates the union of the source . specs and the metadata
 dictionary loaded by the filetype object ."""

implementation:
specs , mdata = [ ] , { }
mdata_clashes = set ( )
for spec in source . specs :
    if key not in spec :
        raise Exception ( "Key %r not available in 'source'." % key )
    mdata = dict ( ( k , v ) for ( k , v ) in filetype . metadata ( spec [ key ] ) . items ( ) if k not in ignore )
    mdata_spec = { }
    mdata_spec . update ( spec )
    mdata_spec . update ( mdata )
    specs . append ( mdata_spec )
    mdata_clashes = mdata_clashes | ( set ( spec . keys ( ) ) & set ( mdata . keys ( ) ) )
# Metadata clashes can be avoided by using the ignore list .
if mdata_clashes :
    self . warning ( "Loaded metadata keys overriding source keys." )
return specs


signature:
async def _push ( self , * args , ** kwargs ) :
    """Push new data into the buffer . Resume looping if paused ."""

implementation:
self . _data . append ( ( args , kwargs ) )
if self . _future is not None :
    future , self . _future = self . _future , None
    future . set_result ( True )


signature:
def newVersion ( ) :
    """increments version counter in swhlab / version . py"""

implementation:
version = None
fname = '../swhlab/version.py'
with open ( fname ) as f :
    raw = f . read ( ) . split ( "\n" )
    for i , line in enumerate ( raw ) :
        if line . startswith ( "__counter__" ) :
            if version is None :
                version = int ( line . split ( "=" ) [ 1 ] )
            raw [ i ] = "__counter__=%d" % ( version + 1 )
with open ( fname , 'w' ) as f :
    f . write ( "\n" . join ( raw ) )
print ( "upgraded from version %03d to %03d" % ( version , version + 1 ) )


signature:
def figureStimulus ( abf , sweeps = [ 0 ] ) :
    """Create a plot of one area of interest of a single sweep ."""

implementation:
stimuli = [ 2.31250 , 2.35270 ]
for sweep in sweeps :
    abf . setsweep ( sweep )
    for stimulus in stimuli :
        S1 = int ( abf . pointsPerSec * stimulus )
        S2 = int ( abf . pointsPerSec * ( stimulus + 0.001 ) )
        # 1ms of blanking
        abf . sweepY [ S1 : S2 ] = np . nan
        # blank out the stimulus area
    I1 = int ( abf . pointsPerSec * 2.2 )
    # time point ( sec ) to start
    I2 = int ( abf . pointsPerSec * 2.6 )
    # time point ( sec ) to end
    baseline = np . average ( abf . sweepY [ int ( abf . pointsPerSec * 2.0 ) : int ( abf . pointsPerSec * 2.2 ) ] )
    Ys = lowPassFilter ( abf . sweepY [ I1 : I2 ] ) - baseline
    Xs = abf . sweepX2 [ I1 : I1 + len ( Ys ) ] . flatten ( )
    plt . plot ( Xs , Ys , alpha = .5 , lw = 2 )
return


signature:
def phasicTonic ( self , m1 = None , m2 = None , chunkMs = 50 , quietPercentile = 10 , histResolution = 1 ) :
    """chunkMs should be ~ 50 ms or greater .
 bin sizes must be equal to or multiples of the data resolution .
 transients smaller than the expected RMS will be silenced ."""

implementation:
# prepare sectioning values to be used later ( marker positions )
m1 = 0 if m1 is None else m1 * self . pointsPerSec
m2 = len ( abf . sweepY ) if m2 is None else m2 * self . pointsPerSec
m1 , m2 = int ( m1 ) , int ( m2 )
# prepare histogram values to be used later
padding = 200
# pA or mV of maximum expected deviation
chunkPoints = int ( chunkMs * self . pointsPerMs )
histBins = int ( ( padding * 2 ) / histResolution )
# center the data at 0 using peak histogram , not the mean
# Y = self . sweepY [ m1 : m2]
Y = self . sweepYfilteredHisto ( ) [ m1 : m2 ]
hist , bins = np . histogram ( Y , bins = 2 * padding )
# Yoffset = bins [ np . where ( hist = = max ( hist ) ) [ 0 ] [ 0 ] ]
# Y = Y - Yoffset # we don ' t have to , but PDF math is easier
# create histogram for all data in the sweep
nChunks = int ( len ( Y ) / chunkPoints )
hist , bins = np . histogram ( Y , bins = histBins , range = ( - padding , padding ) )
# create histogram for just the sweeps with the lowest variance
chunks = np . reshape ( Y [ : nChunks * chunkPoints ] , ( nChunks , chunkPoints ) )
# variances = np . var ( chunks , axis = 1)
variances = np . ptp ( chunks , axis = 1 )
percentiles = np . empty ( len ( variances ) )
for i , variance in enumerate ( variances ) :
    percentiles [ i ] = sorted ( variances ) . index ( variance ) / len ( variances ) * 100
blData = chunks [ np . where ( percentiles <= quietPercentile ) [ 0 ] ] . flatten ( )
blHist , blBins = np . histogram ( blData , bins = histBins , range = ( - padding , padding ) )
blHist = blHist / max ( blHist ) * max ( hist )
# determine the phasic current by subtracting - out the baseline
diff = hist - blHist
return diff / abf . pointsPerSec


signature:
def doStuff ( ABFfolder , analyze = False , convert = False , index = True , overwrite = True , launch = True ) :
    """Inelegant for now , but lets you manually analyze every ABF in a folder ."""

implementation:
IN = INDEX ( ABFfolder )
if analyze :
    IN . analyzeAll ( )
if convert :
    IN . convertImages ( )


signature:
def analyzeSingle ( abfFname ) :
    """Reanalyze data for a single ABF . Also remakes child and parent html ."""

implementation:
assert os . path . exists ( abfFname ) and abfFname . endswith ( ".abf" )
ABFfolder , ABFfname = os . path . split ( abfFname )
abfID = os . path . splitext ( ABFfname ) [ 0 ]
IN = INDEX ( ABFfolder )
IN . analyzeABF ( abfID )
IN . scan ( )
IN . html_single_basic ( [ abfID ] , overwrite = True )
IN . html_single_plot ( [ abfID ] , overwrite = True )
IN . scan ( )
IN . html_index ( )
return


signature:
def scan ( self ) :
    """scan folder1 and folder2 into files1 and files2.
 since we are on windows , simplify things by making them all lowercase .
 this WILL cause problems on ' nix operating systems . If this is the case ,
 just run a script to rename every file to all lowercase ."""

implementation:
t1 = cm . timeit ( )
self . files1 = cm . list_to_lowercase ( sorted ( os . listdir ( self . folder1 ) ) )
self . files2 = cm . list_to_lowercase ( sorted ( os . listdir ( self . folder2 ) ) )
self . files1abf = [ x for x in self . files1 if x . endswith ( ".abf" ) ]
self . files1abf = cm . list_to_lowercase ( cm . abfSort ( self . files1abf ) )
self . IDs = [ x [ : - 4 ] for x in self . files1abf ]
self . log . debug ( "folder1 has %d files" , len ( self . files1 ) )
self . log . debug ( "folder1 has %d abfs" , len ( self . files1abf ) )
self . log . debug ( "folder2 has %d files" , len ( self . files2 ) )
self . log . debug ( "scanning folders took %s" , cm . timeit ( t1 ) )


signature:
def convertImages ( self ) :
    """run this to turn all folder1 TIFs and JPGs into folder2 data .
 TIFs will be treated as micrographs and converted to JPG with enhanced
 contrast . JPGs will simply be copied over ."""

implementation:
# copy over JPGs ( and such )
exts = [ '.jpg' , '.png' ]
for fname in [ x for x in self . files1 if cm . ext ( x ) in exts ] :
    ID = "UNKNOWN"
    if len ( fname ) > 8 and fname [ : 8 ] in self . IDs :
        ID = fname [ : 8 ]
    fname2 = ID + "_jpg_" + fname
    if not fname2 in self . files2 :
        self . log . info ( "copying over [%s]" % fname2 )
        shutil . copy ( os . path . join ( self . folder1 , fname ) , os . path . join ( self . folder2 , fname2 ) )
    if not fname [ : 8 ] + ".abf" in self . files1 :
        self . log . error ( "orphan image: %s" , fname )
# convert TIFs ( and such ) to JPGs
exts = [ '.tif' , '.tiff' ]
for fname in [ x for x in self . files1 if cm . ext ( x ) in exts ] :
    ID = "UNKNOWN"
    if len ( fname ) > 8 and fname [ : 8 ] in self . IDs :
        ID = fname [ : 8 ]
    fname2 = ID + "_tif_" + fname + ".jpg"
    if not fname2 in self . files2 :
        self . log . info ( "converting micrograph [%s]" % fname2 )
        imaging . TIF_to_jpg ( os . path . join ( self . folder1 , fname ) , saveAs = os . path . join ( self . folder2 , fname2 ) )
    if not fname [ : 8 ] + ".abf" in self . files1 :
        self . log . error ( "orphan image: %s" , fname )


signature:
def analyzeAll ( self ) :
    """analyze every unanalyzed ABF in the folder ."""

implementation:
searchableData = str ( self . files2 )
self . log . debug ( "considering analysis for %d ABFs" , len ( self . IDs ) )
for ID in self . IDs :
    if not ID + "_" in searchableData :
        self . log . debug ( "%s needs analysis" , ID )
        try :
            self . analyzeABF ( ID )
        except :
            print ( "EXCEPTION! " * 100 )
    else :
        self . log . debug ( "%s has existing analysis, not overwriting" , ID )
self . log . debug ( "verified analysis of %d ABFs" , len ( self . IDs ) )


signature:
def analyzeABF ( self , ID ) :
    """Analye a single ABF : make data , index it .
 If called directly , will delete all ID _ data _ and recreate it ."""

implementation:
for fname in self . files2 :
    if fname . startswith ( ID + "_data_" ) :
        self . log . debug ( "deleting [%s]" , fname )
        os . remove ( os . path . join ( self . folder2 , fname ) )
self . log . info ( "analyzing (with overwrite) [%s]" , ID )
protocols . analyze ( os . path . join ( self . folder1 , ID + ".abf" ) )


signature:
def htmlFor ( self , fname ) :
    """return appropriate HTML determined by file extension ."""

implementation:
if os . path . splitext ( fname ) [ 1 ] . lower ( ) in [ '.jpg' , '.png' ] :
    html = '' % ( fname , fname )
    if "_tif_" in fname :
        html = html . replace ( '<img ' , '<img class="datapic micrograph"' )
    if "_plot_" in fname :
        html = html . replace ( '<img ' , '<img class="datapic intrinsic" ' )
    if "_experiment_" in fname :
        html = html . replace ( '<img ' , '<img class="datapic experiment" ' )
elif os . path . splitext ( fname ) [ 1 ] . lower ( ) in [ '.html' , '.htm' ] :
    html = 'LINK: %s' % fname
else :
    html = '
Not sure how to show: [%s]
' % fname
return html


signature:
def html_single_basic ( self , abfID , launch = False , overwrite = False ) :
    """generate a generic flat file html for an ABF parent . You could give
 this a single ABF ID , its parent ID , or a list of ABF IDs .
 If a child ABF is given , the parent will automatically be used ."""

implementation:
if type ( abfID ) is str :
    abfID = [ abfID ]
for thisABFid in cm . abfSort ( abfID ) :
    parentID = cm . parent ( self . groups , thisABFid )
    saveAs = os . path . abspath ( "%s/%s_basic.html" % ( self . folder2 , parentID ) )
    if overwrite is False and os . path . basename ( saveAs ) in self . files2 :
        continue
    filesByType = cm . filesByType ( self . groupFiles [ parentID ] )
    html = ""
    html += '
'
    html += 'summary of data from: %s
' % parentID
    html += '%s' % os . path . abspath ( self . folder1 + "/" + parentID + ".abf" )
    html += '
'
    catOrder = [ "experiment" , "plot" , "tif" , "other" ]
    categories = cm . list_order_by ( filesByType . keys ( ) , catOrder )
    for category in [ x for x in categories if len ( filesByType [ x ] ) ] :
        if category == 'experiment' :
            html += "
Experimental Data:
"
        elif category == 'plot' :
            html += "
Intrinsic Properties:
"
        elif category == 'tif' :
            html += "
Micrographs:
"
        elif category == 'other' :
            html += "
Additional Files:
"
        else :
            html += "
????:
"
        # html + = "
 "
        # html + = '
 ' * 3
        for fname in filesByType [ category ] :
            html += self . htmlFor ( fname )
        html += '
' * 3
    print ( "creating" , saveAs , '...' )
    style . save ( html , saveAs , launch = launch )


signature:
def html_single_plot ( self , abfID , launch = False , overwrite = False ) :
    """create ID _ plot . html of just intrinsic properties ."""

implementation:
if type ( abfID ) is str :
    abfID = [ abfID ]
for thisABFid in cm . abfSort ( abfID ) :
    parentID = cm . parent ( self . groups , thisABFid )
    saveAs = os . path . abspath ( "%s/%s_plot.html" % ( self . folder2 , parentID ) )
    if overwrite is False and os . path . basename ( saveAs ) in self . files2 :
        continue
    filesByType = cm . filesByType ( self . groupFiles [ parentID ] )
    html = ""
    html += '
'
    html += 'intrinsic properties for: %s
' % parentID
    html += '%s' % os . path . abspath ( self . folder1 + "/" + parentID + ".abf" )
    html += '
'
    for fname in filesByType [ 'plot' ] :
        html += self . htmlFor ( fname )
    print ( "creating" , saveAs , '...' )
    style . save ( html , saveAs , launch = launch )


signature:
def lowpass ( data , filterSize = None ) :
    """minimal complexity low - pass filtering .
 Filter size is how " wide " the filter will be .
 Sigma will be 1/10 of this filter width .
 If filter size isn ' t given , it will be 1/10 of the data size ."""

implementation:
if filterSize is None :
    filterSize = len ( data ) / 10
kernel = kernel_gaussian ( size = filterSize )
data = convolve ( data , kernel )
# do the convolution with padded edges
return data


signature:
def convolve ( signal , kernel ) :
    """This applies a kernel to a signal through convolution and returns the result .
 Some magic is done at the edges so the result doesn ' t apprach zero :
 1 . extend the signal ' s edges with len ( kernel ) / 2 duplicated values
 2 . perform the convolution ( ' same ' mode )
 3 . slice - off the ends we added
 4 . return the same number of points as the original"""

implementation:
pad = np . ones ( len ( kernel ) / 2 )
signal = np . concatenate ( ( pad * signal [ 0 ] , signal , pad * signal [ - 1 ] ) )
signal = np . convolve ( signal , kernel , mode = 'same' )
signal = signal [ len ( pad ) : - len ( pad ) ]
return signal


signature:
def timeit ( timer = None ) :
    """simple timer . returns a time object , or a string ."""

implementation:
if timer is None :
    return time . time ( )
else :
    took = time . time ( ) - timer
    if took < 1 :
        return "%.02f ms" % ( took * 1000.0 )
    elif took < 60 :
        return "%.02f s" % ( took )
    else :
        return "%.02f min" % ( took / 60.0 )


signature:
def list_move_to_front ( l , value = 'other' ) :
    """if the value is in the list , move it to the front and return it ."""

implementation:
l = list ( l )
if value in l :
    l . remove ( value )
    l . insert ( 0 , value )
return l


signature:
def list_move_to_back ( l , value = 'other' ) :
    """if the value is in the list , move it to the back and return it ."""

implementation:
l = list ( l )
if value in l :
    l . remove ( value )
    l . append ( value )
return l


signature:
def list_order_by ( l , firstItems ) :
    """given a list and a list of items to be first , return the list in the
 same order except that it begins with each of the first items ."""

implementation:
l = list ( l )
for item in firstItems [ : : - 1 ] : # backwards
    if item in l :
        l . remove ( item )
        l . insert ( 0 , item )
return l


signature:
def abfSort ( IDs ) :
    """given a list of goofy ABF names , return it sorted intelligently .
 This places things like 16o01001 after 16901001."""

implementation:
IDs = list ( IDs )
monO = [ ]
monN = [ ]
monD = [ ]
good = [ ]
for ID in IDs :
    if ID is None :
        continue
    if 'o' in ID :
        monO . append ( ID )
    elif 'n' in ID :
        monN . append ( ID )
    elif 'd' in ID :
        monD . append ( ID )
    else :
        good . append ( ID )
return sorted ( good ) + sorted ( monO ) + sorted ( monN ) + sorted ( monD )


signature:
def abfGroups ( abfFolder ) :
    """Given a folder path or list of files , return groups ( dict ) by cell .
 Rules which define parents ( cells ) :
 * assume each cell has one or several ABFs
 * that cell can be labeled by its " ID " or " parent " ABF ( first abf )
 * the ID is just the filename of the first abf without . abf
 * if any file starts with an " ID " , that ID becomes a parent .
 * examples could be 16o14044 . TIF or 16o14044 - cell1 - stuff . jpg
 * usually this is done by saving a pic of the cell with same filename
 Returns a dict of " parent IDs " representing the " children "
 groups [ " 16o14041 " ] = [ " 16o14041 " , " 16o14042 " , " 16o14043 " ]
 From there , getting children files is trivial . Just find all files in
 the same folder whose filenames begin with one of the children ."""

implementation:
# prepare the list of files , filenames , and IDs
files = False
if type ( abfFolder ) is str and os . path . isdir ( abfFolder ) :
    files = abfSort ( os . listdir ( abfFolder ) )
elif type ( abfFolder ) is list :
    files = abfSort ( abfFolder )
assert type ( files ) is list
files = list_to_lowercase ( files )
# group every filename in a different list , and determine parents
abfs , IDs , others , parents , days = [ ] , [ ] , [ ] , [ ] , [ ]
for fname in files :
    if fname . endswith ( ".abf" ) :
        abfs . append ( fname )
        IDs . append ( fname [ : - 4 ] )
        days . append ( fname [ : 5 ] )
    else :
        others . append ( fname )
for ID in IDs :
    for fname in others :
        if fname . startswith ( ID ) :
            parents . append ( ID )
parents = abfSort ( set ( parents ) )
# allow only one copy each
days = abfSort ( set ( days ) )
# allow only one copy each
# match up children with parents , respecting daily orphans .
groups = { }
for day in days :
    parent = None
    for fname in [ x for x in abfs if x . startswith ( day ) ] :
        ID = fname [ : - 4 ]
        if ID in parents :
            parent = ID
        if not parent in groups . keys ( ) :
            groups [ parent ] = [ ]
        groups [ parent ] . extend ( [ ID ] )
return groups


signature:
def abfGroupFiles ( groups , folder ) :
    """when given a dictionary where every key contains a list of IDs , replace
 the keys with the list of files matching those IDs . This is how you get a
 list of files belonging to each child for each parent ."""

implementation:
assert os . path . exists ( folder )
files = os . listdir ( folder )
group2 = { }
for parent in groups . keys ( ) :
    if not parent in group2 . keys ( ) :
        group2 [ parent ] = [ ]
    for ID in groups [ parent ] :
        for fname in [ x . lower ( ) for x in files if ID in x . lower ( ) ] :
            group2 [ parent ] . extend ( [ fname ] )
return group2


signature:
def parent ( groups , ID ) :
    """given a groups dictionary and an ID , return its actual parent ID ."""

implementation:
if ID in groups . keys ( ) :
    return ID
    # already a parent
if not ID in groups . keys ( ) :
    for actualParent in groups . keys ( ) :
        if ID in groups [ actualParent ] :
            return actualParent
            # found the actual parent
return None


signature:
def filesByType ( fileList ) :
    """given a list of files , return them as a dict sorted by type :
 * plot , tif , data , other"""

implementation:
features = [ "plot" , "tif" , "data" , "other" , "experiment" ]
files = { }
for feature in features :
    files [ feature ] = [ ]
for fname in fileList :
    other = True
    for feature in features :
        if "_" + feature + "_" in fname :
            files [ feature ] . extend ( [ fname ] )
            other = False
    if other :
        files [ 'other' ] . extend ( [ fname ] )
return files


signature:
def userFolder ( ) :
    """return the semi - temporary user folder"""

implementation:
# path = os . path . abspath ( tempfile . gettempdir ( ) + " / swhlab / " )
# don ' t use tempdir ! it will get deleted easily .
path = os . path . expanduser ( "~" ) + "/.swhlab/"
# works on windows or linux
# for me , path = r " C : \ Users \ swharden \ . swhlab "
if not os . path . exists ( path ) :
    print ( "creating" , path )
    os . mkdir ( path )
return os . path . abspath ( path )


signature:
def abfFname_Load ( ) :
    """return the path of the last loaded ABF ."""

implementation:
fname = userFolder ( ) + "/abfFname.ini"
if os . path . exists ( fname ) :
    abfFname = open ( fname ) . read ( ) . strip ( )
    if os . path . exists ( abfFname ) or abfFname . endswith ( "_._" ) :
        return abfFname
return os . path . abspath ( os . sep )


signature:
def abfFname_Save ( abfFname ) :
    """return the path of the last loaded ABF ."""

implementation:
fname = userFolder ( ) + "/abfFname.ini"
with open ( fname , 'w' ) as f :
    f . write ( os . path . abspath ( abfFname ) )
return


signature:
def gui_getFile ( ) :
    """Launch an ABF file selection file dialog .
 This is smart , and remembers ( through reboots ) where you last were ."""

implementation:
import tkinter as tk
from tkinter import filedialog
root = tk . Tk ( )
# this is natively supported by python
root . withdraw ( )
# hide main window
root . wm_attributes ( '-topmost' , 1 )
# always on top
fname = filedialog . askopenfilename ( title = "select ABF file" , filetypes = [ ( 'ABF Files' , '.abf' ) ] , initialdir = os . path . dirname ( abfFname_Load ( ) ) )
if fname . endswith ( ".abf" ) :
    abfFname_Save ( fname )
    return fname
else :
    print ( "didn't select an ABF!" )
    return None


signature:
def gui_getFolder ( ) :
    """Launch a folder selection dialog .
 This is smart , and remembers ( through reboots ) where you last were ."""

implementation:
import tkinter as tk
from tkinter import filedialog
root = tk . Tk ( )
# this is natively supported by python
root . withdraw ( )
# hide main window
root . wm_attributes ( '-topmost' , 1 )
# always on top
fname = filedialog . askdirectory ( title = "select folder of ABFs" , initialdir = os . path . dirname ( abfFname_Load ( ) ) )
if len ( fname ) > 3 :
    abfFname_Save ( fname + "/_._" )
    return fname
else :
    print ( "didn't select an ABF!" )
    return None


signature:
async def _try_catch_coro ( emitter , event , listener , coro ) :
    """Coroutine wrapper to catch errors after async scheduling .
 Args :
 emitter ( EventEmitter ) : The event emitter that is attempting to
 call a listener .
 event ( str ) : The event that triggered the emitter .
 listener ( async def ) : The async def that was used to generate the coro .
 coro ( coroutine ) : The coroutine that should be tried .
 If an exception is caught the function will use the emitter to emit the
 failure event . If , however , the current event _ is _ the failure event then
 the method reraises . The reraised exception may show in debug mode for the
 event loop but is otherwise silently dropped ."""

implementation:
try :
    await coro
except Exception as exc :
    if event == emitter . LISTENER_ERROR_EVENT :
        raise
    emitter . emit ( emitter . LISTENER_ERROR_EVENT , event , listener , exc )


signature:
def _check_limit ( self , event ) :
    """Check if the listener limit is hit and warn if needed ."""

implementation:
if self . count ( event ) > self . max_listeners :
    warnings . warn ( 'Too many listeners for event {}' . format ( event ) , ResourceWarning , )


signature:
def add_listener ( self , event , listener ) :
    """Bind a listener to a particular event .
 Args :
 event ( str ) : The name of the event to listen for . This may be any
 string value .
 listener ( def or async def ) : The callback to execute when the event
 fires . This may be a sync or async function ."""

implementation:
self . emit ( 'new_listener' , event , listener )
self . _listeners [ event ] . append ( listener )
self . _check_limit ( event )
return self


signature:
def once ( self , event , listener ) :
    """Add a listener that is only called once ."""

implementation:
self . emit ( 'new_listener' , event , listener )
self . _once [ event ] . append ( listener )
self . _check_limit ( event )
return self


signature:
def remove_listener ( self , event , listener ) :
    """Remove a listener from the emitter .
 Args :
 event ( str ) : The event name on which the listener is bound .
 listener : A reference to the same object given to add _ listener .
 Returns :
 bool : True if a listener was removed else False .
 This method only removes one listener at a time . If a listener is
 attached multiple times then this method must be called repeatedly .
 Additionally , this method removes listeners first from the those
 registered with ' on ' or ' add _ listener ' . If none are found it continue
 to remove afterwards from those added with ' once ' ."""

implementation:
with contextlib . suppress ( ValueError ) :
    self . _listeners [ event ] . remove ( listener )
    return True
with contextlib . suppress ( ValueError ) :
    self . _once [ event ] . remove ( listener )
    return True
return False


signature:
def remove_all_listeners ( self , event = None ) :
    """Remove all listeners , or those of the specified * event * .
 It ' s not a good idea to remove listeners that were added elsewhere in
 the code , especially when it ' s on an emitter that you didn ' t create
 ( e . g . sockets or file streams ) ."""

implementation:
if event is None :
    self . _listeners = collections . defaultdict ( list )
    self . _once = collections . defaultdict ( list )
else :
    del self . _listeners [ event ]
    del self . _once [ event ]


signature:
def _dispatch_coroutine ( self , event , listener , * args , ** kwargs ) :
    """Schedule a coroutine for execution .
 Args :
 event ( str ) : The name of the event that triggered this call .
 listener ( async def ) : The async def that needs to be executed .
 * args : Any number of positional arguments .
 * * kwargs : Any number of keyword arguments .
 The values of * args and * * kwargs are passed , unaltered , to the async
 def when generating the coro . If there is an exception generating the
 coro , such as the wrong number of arguments , the emitter ' s error event
 is triggered . If the triggering event _ is _ the emitter ' s error event
 then the exception is reraised . The reraised exception may show in
 debug mode for the event loop but is otherwise silently dropped ."""

implementation:
try :
    coro = listener ( * args , ** kwargs )
except Exception as exc :
    if event == self . LISTENER_ERROR_EVENT :
        raise
    return self . emit ( self . LISTENER_ERROR_EVENT , event , listener , exc )
asyncio . ensure_future ( _try_catch_coro ( self , event , listener , coro ) , loop = self . _loop , )


signature:
def _dispatch_function ( self , event , listener , * args , ** kwargs ) :
    """Execute a sync function .
 Args :
 event ( str ) : The name of the event that triggered this call .
 listener ( def ) : The def that needs to be executed .
 * args : Any number of positional arguments .
 * * kwargs : Any number of keyword arguments .
 The values of * args and * * kwargs are passed , unaltered , to the def
 when exceuting . If there is an exception executing the def , such as the
 wrong number of arguments , the emitter ' s error event is triggered . If
 the triggering event _ is _ the emitter ' s error event then the exception
 is reraised . The reraised exception may show in debug mode for the
 event loop but is otherwise silently dropped ."""

implementation:
try :
    return listener ( * args , ** kwargs )
except Exception as exc :
    if event == self . LISTENER_ERROR_EVENT :
        raise
    return self . emit ( self . LISTENER_ERROR_EVENT , event , listener , exc )


signature:
def _dispatch ( self , event , listener , * args , ** kwargs ) :
    """Dispatch an event to a listener .
 Args :
 event ( str ) : The name of the event that triggered this call .
 listener ( def or async def ) : The listener to trigger .
 * args : Any number of positional arguments .
 * * kwargs : Any number of keyword arguments .
 This method inspects the listener . If it is a def it dispatches the
 listener to a method that will execute that def . If it is an async def
 it dispatches it to a method that will schedule the resulting coro with
 the event loop ."""

implementation:
if ( asyncio . iscoroutinefunction ( listener ) or isinstance ( listener , functools . partial ) and asyncio . iscoroutinefunction ( listener . func ) ) :
    return self . _dispatch_coroutine ( event , listener , * args , ** kwargs )
return self . _dispatch_function ( event , listener , * args , ** kwargs )


signature:
def emit ( self , event , * args , ** kwargs ) :
    """Call each listener for the event with the given arguments .
 Args :
 event ( str ) : The event to trigger listeners on .
 * args : Any number of positional arguments .
 * * kwargs : Any number of keyword arguments .
 This method passes all arguments other than the event name directly
 to the listeners . If a listener raises an exception for any reason the
 ' listener - error ' , or current value of LISTENER _ ERROR _ EVENT , is emitted .
 Listeners to this event are given the event name , listener object , and
 the exception raised . If an error listener fails it does so silently .
 All event listeners are fired in a deferred way so this method returns
 immediately . The calling coro must yield at some point for the event
 to propagate to the listeners ."""

implementation:
listeners = self . _listeners [ event ]
listeners = itertools . chain ( listeners , self . _once [ event ] )
self . _once [ event ] = [ ]
for listener in listeners :
    self . _loop . call_soon ( functools . partial ( self . _dispatch , event , listener , * args , ** kwargs , ) )
return self


signature:
def count ( self , event ) :
    """Get the number of listeners for the event .
 Args :
 event ( str ) : The event for which to count all listeners .
 The resulting count is a combination of listeners added using
 ' on ' / ' add _ listener ' and ' once ' ."""

implementation: return len ( self . _listeners [ event ] ) + len ( self . _once [ event ] )


signature:
def phasicTonic ( self , m1 = None , m2 = None , chunkMs = 50 , quietPercentile = 10 , histResolution = .5 , plotToo = False ) :
    """let ' s keep the chunkMs as high as we reasonably can . 50ms is good .
 Things get flakey at lower numbers like 10ms .
 IMPORTANT ! for this to work , prevent 0s from averaging in , so keep
 bin sizes well above the data resolution ."""

implementation:
# prepare sectioning values to be used later
m1 = 0 if m1 is None else m1 * self . pointsPerSec
m2 = len ( abf . sweepY ) if m2 is None else m2 * self . pointsPerSec
m1 , m2 = int ( m1 ) , int ( m2 )
# prepare histogram values to be used later
padding = 200
# pA or mV of maximum expected deviation
chunkPoints = int ( chunkMs * self . pointsPerMs )
histBins = int ( ( padding * 2 ) / histResolution )
# center the data at 0 using peak histogram , not the mean
Y = self . sweepY [ m1 : m2 ]
hist , bins = np . histogram ( Y , bins = 2 * padding )
Yoffset = bins [ np . where ( hist == max ( hist ) ) [ 0 ] [ 0 ] ]
Y = Y - Yoffset
# we don ' t have to , but PDF math is easier
# calculate all histogram
nChunks = int ( len ( Y ) / chunkPoints )
hist , bins = np . histogram ( Y , bins = histBins , range = ( - padding , padding ) )
hist = hist / len ( Y )
# count as a fraction of total
Xs = bins [ 1 : ]
# get baseline data from chunks with smallest variance
chunks = np . reshape ( Y [ : nChunks * chunkPoints ] , ( nChunks , chunkPoints ) )
variances = np . var ( chunks , axis = 1 )
percentiles = np . empty ( len ( variances ) )
for i , variance in enumerate ( variances ) :
    percentiles [ i ] = sorted ( variances ) . index ( variance ) / len ( variances ) * 100
blData = chunks [ np . where ( percentiles <= quietPercentile ) [ 0 ] ] . flatten ( )
# generate the standard curve and pull it to the histogram height
sigma = np . sqrt ( np . var ( blData ) )
center = np . average ( blData ) + histResolution / 2
blCurve = mlab . normpdf ( Xs , center , sigma )
blCurve = blCurve * max ( hist ) / max ( blCurve )
# determine the phasic current by subtracting - out the baseline
# diff = hist - blCurve
diff = hist
IGNORE_DISTANCE = 5
# KEEP THIS FIXED , NOT A FUNCTION OF VARIANCE
ignrCenter = len ( Xs ) / 2
ignrPad = IGNORE_DISTANCE / histResolution
ignr1 , ignt2 = int ( ignrCenter - ignrPad ) , int ( ignrCenter + ignrPad )
diff [ ignr1 : ignt2 ] = 0
# optionally graph all this
if plotToo :
    plt . figure ( figsize = ( 15 , 5 ) )
    plt . plot ( Y )
    plt . figure ( figsize = ( 7 , 7 ) )
    ax1 = plt . subplot ( 211 )
    plt . title ( abf . ID + " phasic analysis" )
    plt . ylabel ( "fraction" )
    plt . plot ( Xs , hist , '-' , alpha = .8 , color = 'b' , lw = 3 )
    plt . plot ( Xs , blCurve , lw = 3 , alpha = .5 , color = 'r' )
    plt . margins ( 0 , .1 )
    plt . subplot ( 212 , sharex = ax1 )
    plt . title ( "baseline subtracted" )
    plt . ylabel ( "fraction" )
    plt . xlabel ( "data points (%s)" % abf . units )
    plt . plot ( Xs , diff , '-' , alpha = .8 , color = 'b' , lw = 3 )
    plt . axhline ( 0 , lw = 3 , alpha = .5 , color = 'r' )
    plt . axvline ( 0 , lw = 3 , alpha = .5 , color = 'k' )
    plt . margins ( 0 , .1 )
    plt . axis ( [ - 50 , 50 , None , None ] )
    plt . tight_layout ( )
    plt . show ( )
    print ( np . sum ( np . split ( diff , 2 ) , 1 ) )
return diff / len ( Y ) * abf . pointsPerSec


signature:
def genPNGs ( folder , files = None ) :
    """Convert each TIF to PNG . Return filenames of new PNGs ."""

implementation:
if files is None :
    files = glob . glob ( folder + "/*.*" )
new = [ ]
for fname in files :
    ext = os . path . basename ( fname ) . split ( "." ) [ - 1 ] . lower ( )
    if ext in [ 'tif' , 'tiff' ] :
        if not os . path . exists ( fname + ".png" ) :
            print ( " -- converting %s to PNG..." % os . path . basename ( fname ) )
            cm . image_convert ( fname )
            new . append ( fname )
            # fancy burn - in of image data
        else :
            pass
            # print ( " - - already converted % s to PNG . . . " % os . path . basename ( fname ) )
return new


signature:
def htmlABFcontent ( ID , group , d ) :
    """generate text to go inside < body > for single ABF page ."""

implementation:
html = ""
files = [ ]
for abfID in group :
    files . extend ( d [ abfID ] )
files = sorted ( files )
# start with root images
html += "
"
for fname in files :
    if ".png" in fname . lower ( ) and not "swhlab4" in fname :
        fname = "../" + os . path . basename ( fname )
        html += ' ' % ( fname , fname )
# progress to / swhlab4 / images
html += "
"
# ABFinfo
lastID = ''
for fname in sorted ( files ) :
    if not "swhlab4" in fname :
        continue
    ID = os . path . basename ( fname ) . split ( "_" ) [ 0 ]
    if not ID == lastID :
        lastID = ID
        html += "
%s
" % os . path . basename ( fname ) . split ( "_" ) [ 0 ]
    if ".png" in fname . lower ( ) :
        fname = os . path . basename ( fname )
        html += ' ' % ( fname , fname )
        continue
html += "
"
for fname in files :
    if not "swhlab4" in fname :
        continue
    if ".pkl" in fname :
        callit = os . path . basename ( fname )
        thing = cm . getPkl ( fname )
        if "_APs.pkl" in fname :
            callit += " (first AP)"
            thing = cm . dictFlat ( thing )
            if len ( thing ) :
                thing = thing [ 0 ]
        elif "_MTs.pkl" in fname :
            if type ( thing ) == dict :
                callit += " (from AVG of all sweeps)"
            else :
                callit += " (first sweep)"
                thing = thing [ 0 ]
        elif "_SAP.pkl" in fname :
            continue
            # don ' t plot those , too complicated
        elif "_info.pkl" in fname or "_iv.pkl" in fname :
            pass
            # no trouble , go for it
        else :
            print ( " ?? not sure how to index [%s]" % os . path . basename ( fname ) )
            continue
        if type ( thing ) is dict :
            thing = cm . msgDict ( thing )
        if type ( thing ) is list :
            out = ''
            for item in thing :
                out += str ( item ) + "\n"
            thing = out
        thing = str ( thing )
        # lol stringthing
        thing = "### %s ###\n" % os . path . basename ( fname ) + thing
        # putting it in a textbox is obnoxious . put it in the source instead .
        # html + = '

 < textarea rows = " % d " cols = " 70 " > % s < / textarea > ' % ( str ( thing ) . count ( " \ n " ) + 5 , thing )
        html += "(view source for %s) " % ( os . path . basename ( fname ) , thing )
return html


signature:
def htmlABF ( ID , group , d , folder , overwrite = False ) :
    """given an ID and the dict of files , generate a static html for that abf ."""

implementation:
fname = folder + "/swhlab4/%s_index.html" % ID
if overwrite is False and os . path . exists ( fname ) :
    return
html = TEMPLATES [ 'abf' ]
html = html . replace ( "~ID~" , ID )
html = html . replace ( "~CONTENT~" , htmlABFcontent ( ID , group , d ) )
print ( " <- writing [%s]" % os . path . basename ( fname ) )
with open ( fname , 'w' ) as f :
    f . write ( html )
return


signature:
def expMenu ( groups , folder ) :
    """read experiment . txt and return a dict with [ firstOfNewExp , color , star , comments ] ."""

implementation:
# # # GENERATE THE MENU DATA BASED ON EXPERIMENT FILE
orphans = sorted ( list ( groups . keys ( ) ) )
menu = [ ]
if os . path . exists ( folder + '/experiment.txt' ) :
    with open ( folder + '/experiment.txt' ) as f :
        raw = f . read ( )
else :
    raw = ""
for line in raw . split ( "\n" ) :
    item = { }
    if len ( line ) == 0 :
        continue
    if line . startswith ( "~" ) :
        line = line [ 1 : ] . split ( " " , 2 )
        item [ "ID" ] = line [ 0 ]
        item [ "symbol" ] = ''
        if len ( line ) > 1 :
            item [ "color" ] = line [ 1 ]
        else :
            item [ "color" ] = "white"
        if len ( line ) > 2 and len ( line [ 2 ] ) :
            item [ "comment" ] = line [ 2 ]
            if item [ "comment" ] [ 0 ] == "*" :
                item [ "symbol" ] = '*'
        else :
            item [ "comment" ] = ''
        if item [ "ID" ] in orphans :
            orphans . remove ( item [ "ID" ] )
    elif line . startswith ( "###" ) :
        line = line [ 3 : ] . strip ( ) . split ( " " , 1 )
        item [ "title" ] = line [ 0 ]
        item [ "comment" ] = ''
        if len ( line ) > 1 :
            if line [ 1 ] . startswith ( "- " ) :
                line [ 1 ] = line [ 1 ] [ 2 : ]
            item [ "comment" ] = line [ 1 ]
    else :
        item [ "unknown" ] = line
    menu . append ( item )
menu . append ( { "title" : "orphans" , "comment" : "" } )
for ophan in orphans :
    menu . append ( { "orphan" : ophan , "ID" : ophan , "color" : '' , "symbol" : '' , "comment" : '' } )
return menu


signature:
def genIndex ( folder , forceIDs = [ ] ) :
    """expects a folder of ABFs ."""

implementation:
if not os . path . exists ( folder + "/swhlab4/" ) :
    print ( " !! cannot index if no /swhlab4/" )
    return
timestart = cm . timethis ( )
files = glob . glob ( folder + "/*.*" )
# ABF folder
files . extend ( glob . glob ( folder + "/swhlab4/*.*" ) )
print ( " -- indexing glob took %.02f ms" % ( cm . timethis ( timestart ) * 1000 ) )
files . extend ( genPNGs ( folder , files ) )
files = sorted ( files )
timestart = cm . timethis ( )
d = cm . getIDfileDict ( files )
# TODO : this is really slow
print ( " -- filedict length:" , len ( d ) )
print ( " -- generating ID dict took %.02f ms" % ( cm . timethis ( timestart ) * 1000 ) )
groups = cm . getABFgroups ( files )
print ( " -- groups length:" , len ( groups ) )
for ID in sorted ( list ( groups . keys ( ) ) ) :
    overwrite = False
    for abfID in groups [ ID ] :
        if abfID in forceIDs :
            overwrite = True
    try :
        htmlABF ( ID , groups [ ID ] , d , folder , overwrite )
    except :
        print ( "~~ HTML GENERATION FAILED!!!" )
menu = expMenu ( groups , folder )
makeSplash ( menu , folder )
makeMenu ( menu , folder )
htmlFrames ( d , folder )
makeMenu ( menu , folder )
makeSplash ( menu , folder )


signature:
def drawPhasePlot ( abf , m1 = 0 , m2 = None ) :
    """Given an ABF object ( SWHLab ) , draw its phase plot of the current sweep .
 m1 and m2 are optional marks ( in seconds ) for plotting only a range of data .
 Assume a matplotlib figure is already open and just draw on top if it ."""

implementation:
if not m2 :
    m2 = abf . sweepLength
cm = plt . get_cmap ( 'CMRmap' )
# cm = plt . get _ cmap ( ' CMRmap _ r ' )
# cm = plt . get _ cmap ( ' spectral ' )
# cm = plt . get _ cmap ( ' winter ' )
# prepare Xs , Ys , and dYs
Y = abf . sweepY
Y = Y [ int ( abf . pointsPerSec * m1 ) : int ( abf . pointsPerSec * m2 ) ]
dY = ( Y [ 1 : ] - Y [ : - 1 ] ) * abf . rate / 1000.0
# mV / ms
dY = np . append ( dY , dY [ - 1 ] )
Xs = np . arange ( len ( dY ) ) / abf . pointsPerSec
Xs = Xs + Xs [ - 1 ] * abf . sweep
# plot the voltage
plt . subplot ( 131 )
plt . grid ( alpha = .5 )
plt . plot ( Xs , Y , lw = .5 , color = cm ( abf . sweep / abf . sweeps ) )
plt . title ( "membrane voltage" )
plt . ylabel ( "V (mV)" )
plt . xlabel ( "time (sec)" )
plt . margins ( 0 , .1 )
# plot the first derivative of the voltage
plt . subplot ( 132 )
plt . grid ( alpha = .5 )
plt . plot ( Xs , dY , lw = .5 , color = cm ( abf . sweep / abf . sweeps ) )
plt . title ( "voltage velocity" )
plt . ylabel ( "dV (mV/ms)" )
plt . xlabel ( "time (sec)" )
plt . margins ( 0 , .1 )
# make the phase plot
plt . subplot ( 133 )
plt . grid ( alpha = .5 )
plt . plot ( Y , dY , alpha = .5 , lw = .5 , color = cm ( abf . sweep / abf . sweeps ) )
plt . title ( "phase plot" )
plt . ylabel ( "dV (mV/ms)" )
plt . xlabel ( "V (mV)" )
plt . margins ( .1 , .1 )
# tighten up the figure
plt . tight_layout ( )


signature:
def plotAllSweeps ( abfFile ) :
    """simple example how to load an ABF file and plot every sweep ."""

implementation:
r = io . AxonIO ( filename = abfFile )
bl = r . read_block ( lazy = False , cascade = True )
print ( abfFile + "\nplotting %d sweeps..." % len ( bl . segments ) )
plt . figure ( figsize = ( 12 , 10 ) )
plt . title ( abfFile )
for sweep in range ( len ( bl . segments ) ) :
    trace = bl . segments [ sweep ] . analogsignals [ 0 ]
    plt . plot ( trace . times - trace . times [ 0 ] , trace . magnitude , alpha = .5 )
plt . ylabel ( trace . dimensionality )
plt . xlabel ( "seconds" )
plt . show ( )
plt . close ( )


signature:
def TIF_to_jpg ( fnameTiff , overwrite = False , saveAs = "" ) :
    """given a TIF taken by our cameras , make it a pretty labeled JPG .
 if the filename contains " f10 " or " f20 " , add appropraite scale bars .
 automatic contrast adjustment is different depending on if its a DIC
 image or fluorescent image ( which is detected automatically ) ."""

implementation:
if saveAs == "" :
    saveAs = fnameTiff + ".jpg"
if overwrite is False and os . path . exists ( saveAs ) :
    print ( "file exists, not overwriting..." )
    return
# load the image
img = pylab . imread ( fnameTiff )
img = img / np . max ( img )
# now the data is from 0 to 1
# determine the old histogram
hist1 , bins1 = np . histogram ( img . ravel ( ) , bins = 256 , range = ( 0 , 1 ) )
# pylab . plot ( bins [ : - 1 ] , hist )
# detect darkfield by average :
if np . average ( img ) < .2 :
    vmin = None
    vmax = None
    msg = " | FLU"
    while np . average ( img ) < .5 :
        img = np . sqrt ( img )
        msg += "^(.5)"
else :
    msg = " | DIC"
    percentile = .005
    vmin = np . percentile ( img . ravel ( ) , percentile )
    vmax = np . percentile ( img . ravel ( ) , 100 - percentile )
# determine the new histogram
hist2 , bins2 = np . histogram ( img . ravel ( ) , bins = 256 , range = ( 0 , 1 ) )
# plot it with resizing magic
fig = pylab . figure ( facecolor = 'r' )
fig . gca ( ) . imshow ( img , cmap = pylab . gray ( ) , vmin = vmin , vmax = vmax )
pylab . subplots_adjust ( top = 1 , bottom = 0 , right = 1 , left = 0 , hspace = 0 , wspace = 0 )
pylab . gca ( ) . xaxis . set_major_locator ( pylab . NullLocator ( ) )
pylab . gca ( ) . yaxis . set_major_locator ( pylab . NullLocator ( ) )
pylab . axis ( 'off' )
# resize it to the original size
fig . set_size_inches ( img . shape [ 1 ] / 100 , img . shape [ 0 ] / 100 )
# add text
msg = "%s | %s" % ( os . path . basename ( fnameTiff ) , datetime . datetime . fromtimestamp ( os . path . getmtime ( fnameTiff ) ) ) + msg
center = 10
pylab . text ( center , center , "%s" % ( msg ) , va = "top" , color = 'w' , size = 'small' , family = 'monospace' , weight = 'bold' , bbox = dict ( facecolor = 'k' , alpha = .5 ) )
# add scale bar
scaleWidthPx = False
if "f10" in fnameTiff :
    scaleWidthPx , scaleBarText = 39 , "25 um"
if "f20" in fnameTiff :
    scaleWidthPx , scaleBarText = 31 , "10 um"
if scaleWidthPx :
    scaleBarPadding = 10
    x2 , y2 = img . shape [ 1 ] - scaleBarPadding , img . shape [ 0 ] - scaleBarPadding
    x1 , y1 = x2 - scaleWidthPx , y2
    for offset , color , alpha in [ [ 2 , 'k' , .5 ] , [ 0 , 'w' , 1 ] ] :
        pylab . plot ( [ x1 + offset , x2 + offset ] , [ y1 + offset , y2 + offset ] , '-' , color = color , lw = 4 , alpha = alpha )
        pylab . text ( ( x1 + x2 ) / 2 + offset , y1 - 5 + offset , scaleBarText , color = color , ha = "center" , weight = "bold" , alpha = alpha , size = "small" , va = "bottom" , family = "monospace" )
# add histogram
# pylab . plot ( img . shape [ 1 ] - bins1 [ : - 1 ] [ : : - 1 ] * 200 , - hist1 / max ( hist1 ) * 100 + 110 , color = ' g ' )
# pylab . plot ( img . shape [ 1 ] - bins2 [ : - 1 ] [ : : - 1 ] * 200 , - hist2 / max ( hist2 ) * 100 + 110 , color = ' b ' )
# pylab . show ( )
# save it
pylab . savefig ( saveAs , dpi = 100 )
# clean up
pylab . close ( )


signature:
def TIF_to_jpg_all ( path ) :
    """run TIF _ to _ jpg ( ) on every TIF of a folder ."""

implementation:
for fname in sorted ( glob . glob ( path + "/*.tif" ) ) :
    print ( fname )
    TIF_to_jpg ( fname )


signature:
def analyzeSweep ( abf , sweep , m1 = None , m2 = None , plotToo = False ) :
    """m1 and m2 , if given , are in seconds .
 returns [ # EPSCs , # IPSCs ]"""

implementation:
abf . setsweep ( sweep )
if m1 is None :
    m1 = 0
else :
    m1 = m1 * abf . pointsPerSec
if m2 is None :
    m2 = - 1
else :
    m2 = m2 * abf . pointsPerSec
# obtain X and Y
Yorig = abf . sweepY [ int ( m1 ) : int ( m2 ) ]
X = np . arange ( len ( Yorig ) ) / abf . pointsPerSec
Ylpf = linear_gaussian ( Yorig , sigmaSize = abf . pointsPerMs * 300 , forwardOnly = False )
Yflat = Yorig - Ylpf
EPSCs , IPSCs = [ ] , [ ]
if plotToo :
    plt . figure ( figsize = ( 15 , 6 ) )
    ax1 = plt . subplot ( 211 )
    plt . title ( "%s sweep %d" % ( abf . ID , sweep ) )
    plt . grid ( )
    plt . plot ( X , Yorig , alpha = .5 )
    plt . plot ( X , Ylpf , 'k' , alpha = .5 , lw = 2 )
    plt . margins ( 0 , .2 )
    plt . subplot ( 212 , sharex = ax1 )
    plt . title ( "gaussian baseline subtraction" )
    plt . grid ( )
    plt . plot ( X , Yflat , alpha = .5 )
    plt . axhline ( 0 , color = 'k' , lw = 2 , alpha = .5 )
    plt . tight_layout ( )
    plt . show ( )
# TEST GAUSS
hist , bin_edges = np . histogram ( Yflat , density = True , bins = 200 )
peakPa = bin_edges [ np . where ( hist == max ( hist ) ) [ 0 ] [ 0 ] + 1 ]
if plotToo :
    plt . figure ( )
    plt . grid ( )
    plt . plot ( bin_edges [ 1 : ] , hist , alpha = .5 )
    plt . axvline ( 0 , color = 'k' )
    plt . axvline ( peakPa , color = 'r' , ls = '--' , lw = 2 , alpha = .5 )
    plt . semilogy ( )
    plt . title ( "sweep data distribution" )
    plt . ylabel ( "power" )
    plt . xlabel ( "pA deviation" )
    plt . show ( )
return peakPa


signature:
def convert ( fname , saveAs = True , showToo = False ) :
    """Convert weird TIF files into web - friendly versions .
 Auto contrast is applied ( saturating lower and upper 0.1 % ) .
 make saveAs True to save as . TIF . png
 make saveAs False and it won ' t save at all
 make saveAs " someFile . jpg " to save it as a different path / format"""

implementation:
# load the image
# im = Image . open ( fname ) # PIL can ' t handle 12 - bit TIFs well
im = ndimage . imread ( fname )
# scipy does better with it
im = np . array ( im , dtype = float )
# now it ' s a numpy array
# do all image enhancement here
cutoffLow = np . percentile ( im , .01 )
cutoffHigh = np . percentile ( im , 99.99 )
im [ np . where ( im < cutoffLow ) ] = cutoffLow
im [ np . where ( im > cutoffHigh ) ] = cutoffHigh
# IMAGE FORMATTING
im -= np . min ( im )
# auto contrast
im /= np . max ( im )
# normalize
im *= 255
# stretch contrast ( 8 - bit )
im = Image . fromarray ( im )
# IMAGE DRAWING
msg = "%s\n" % os . path . basename ( fname )
msg += "%s\n" % cm . epochToString ( os . path . getmtime ( fname ) )
d = ImageDraw . Draw ( im )
fnt = ImageFont . truetype ( "arial.ttf" , 20 )
d . text ( ( 6 , 6 ) , msg , font = fnt , fill = 0 )
d . text ( ( 4 , 4 ) , msg , font = fnt , fill = 255 )
if showToo :
    im . show ( )
if saveAs is False :
    return
if saveAs is True :
    saveAs = fname + ".png"
im . convert ( 'RGB' ) . save ( saveAs )
return saveAs


signature:
def plot_shaded_data ( X , Y , variances , varianceX ) :
    """plot X and Y data , then shade its background by variance ."""

implementation:
plt . plot ( X , Y , color = 'k' , lw = 2 )
nChunks = int ( len ( Y ) / CHUNK_POINTS )
for i in range ( 0 , 100 , PERCENT_STEP ) :
    varLimitLow = np . percentile ( variances , i )
    varLimitHigh = np . percentile ( variances , i + PERCENT_STEP )
    varianceIsAboveMin = np . where ( variances >= varLimitLow ) [ 0 ]
    varianceIsBelowMax = np . where ( variances <= varLimitHigh ) [ 0 ]
    varianceIsRange = [ chunkNumber for chunkNumber in range ( nChunks ) if chunkNumber in varianceIsAboveMin and chunkNumber in varianceIsBelowMax ]
    for chunkNumber in varianceIsRange :
        t1 = chunkNumber * CHUNK_POINTS / POINTS_PER_SEC
        t2 = t1 + CHUNK_POINTS / POINTS_PER_SEC
        plt . axvspan ( t1 , t2 , alpha = .3 , color = COLORMAP ( i / 100 ) , lw = 0 )


signature:
def show_variances ( Y , variances , varianceX , logScale = False ) :
    """create some fancy graphs to show color - coded variances ."""

implementation:
plt . figure ( 1 , figsize = ( 10 , 7 ) )
plt . figure ( 2 , figsize = ( 10 , 7 ) )
varSorted = sorted ( variances )
plt . figure ( 1 )
plt . subplot ( 211 )
plt . grid ( )
plt . title ( "chronological variance" )
plt . ylabel ( "original data" )
plot_shaded_data ( X , Y , variances , varianceX )
plt . margins ( 0 , .1 )
plt . subplot ( 212 )
plt . ylabel ( "variance (pA) (log%s)" % str ( logScale ) )
plt . xlabel ( "time in sweep (sec)" )
plt . plot ( varianceX , variances , 'k-' , lw = 2 )
plt . figure ( 2 )
plt . ylabel ( "variance (pA) (log%s)" % str ( logScale ) )
plt . xlabel ( "chunk number" )
plt . title ( "sorted variance" )
plt . plot ( varSorted , 'k-' , lw = 2 )
for i in range ( 0 , 100 , PERCENT_STEP ) :
    varLimitLow = np . percentile ( variances , i )
    varLimitHigh = np . percentile ( variances , i + PERCENT_STEP )
    label = "%2d-%d percentile" % ( i , i + + PERCENT_STEP )
    color = COLORMAP ( i / 100 )
    print ( "%s: variance = %.02f - %.02f" % ( label , varLimitLow , varLimitHigh ) )
    plt . figure ( 1 )
    plt . axhspan ( varLimitLow , varLimitHigh , alpha = .5 , lw = 0 , color = color , label = label )
    plt . figure ( 2 )
    chunkLow = np . where ( varSorted >= varLimitLow ) [ 0 ] [ 0 ]
    chunkHigh = np . where ( varSorted >= varLimitHigh ) [ 0 ] [ 0 ]
    plt . axvspan ( chunkLow , chunkHigh , alpha = .5 , lw = 0 , color = color , label = label )
for fignum in [ 1 , 2 ] :
    plt . figure ( fignum )
    if logScale :
        plt . semilogy ( )
    plt . margins ( 0 , 0 )
    plt . grid ( )
    if fignum is 2 :
        plt . legend ( fontsize = 10 , loc = 'upper left' , shadow = True )
    plt . tight_layout ( )
    plt . savefig ( '2016-12-15-variance-%d-log%s.png' % ( fignum , str ( logScale ) ) )
plt . show ( )


signature:
def ensureDetection ( self ) :
    """run this before analysis . Checks if event detection occured .
 If not , runs AP detection on all sweeps ."""

implementation:
if self . APs == False :
    self . log . debug ( "analysis attempted before event detection..." )
    self . detect ( )


signature:
def detect ( self ) :
    """runs AP detection on every sweep ."""

implementation:
self . log . info ( "initializing AP detection on all sweeps..." )
t1 = cm . timeit ( )
for sweep in range ( self . abf . sweeps ) :
    self . detectSweep ( sweep )
self . log . info ( "AP analysis of %d sweeps found %d APs (completed in %s)" , self . abf . sweeps , len ( self . APs ) , cm . timeit ( t1 ) )


signature:
def detectSweep ( self , sweep = 0 ) :
    """perform AP detection on current sweep ."""

implementation:
if self . APs is False : # indicates detection never happened
    self . APs = [ ]
# now indicates detection occured
# delete every AP from this sweep from the existing array
for i , ap in enumerate ( self . APs ) :
    if ap [ "sweep" ] == sweep :
        self . APs [ i ] = None
if self . APs . count ( None ) :
    self . log . debug ( "deleting %d existing APs from memory" , self . APs . count ( None ) )
    while None in self . APs :
        self . APs . remove ( None )
self . log . debug ( "initiating AP detection (%d already in memory)" , len ( self . APs ) )
self . abf . derivative = True
self . abf . setsweep ( sweep )
# detect potential AP ( Is ) by a dV / dT threshold crossing
Is = cm . where_cross ( self . abf . sweepD , self . detect_over )
self . log . debug ( "initial AP detection: %d APs" % len ( Is ) )
# eliminate APs where dV / dT doesn ' t cross below - 10 V / S within 2 ms
for i , I in enumerate ( Is ) :
    if np . min ( self . abf . sweepD [ I : I + 2 * self . abf . pointsPerMs ] ) > - 10 :
        Is [ i ] = 0
Is = Is [ np . nonzero ( Is ) ]
self . log . debug ( "after lower threshold checking: %d APs" % len ( Is ) )
# walk 1ms backwards and find point of + 10 V / S threshold crossing
for i , I in enumerate ( Is ) :
    stepBack = 0
    while ( self . abf . sweepD [ I - stepBack ] ) > 10 and stepBack / self . abf . pointsPerMs < 1 : # 2ms max
        stepBack += 1
    Is [ i ] -= stepBack
# analyze each AP
sweepAPs = [ ]
for i , I in enumerate ( Is ) :
    try :
        timeInSweep = I / self . abf . pointsPerSec
        if timeInSweep < self . detect_time1 or timeInSweep > self . detect_time2 :
            continue
            # skip because it ' s not within the marks
        ap = { }
        # create the AP entry
        ap [ "sweep" ] = sweep
        # number of the sweep containing this AP
        ap [ "I" ] = I
        # index sweep point of start of AP ( 10 mV / ms threshold crossing )
        ap [ "Tsweep" ] = I / self . abf . pointsPerSec
        # time in the sweep of index crossing ( sec )
        ap [ "T" ] = ap [ "Tsweep" ] + self . abf . sweepInterval * sweep
        # time in the experiment
        ap [ "Vthreshold" ] = self . abf . sweepY [ I ]
        # threshold at rate of - 10mV / ms
        # determine how many points from the start dV / dt goes below - 10 ( from a 5ms chunk )
        chunk = self . abf . sweepD [ I : I + 5 * self . abf . pointsPerMs ]
        # give it 5ms to cross once
        I_toNegTen = np . where ( chunk < - 10 ) [ 0 ] [ 0 ]
        chunk = self . abf . sweepD [ I + I_toNegTen : I + I_toNegTen + 10 * self . abf . pointsPerMs ]
        # give it 30ms to cross back
        if not max ( chunk ) > - 10 :
            self . log . debug ( "skipping unreal AP at T=%f" % ap [ "T" ] )
            self . log . error ( "^^^ can you confirm this is legit?" )
            continue
            # probably a pre - AP " bump " to be ignored
        I_recover = np . where ( chunk > - 10 ) [ 0 ] [ 0 ] + I_toNegTen + I
        # point where trace returns to above - 10 V / S
        ap [ "dVfastIs" ] = [ I , I_recover ]
        # span of the fast component of the dV / dt trace
        ap [ "dVfastMS" ] = ( I_recover - I ) / self . abf . pointsPerMs
        # time ( in ms ) of this fast AP component
        # determine derivative min / max from a 2ms chunk which we expect to capture the fast AP
        chunk = self . abf . sweepD [ ap [ "dVfastIs" ] [ 0 ] : ap [ "dVfastIs" ] [ 1 ] ]
        ap [ "dVmax" ] = np . max ( chunk )
        ap [ "dVmaxI" ] = np . where ( chunk == ap [ "dVmax" ] ) [ 0 ] [ 0 ] + I
        ap [ "dVmin" ] = np . min ( chunk )
        ap [ "dVminI" ] = np . where ( chunk == ap [ "dVmin" ] ) [ 0 ] [ 0 ] + I
        if ap [ "dVmax" ] < 10 or ap [ "dVmin" ] > - 10 :
            self . log . debug ( "throwing out AP with low dV/dt to be an AP" )
            self . log . error ( "^^^ can you confirm this is legit?" )
            continue
        # before determining AP shape stats , see where trace recovers to threshold
        chunkSize = self . abf . pointsPerMs * 10
        # AP shape may be 10ms
        if len ( Is ) - 1 > i and Is [ i + 1 ] < ( I + chunkSize ) : # if slow AP runs into next AP
            chunkSize = Is [ i + 1 ] - I
            # chop it down
        if chunkSize < ( self . abf . pointsPerMs * 2 ) :
            continue
            # next AP is so soon , it ' s > 500 Hz . Can ' t be real .
        ap [ "VslowIs" ] = [ I , I + chunkSize ]
        # time range of slow AP dynamics
        chunk = self . abf . sweepY [ I : I + chunkSize ]
        # determine AP peak and minimum
        ap [ "Vmax" ] = np . max ( chunk )
        ap [ "VmaxI" ] = np . where ( chunk == ap [ "Vmax" ] ) [ 0 ] [ 0 ] + I
        chunkForMin = np . copy ( chunk )
        # so we can destroy it
        chunkForMin [ : ap [ "VmaxI" ] - I ] = np . inf
        # minimum won ' t be before peak now
        ap [ "Vmin" ] = np . min ( chunkForMin )
        # supposedly the minimum is the AHP
        ap [ "VminI" ] = np . where ( chunkForMin == ap [ "Vmin" ] ) [ 0 ] [ 0 ] + I
        if ap [ "VminI" ] < ap [ "VmaxI" ] :
            self . log . error ( "-------------------------------" )
            self . log . error ( "how is the AHP before the peak?" )
            # TODO : start chunk at the peak
            self . log . error ( "-------------------------------" )
        # print ( ( I + len ( chunk ) ) - ap [ " VminI " ] , len ( chunk ) )
        if ( len ( chunk ) ) - ( ( I + len ( chunk ) ) - ap [ "VminI" ] ) < 10 :
            self . log . error ( "-------------------------------" )
            self . log . error ( "HP too close for comfort!" )
            self . log . error ( "-------------------------------" )
        ap [ "msRiseTime" ] = ( ap [ "VmaxI" ] - I ) / self . abf . pointsPerMs
        # time from threshold to peak
        ap [ "msFallTime" ] = ( ap [ "VminI" ] - ap [ "VmaxI" ] ) / self . abf . pointsPerMs
        # time from peak to nadir
        # determine halfwidth
        ap [ "Vhalf" ] = np . average ( [ ap [ "Vmax" ] , ap [ "Vthreshold" ] ] )
        # half way from threshold to peak
        ap [ "VhalfI1" ] = cm . where_cross ( chunk , ap [ "Vhalf" ] ) [ 0 ] + I
        # time it ' s first crossed
        ap [ "VhalfI2" ] = cm . where_cross ( - chunk , - ap [ "Vhalf" ] ) [ 1 ] + I
        # time it ' s second crossed
        ap [ "msHalfwidth" ] = ( ap [ "VhalfI2" ] - ap [ "VhalfI1" ] ) / self . abf . pointsPerMs
        # time between crossings
        # AP error checking goes here
        # TODO :
        # if we got this far , add the AP to the list
        sweepAPs . extend ( [ ap ] )
    except Exception as e :
        self . log . error ( "crashed analyzing AP %d of %d" , i , len ( Is ) )
        self . log . error ( cm . exceptionToString ( e ) )
        # cm . pause ( )
        # cm . waitFor ( 30)
        # self . log . error ( " EXCEPTION ! : \ n % s " % str ( sys . exc _ info ( ) ) )
self . log . debug ( "finished analyzing sweep. Found %d APs" , len ( sweepAPs ) )
self . APs . extend ( sweepAPs )
self . abf . derivative = False


signature:
def get_times ( self ) :
    """return an array of times ( in sec ) of all APs ."""

implementation:
self . ensureDetection ( )
times = [ ]
for ap in self . APs :
    times . append ( ap [ "T" ] )
return np . array ( sorted ( times ) )


signature:
def get_bySweep ( self , feature = "freqs" ) :
    """returns AP info by sweep arranged as a list ( by sweep ) .
 feature :
 * " freqs " - list of instantaneous frequencies by sweep .
 * " firsts " - list of first instantaneous frequency by sweep .
 * " times " - list of times of each AP in the sweep .
 * " count " - numer of APs per sweep .
 * " average " - average instanteous frequency per sweep .
 * " median " - median instanteous frequency per sweep ."""

implementation:
self . ensureDetection ( )
bySweepTimes = [ [ ] ] * self . abf . sweeps
# determine AP spike times by sweep
for sweep in range ( self . abf . sweeps ) :
    sweepTimes = [ ]
    for ap in self . APs :
        if ap [ "sweep" ] == sweep :
            sweepTimes . append ( ap [ "Tsweep" ] )
    bySweepTimes [ sweep ] = sweepTimes
# determine instantaneous frequencies by sweep
bySweepFreqs = [ [ ] ] * self . abf . sweeps
for i , times in enumerate ( bySweepTimes ) :
    if len ( times ) < 2 :
        continue
    diffs = np . array ( times [ 1 : ] ) - np . array ( times [ : - 1 ] )
    bySweepFreqs [ i ] = np . array ( 1 / diffs ) . tolist ( )
# give the user what they want
if feature == "freqs" :
    return bySweepFreqs
elif feature == "firsts" :
    result = np . zeros ( self . abf . sweeps )
    # initialize to this
    for i , freqs in enumerate ( bySweepFreqs ) :
        if len ( freqs ) :
            result [ i ] = freqs [ 0 ]
    return result
elif feature == "times" :
    return bySweepTimes
elif feature == "count" :
    result = np . zeros ( self . abf . sweeps )
    # initialize to this
    for i , times in enumerate ( bySweepTimes ) :
        result [ i ] = len ( bySweepTimes [ i ] )
    return result
elif feature == "average" :
    result = np . zeros ( self . abf . sweeps )
    # initialize to this
    for i , freqs in enumerate ( bySweepFreqs ) :
        if len ( freqs ) :
            result [ i ] = np . nanmean ( freqs )
    return result
elif feature == "median" :
    result = np . zeros ( self . abf . sweeps )
    # initialize to this
    for i , freqs in enumerate ( bySweepFreqs ) :
        if len ( freqs ) :
            result [ i ] = np . nanmedian ( freqs )
    return result
else :
    self . log . error ( "get_bySweep() can't handle [%s]" , feature )
    return None


signature:
def get_author_and_version ( package ) :
    """Return package author and version as listed in ` init . py ` ."""

implementation:
init_py = open ( os . path . join ( package , '__init__.py' ) ) . read ( )
author = re . search ( "__author__ = ['\"]([^'\"]+)['\"]" , init_py ) . group ( 1 )
version = re . search ( "__version__ = ['\"]([^'\"]+)['\"]" , init_py ) . group ( 1 )
return author , version


signature:
def api_subclass_factory ( name , docstring , remove_methods , base = SlackApi ) :
    """Create an API subclass with fewer methods than its base class .
 Arguments :
 name ( : py : class : ` str ` ) : The name of the new class .
 docstring ( : py : class : ` str ` ) : The docstring for the new class .
 remove _ methods ( : py : class : ` dict ` ) : The methods to remove from
 the base class ' s : py : attr : ` API _ METHODS ` for the subclass . The
 key is the name of the root method ( e . g . ` ` ' auth ' ` ` for
 ` ` ' auth . test ' ` ` , the value is either a tuple of child method
 names ( e . g . ` ` ( ' test ' , ) ` ` ) or , if all children should be
 removed , the special value : py : const : ` ALL ` .
 base ( : py : class : ` type ` , optional ) : The base class ( defaults to
 : py : class : ` SlackApi ` ) .
 Returns :
 : py : class : ` type ` : The new subclass .
 Raises :
 : py : class : ` KeyError ` : If the method wasn ' t in the superclass ."""

implementation:
methods = deepcopy ( base . API_METHODS )
for parent , to_remove in remove_methods . items ( ) :
    if to_remove is ALL :
        del methods [ parent ]
    else :
        for method in to_remove :
            del methods [ parent ] [ method ]
return type ( name , ( base , ) , dict ( API_METHODS = methods , __doc__ = docstring ) )


signature:
async def execute_method ( self , method , ** params ) :
    """Execute a specified Slack Web API method .
 Arguments :
 method ( : py : class : ` str ` ) : The name of the method .
 * * params ( : py : class : ` dict ` ) : Any additional parameters
 required .
 Returns :
 : py : class : ` dict ` : The JSON data from the response .
 Raises :
 : py : class : ` aiohttp . web _ exceptions . HTTPException ` : If the HTTP
 request returns a code other than 200 ( OK ) .
 SlackApiError : If the Slack API is reached but the response
 contains an error message ."""

implementation:
url = self . url_builder ( method , url_params = params )
logger . info ( 'Executing method %r' , method )
response = await aiohttp . get ( url )
logger . info ( 'Status: %r' , response . status )
if response . status == 200 :
    json = await response . json ( )
    logger . debug ( '...with JSON %r' , json )
    if json . get ( 'ok' ) :
        return json
    raise SlackApiError ( json [ 'error' ] )
else :
    raise_for_status ( response )


signature:
def method_exists ( cls , method ) :
    """Whether a given method exists in the known API .
 Arguments :
 method ( : py : class : ` str ` ) : The name of the method .
 Returns :
 : py : class : ` bool ` : Whether the method is in the known API ."""

implementation:
methods = cls . API_METHODS
for key in method . split ( '.' ) :
    methods = methods . get ( key )
    if methods is None :
        break
if isinstance ( methods , str ) :
    logger . debug ( '%r: %r' , method , methods )
    return True
return False


signature:
def _add_parsley_ns ( cls , namespace_dict ) :
    """Extend XPath evaluation with Parsley extensions ' namespace"""

implementation:
namespace_dict . update ( { 'parslepy' : cls . LOCAL_NAMESPACE , 'parsley' : cls . LOCAL_NAMESPACE , } )
return namespace_dict


signature:
def make ( self , selection ) :
    """XPath expression can also use EXSLT functions ( as long as they are
 understood by libxslt )"""

implementation:
cached = self . _selector_cache . get ( selection )
if cached :
    return cached
try :
    selector = lxml . etree . XPath ( selection , namespaces = self . namespaces , extensions = self . extensions , smart_strings = ( self . SMART_STRINGS or self . _test_smart_strings_needed ( selection ) ) , )
except lxml . etree . XPathSyntaxError as syntax_error :
    syntax_error . msg += ": %s" % selection
    raise syntax_error
except Exception as e :
    if self . DEBUG :
        print ( repr ( e ) , selection )
    raise
# wrap it / cache it
self . _selector_cache [ selection ] = Selector ( selector )
return self . _selector_cache [ selection ]


signature:
def extract ( self , document , selector , debug_offset = '' ) :
    """Try and convert matching Elements to unicode strings .
 If this fails , the selector evaluation probably already
 returned some string ( s ) of some sort , or boolean value ,
 or int / float , so return that instead ."""

implementation:
selected = self . select ( document , selector )
if selected is not None :
    if isinstance ( selected , ( list , tuple ) ) : # FIXME : return None or return empty list ?
        if not len ( selected ) :
            return
        return [ self . _extract_single ( m ) for m in selected ]
    else :
        return self . _extract_single ( selected )
# selector did not match anything
else :
    if self . DEBUG :
        print ( debug_offset , "selector did not match anything; return None" )
    return None


signature:
def make ( self , selection ) :
    """Scopes and selectors are tested in this order :
 * is this a CSS selector with an appended @ something attribute ?
 * is this a regular CSS selector ?
 * is this an XPath expression ?
 XPath expression can also use EXSLT functions ( as long as they are
 understood by libxslt )"""

implementation:
cached = self . _selector_cache . get ( selection )
if cached :
    return cached
namespaces = self . EXSLT_NAMESPACES
self . _add_parsley_ns ( namespaces )
try : # CSS with attribute ? ( non - standard but convenient )
# CSS selector cannot select attributes
# this " < css selector > @ < attr > " syntax is a Parsley extension
# construct CSS selector and append attribute to XPath expression
    m = self . REGEX_ENDING_ATTRIBUTE . match ( selection )
    if m : # the selector should be a regular CSS selector
        cssxpath = css_to_xpath ( m . group ( "expr" ) )
        # if " | " is used for namespace prefix reference ,
        # convert it to XPath prefix syntax
        attribute = m . group ( "attr" ) . replace ( '|' , ':' )
        cssxpath = "%s/%s" % ( cssxpath , attribute )
    else :
        cssxpath = css_to_xpath ( selection )
    selector = lxml . etree . XPath ( cssxpath , namespaces = self . namespaces , extensions = self . extensions , smart_strings = ( self . SMART_STRINGS or self . _test_smart_strings_needed ( selection ) ) , )
except tuple ( self . CSSSELECT_SYNTAXERROR_EXCEPTIONS ) as syntax_error :
    if self . DEBUG :
        print ( repr ( syntax_error ) , selection )
        print ( "Try interpreting as XPath selector" )
    try :
        selector = lxml . etree . XPath ( selection , namespaces = self . namespaces , extensions = self . extensions , smart_strings = ( self . SMART_STRINGS or self . _test_smart_strings_needed ( selection ) ) , )
    except lxml . etree . XPathSyntaxError as syntax_error :
        syntax_error . msg += ": %s" % selection
        raise syntax_error
    except Exception as e :
        if self . DEBUG :
            print ( repr ( e ) , selection )
        raise
# for exception when trying to convert < cssselector > @ < attribute > syntax
except lxml . etree . XPathSyntaxError as syntax_error :
    syntax_error . msg += ": %s" % selection
    raise syntax_error
except Exception as e :
    if self . DEBUG :
        print ( repr ( e ) , selection )
    raise
# wrap it / cache it
self . _selector_cache [ selection ] = Selector ( selector )
return self . _selector_cache [ selection ]


signature:
async def join_rtm ( self , filters = None ) :
    """Join the real - time messaging service .
 Arguments :
 filters ( : py : class : ` dict ` , optional ) : Dictionary mapping
 message filters to the functions they should dispatch to .
 Use a : py : class : ` collections . OrderedDict ` if precedence is
 important ; only one filter , the first match , will be
 applied to each message ."""

implementation:
if filters is None :
    filters = [ cls ( self ) for cls in self . MESSAGE_FILTERS ]
url = await self . _get_socket_url ( )
logger . debug ( 'Connecting to %r' , url )
async with ws_connect ( url ) as socket :
    first_msg = await socket . receive ( )
    self . _validate_first_message ( first_msg )
    self . socket = socket
    async for message in socket :
        if message . tp == MsgType . text :
            await self . handle_message ( message , filters )
        elif message . tp in ( MsgType . closed , MsgType . error ) :
            if not socket . closed :
                await socket . close ( )
            self . socket = None
            break
logger . info ( 'Left real-time messaging.' )


signature:
async def handle_message ( self , message , filters ) :
    """Handle an incoming message appropriately .
 Arguments :
 message ( : py : class : ` aiohttp . websocket . Message ` ) : The incoming
 message to handle .
 filters ( : py : class : ` list ` ) : The filters to apply to incoming
 messages ."""

implementation:
data = self . _unpack_message ( message )
logger . debug ( data )
if data . get ( 'type' ) == 'error' :
    raise SlackApiError ( data . get ( 'error' , { } ) . get ( 'msg' , str ( data ) ) )
elif self . message_is_to_me ( data ) :
    text = data [ 'text' ] [ len ( self . address_as ) : ] . strip ( )
    if text == 'help' :
        return self . _respond ( channel = data [ 'channel' ] , text = self . _instruction_list ( filters ) , )
    elif text == 'version' :
        return self . _respond ( channel = data [ 'channel' ] , text = self . VERSION , )
for _filter in filters :
    if _filter . matches ( data ) :
        logger . debug ( 'Response triggered' )
        async for response in _filter :
            self . _respond ( channel = data [ 'channel' ] , text = response )


signature:
def message_is_to_me ( self , data ) :
    """If you send a message directly to me"""

implementation: return ( data . get ( 'type' ) == 'message' and data . get ( 'text' , '' ) . startswith ( self . address_as ) )


signature:
async def from_api_token ( cls , token = None , api_cls = SlackBotApi ) :
    """Create a new instance from the API token .
 Arguments :
 token ( : py : class : ` str ` , optional ) : The bot ' s API token
 ( defaults to ` ` None ` ` , which means looking in the
 environment ) .
 api _ cls ( : py : class : ` type ` , optional ) : The class to create
 as the ` ` api ` ` argument for API access ( defaults to
 : py : class : ` aslack . slack _ api . SlackBotApi ` ) .
 Returns :
 : py : class : ` SlackBot ` : The new instance ."""

implementation:
api = api_cls . from_env ( ) if token is None else api_cls ( api_token = token )
data = await api . execute_method ( cls . API_AUTH_ENDPOINT )
return cls ( data [ 'user_id' ] , data [ 'user' ] , api )


signature:
def _format_message ( self , channel , text ) :
    """Format an outgoing message for transmission .
 Note :
 Adds the message type ( ` ` ' message ' ` ` ) and incremental ID .
 Arguments :
 channel ( : py : class : ` str ` ) : The channel to send to .
 text ( : py : class : ` str ` ) : The message text to send .
 Returns :
 : py : class : ` str ` : The JSON string of the message ."""

implementation:
payload = { 'type' : 'message' , 'id' : next ( self . _msg_ids ) }
payload . update ( channel = channel , text = text )
return json . dumps ( payload )


signature:
async def _get_socket_url ( self ) :
    """Get the WebSocket URL for the RTM session .
 Warning :
 The URL expires if the session is not joined within 30
 seconds of the API call to the start endpoint .
 Returns :
 : py : class : ` str ` : The socket URL ."""

implementation:
data = await self . api . execute_method ( self . RTM_START_ENDPOINT , simple_latest = True , no_unreads = True , )
return data [ 'url' ]


signature:
def _instruction_list ( self , filters ) :
    """Generates the instructions for a bot and its filters .
 Note :
 The guidance for each filter is generated by combining the
 docstrings of the predicate filter and resulting dispatch
 function with a single space between . The class ' s
 : py : attr : ` INSTRUCTIONS ` and the default help command are
 added .
 Arguments :
 filters ( : py : class : ` list ` ) : The filters to apply to incoming
 messages .
 Returns :
 : py : class : ` str ` : The bot ' s instructions ."""

implementation: return '\n\n' . join ( [ self . INSTRUCTIONS . strip ( ) , '*Supported methods:*' , 'If you send "@{}: help" to me I reply with these ' 'instructions.' . format ( self . user ) , 'If you send "@{}: version" to me I reply with my current ' 'version.' . format ( self . user ) , ] + [ filter . description ( ) for filter in filters ] )


signature:
def _respond ( self , channel , text ) :
    """Respond to a message on the current socket .
 Args :
 channel ( : py : class : ` str ` ) : The channel to send to .
 text ( : py : class : ` str ` ) : The message text to send ."""

implementation:
result = self . _format_message ( channel , text )
if result is not None :
    logger . info ( 'Sending message: %r' , truncate ( result , max_len = 50 ) , )
self . socket . send_str ( result )


signature:
def _validate_first_message ( cls , msg ) :
    """Check the first message matches the expected handshake .
 Note :
 The handshake is provided as : py : attr : ` RTM _ HANDSHAKE ` .
 Arguments :
 msg ( : py : class : ` aiohttp . Message ` ) : The message to validate .
 Raises :
 : py : class : ` SlackApiError ` : If the data doesn ' t match the
 expected handshake ."""

implementation:
data = cls . _unpack_message ( msg )
logger . debug ( data )
if data != cls . RTM_HANDSHAKE :
    raise SlackApiError ( 'Unexpected response: {!r}' . format ( data ) )
logger . info ( 'Joined real-time messaging.' )


signature:
def find_first_existing_executable ( exe_list ) :
    """Accepts list of [ ( ' executable _ file _ path ' , ' options ' ) ] ,
 Returns first working executable _ file _ path"""

implementation:
for filepath , opts in exe_list :
    try :
        proc = subprocess . Popen ( [ filepath , opts ] , stdout = subprocess . PIPE , stderr = subprocess . PIPE )
        proc . communicate ( )
    except OSError :
        pass
    else :
        return filepath