signature:
def transform ( self , jam ) :
    '''Bypass transformations .
 Parameters
 jam : pyjams . JAMS
 A muda - enabled JAMS object
 Yields
 jam _ out : pyjams . JAMS iterator
 The first result is ` jam ` ( unmodified ) , by reference
 All subsequent results are generated by ` transformer `'''

implementation:
# Step 1 : yield the unmodified jam
yield jam
# Step 2 : yield from the transformer
for jam_out in self . transformer . transform ( jam ) :
    yield jam_out


signature:
def __sox ( y , sr , * args ) :
    '''Execute sox
 Parameters
 y : np . ndarray
 Audio time series
 sr : int > 0
 Sampling rate of ` y `
 * args
 Additional arguments to sox
 Returns
 y _ out : np . ndarray
 ` y ` after sox transformation'''

implementation:
assert sr > 0
fdesc , infile = tempfile . mkstemp ( suffix = '.wav' )
os . close ( fdesc )
fdesc , outfile = tempfile . mkstemp ( suffix = '.wav' )
os . close ( fdesc )
# Dump the audio
librosa . output . write_wav ( infile , y , sr )
try :
    arguments = [ 'sox' , infile , outfile , '-q' ]
    arguments . extend ( args )
    subprocess . check_call ( arguments )
    y_out , sr = psf . read ( outfile )
    y_out = y_out . T
    if y . ndim == 1 :
        y_out = librosa . to_mono ( y_out )
finally :
    os . unlink ( infile )
    os . unlink ( outfile )
return y_out


signature:
def transpose ( label , n_semitones ) :
    '''Transpose a chord label by some number of semitones
 Parameters
 label : str
 A chord string
 n _ semitones : float
 The number of semitones to move ` label `
 Returns
 label _ transpose : str
 The transposed chord label'''

implementation:
# Otherwise , split off the note from the modifier
match = re . match ( six . text_type ( '(?P<note>[A-G][b#]*)(?P<mod>.*)' ) , six . text_type ( label ) )
if not match :
    return label
note = match . group ( 'note' )
new_note = librosa . midi_to_note ( librosa . note_to_midi ( note ) + n_semitones , octave = False )
return new_note + match . group ( 'mod' )


signature:
def jam_pack ( jam , ** kwargs ) :
    '''Pack data into a jams sandbox .
 If not already present , this creates a ` muda ` field within ` jam . sandbox ` ,
 along with ` history ` , ` state ` , and version arrays which are populated by
 deformation objects .
 Any additional fields can be added to the ` muda ` sandbox by supplying
 keyword arguments .
 Parameters
 jam : jams . JAMS
 A JAMS object
 Returns
 jam : jams . JAMS
 The updated JAMS object
 Examples
 > > > jam = jams . JAMS ( )
 > > > muda . jam _ pack ( jam , my _ data = dict ( foo = 5 , bar = None ) )
 > > > jam . sandbox
 < Sandbox : muda >
 > > > jam . sandbox . muda
 < Sandbox : state , version , my _ data , history >
 > > > jam . sandbox . muda . my _ data
 { ' foo ' : 5 , ' bar ' : None }'''

implementation:
if not hasattr ( jam . sandbox , 'muda' ) : # If there ' s no mudabox , create one
    jam . sandbox . muda = jams . Sandbox ( history = [ ] , state = [ ] , version = dict ( muda = version , librosa = librosa . __version__ , jams = jams . __version__ , pysoundfile = psf . __version__ ) )
elif not isinstance ( jam . sandbox . muda , jams . Sandbox ) : # If there is a muda entry , but it ' s not a sandbox , coerce it
    jam . sandbox . muda = jams . Sandbox ( ** jam . sandbox . muda )
jam . sandbox . muda . update ( ** kwargs )
return jam


signature:
def load_jam_audio ( jam_in , audio_file , validate = True , strict = True , fmt = 'auto' , ** kwargs ) :
    '''Load a jam and pack it with audio .
 Parameters
 jam _ in : str , file descriptor , or jams . JAMS
 JAMS filename , open file - descriptor , or object to load .
 See ` ` jams . load ` ` for acceptable formats .
 audio _ file : str
 Audio filename to load
 validate : bool
 strict : bool
 fmt : str
 Parameters to ` jams . load `
 kwargs : additional keyword arguments
 See ` librosa . load `
 Returns
 jam : jams . JAMS
 A jams object with audio data in the top - level sandbox
 Notes
 This operation can modify the ` file _ metadata . duration ` field of ` jam _ in ` :
 If it is not currently set , it will be populated with the duration of the
 audio file .
 See Also
 jams . load
 librosa . core . load'''

implementation:
if isinstance ( jam_in , jams . JAMS ) :
    jam = jam_in
else :
    jam = jams . load ( jam_in , validate = validate , strict = strict , fmt = fmt )
y , sr = librosa . load ( audio_file , ** kwargs )
if jam . file_metadata . duration is None :
    jam . file_metadata . duration = librosa . get_duration ( y = y , sr = sr )
return jam_pack ( jam , _audio = dict ( y = y , sr = sr ) )


signature:
def save ( filename_audio , filename_jam , jam , strict = True , fmt = 'auto' , ** kwargs ) :
    '''Save a muda jam to disk
 Parameters
 filename _ audio : str
 The path to store the audio file
 filename _ jam : str
 The path to store the jams object
 strict : bool
 Strict safety checking for jams output
 fmt : str
 Output format parameter for ` jams . JAMS . save `
 kwargs
 Additional parameters to ` soundfile . write `'''

implementation:
y = jam . sandbox . muda . _audio [ 'y' ]
sr = jam . sandbox . muda . _audio [ 'sr' ]
# First , dump the audio file
psf . write ( filename_audio , y , sr , ** kwargs )
# Then dump the jam
jam . save ( filename_jam , strict = strict , fmt = fmt )


signature:
def __reconstruct ( params ) :
    '''Reconstruct a transformation or pipeline given a parameter dump .'''

implementation:
if isinstance ( params , dict ) :
    if '__class__' in params :
        cls = params [ '__class__' ]
        data = __reconstruct ( params [ 'params' ] )
        return cls ( ** data )
    else :
        data = dict ( )
        for key , value in six . iteritems ( params ) :
            data [ key ] = __reconstruct ( value )
        return data
elif isinstance ( params , ( list , tuple ) ) :
    return [ __reconstruct ( v ) for v in params ]
else :
    return params


signature:
def serialize ( transform , ** kwargs ) :
    '''Serialize a transformation object or pipeline .
 Parameters
 transform : BaseTransform or Pipeline
 The transformation object to be serialized
 kwargs
 Additional keyword arguments to ` jsonpickle . encode ( ) `
 Returns
 json _ str : str
 A JSON encoding of the transformation
 See Also
 deserialize
 Examples
 > > > D = muda . deformers . TimeStretch ( rate = 1.5)
 > > > muda . serialize ( D )
 ' { " params " : { " rate " : 1.5 } ,
 " _ _ class _ _ " : { " py / type " : " muda . deformers . time . TimeStretch " } } ' '''

implementation:
params = transform . get_params ( )
return jsonpickle . encode ( params , ** kwargs )


signature:
def deserialize ( encoded , ** kwargs ) :
    '''Construct a muda transformation from a JSON encoded string .
 Parameters
 encoded : str
 JSON encoding of the transformation or pipeline
 kwargs
 Additional keyword arguments to ` jsonpickle . decode ( ) `
 Returns
 obj
 The transformation
 See Also
 serialize
 Examples
 > > > D = muda . deformers . TimeStretch ( rate = 1.5)
 > > > D _ serial = muda . serialize ( D )
 > > > D2 = muda . deserialize ( D _ serial )
 > > > D2
 TimeStretch ( rate = 1.5)'''

implementation:
params = jsonpickle . decode ( encoded , ** kwargs )
return __reconstruct ( params )


signature:
def _pprint ( params , offset = 0 , printer = repr ) :
    """Pretty print the dictionary ' params '
 Parameters
 params : dict
 The dictionary to pretty print
 offset : int
 The offset in characters to add at the begin of each line .
 printer :
 The function to convert entries to strings , typically
 the builtin str or repr"""

implementation:
# Do a multi - line justified repr :
options = np . get_printoptions ( )
np . set_printoptions ( precision = 5 , threshold = 64 , edgeitems = 2 )
params_list = list ( )
this_line_length = offset
line_sep = ',\n' + ( 1 + offset // 2 ) * ' '
for i , ( k , v ) in enumerate ( sorted ( six . iteritems ( params ) ) ) :
    if type ( v ) is float : # use str for representing floating point numbers
    # this way we get consistent representation across
    # architectures and versions .
        this_repr = '%s=%s' % ( k , str ( v ) )
    else : # use repr of the rest
        this_repr = '%s=%s' % ( k , printer ( v ) )
    if len ( this_repr ) > 500 :
        this_repr = this_repr [ : 300 ] + '...' + this_repr [ - 100 : ]
    if i > 0 :
        if ( this_line_length + len ( this_repr ) >= 75 or '\n' in this_repr ) :
            params_list . append ( line_sep )
            this_line_length = len ( line_sep )
        else :
            params_list . append ( ', ' )
            this_line_length += 2
    params_list . append ( this_repr )
    this_line_length += len ( this_repr )
np . set_printoptions ( ** options )
lines = '' . join ( params_list )
# Strip trailing space to avoid nightmare in doctests
lines = '\n' . join ( l . rstrip ( ' ' ) for l in lines . split ( '\n' ) )
return lines


signature:
def _get_param_names ( cls ) :
    '''Get the list of parameter names for the object'''

implementation:
init = cls . __init__
args , varargs = inspect . getargspec ( init ) [ : 2 ]
if varargs is not None :
    raise RuntimeError ( 'BaseTransformer objects cannot have varargs' )
args . pop ( 0 )
args . sort ( )
return args


signature:
def get_params ( self , deep = True ) :
    '''Get the parameters for this object . Returns as a dict .
 Parameters
 deep : bool
 Recurse on nested objects
 Returns
 params : dict
 A dictionary containing all parameters for this object'''

implementation:
out = dict ( __class__ = self . __class__ , params = dict ( ) )
for key in self . _get_param_names ( ) :
    value = getattr ( self , key , None )
    if deep and hasattr ( value , 'get_params' ) :
        deep_items = value . get_params ( ) . items ( )
        out [ 'params' ] [ key ] = dict ( __class__ = value . __class__ )
        out [ 'params' ] [ key ] . update ( ( k , val ) for k , val in deep_items )
    else :
        out [ 'params' ] [ key ] = value
return out


signature:
def _transform ( self , jam , state ) :
    '''Apply the transformation to audio and annotations .
 The input jam is copied and modified , and returned
 contained in a list .
 Parameters
 jam : jams . JAMS
 A single jam object to modify
 Returns
 jam _ list : list
 A length - 1 list containing ` jam ` after transformation
 See also
 core . load _ jam _ audio'''

implementation:
if not hasattr ( jam . sandbox , 'muda' ) :
    raise RuntimeError ( 'No muda state found in jams sandbox.' )
# We ' ll need a working copy of this object for modification purposes
jam_w = copy . deepcopy ( jam )
# Push our reconstructor onto the history stack
jam_w . sandbox . muda [ 'history' ] . append ( { 'transformer' : self . __serialize__ , 'state' : state } )
if hasattr ( self , 'audio' ) :
    self . audio ( jam_w . sandbox . muda , state )
if hasattr ( self , 'metadata' ) :
    self . metadata ( jam_w . file_metadata , state )
# Walk over the list of deformers
for query , function_name in six . iteritems ( self . dispatch ) :
    function = getattr ( self , function_name )
    for matched_annotation in jam_w . search ( namespace = query ) :
        function ( matched_annotation , state )
return jam_w


signature:
def transform ( self , jam ) :
    '''Iterative transformation generator
 Applies the deformation to an input jams object .
 This generates a sequence of deformed output JAMS .
 Parameters
 jam : jams . JAMS
 The jam to transform
 Examples
 > > > for jam _ out in deformer . transform ( jam _ in ) :
 . . . process ( jam _ out )'''

implementation:
for state in self . states ( jam ) :
    yield self . _transform ( jam , state )


signature:
def get_params ( self ) :
    '''Get the parameters for this object . Returns as a dict .'''

implementation:
out = { }
out [ '__class__' ] = self . __class__
out [ 'params' ] = dict ( steps = [ ] )
for name , step in self . steps :
    out [ 'params' ] [ 'steps' ] . append ( [ name , step . get_params ( deep = True ) ] )
return out


signature:
def __recursive_transform ( self , jam , steps ) :
    '''A recursive transformation pipeline'''

implementation:
if len ( steps ) > 0 :
    head_transformer = steps [ 0 ] [ 1 ]
    for t_jam in head_transformer . transform ( jam ) :
        for q in self . __recursive_transform ( t_jam , steps [ 1 : ] ) :
            yield q
else :
    yield jam


signature:
def transform ( self , jam ) :
    '''Apply the sequence of transformations to a single jam object .
 Parameters
 jam : jams . JAMS
 The jam object to transform
 Yields
 jam _ out : jams . JAMS
 The jam objects produced by the transformation sequence'''

implementation:
for output in self . __recursive_transform ( jam , self . steps ) :
    yield output


signature:
def __serial_transform ( self , jam , steps ) :
    '''A serial transformation union'''

implementation:
# This uses the round - robin itertools recipe
if six . PY2 :
    attr = 'next'
else :
    attr = '__next__'
pending = len ( steps )
nexts = itertools . cycle ( getattr ( iter ( D . transform ( jam ) ) , attr ) for ( name , D ) in steps )
while pending :
    try :
        for next_jam in nexts :
            yield next_jam ( )
    except StopIteration :
        pending -= 1
        nexts = itertools . cycle ( itertools . islice ( nexts , pending ) )


signature:
def transform ( self , jam ) :
    '''Apply the sequence of transformations to a single jam object .
 Parameters
 jam : jams . JAMS
 The jam object to transform
 Yields
 jam _ out : jams . JAMS
 The jam objects produced by each member of the union'''

implementation:
for output in self . __serial_transform ( jam , self . steps ) :
    yield output


signature:
def sample_clip_indices ( filename , n_samples , sr ) :
    '''Calculate the indices at which to sample a fragment of audio from a file .
 Parameters
 filename : str
 Path to the input file
 n _ samples : int > 0
 The number of samples to load
 sr : int > 0
 The target sampling rate
 Returns
 start : int
 The sample index from ` filename ` at which the audio fragment starts
 stop : int
 The sample index from ` filename ` at which the audio fragment stops ( e . g . y = audio [ start : stop ] )'''

implementation:
with psf . SoundFile ( str ( filename ) , mode = 'r' ) as soundf : # Measure required length of fragment
    n_target = int ( np . ceil ( n_samples * soundf . samplerate / float ( sr ) ) )
    # Raise exception if source is too short
    if len ( soundf ) < n_target :
        raise RuntimeError ( 'Source {} (length={})' . format ( filename , len ( soundf ) ) + ' must be at least the length of the input ({})' . format ( n_target ) )
    # Draw a starting point at random in the background waveform
    start = np . random . randint ( 0 , 1 + len ( soundf ) - n_target )
    stop = start + n_target
    return start , stop


signature:
def slice_clip ( filename , start , stop , n_samples , sr , mono = True ) :
    '''Slice a fragment of audio from a file .
 This uses pysoundfile to efficiently seek without
 loading the entire stream .
 Parameters
 filename : str
 Path to the input file
 start : int
 The sample index of ` filename ` at which the audio fragment should start
 stop : int
 The sample index of ` filename ` at which the audio fragment should stop ( e . g . y = audio [ start : stop ] )
 n _ samples : int > 0
 The number of samples to load
 sr : int > 0
 The target sampling rate
 mono : bool
 Ensure monophonic audio
 Returns
 y : np . ndarray [ shape = ( n _ samples , ) ]
 A fragment of audio sampled from ` filename `
 Raises
 ValueError
 If the source file is shorter than the requested length'''

implementation:
with psf . SoundFile ( str ( filename ) , mode = 'r' ) as soundf :
    n_target = stop - start
    soundf . seek ( start )
    y = soundf . read ( n_target ) . T
    if mono :
        y = librosa . to_mono ( y )
    # Resample to initial sr
    y = librosa . resample ( y , soundf . samplerate , sr )
    # Clip to the target length exactly
    y = librosa . util . fix_length ( y , n_samples )
    return y


signature:
def norm_remote_path ( path ) :
    """Normalize ` path ` .
 All remote paths are absolute ."""

implementation:
path = os . path . normpath ( path )
if path . startswith ( os . path . sep ) :
    return path [ 1 : ]
else :
    return path


signature:
def split_storage ( path , default = 'osfstorage' ) :
    """Extract storage name from file path .
 If a path begins with a known storage provider the name is removed
 from the path . Otherwise the ` default ` storage provider is returned
 and the path is not modified ."""

implementation:
path = norm_remote_path ( path )
for provider in KNOWN_PROVIDERS :
    if path . startswith ( provider + '/' ) :
        if six . PY3 :
            return path . split ( '/' , maxsplit = 1 )
        else :
            return path . split ( '/' , 1 )
return ( default , path )


signature:
def file_empty ( fp ) :
    """Determine if a file is empty or not ."""

implementation:
# for python 2 we need to use a homemade peek ( )
if six . PY2 :
    contents = fp . read ( )
    fp . seek ( 0 )
    return not bool ( contents )
else :
    return not fp . peek ( )


signature:
def checksum ( file_path , hash_type = 'md5' , block_size = 65536 ) :
    """Returns either the md5 or sha256 hash of a file at ` file _ path ` .
 md5 is the default hash _ type as it is faster than sha256
 The default block size is 64 kb , which appears to be one of a few command
 choices according to https : / / stackoverflow . com / a / 44873382/2680 . The code
 below is an extension of the example presented in that post ."""

implementation:
if hash_type == 'md5' :
    hash_ = hashlib . md5 ( )
elif hash_type == 'sha256' :
    hash_ = hashlib . sha256 ( )
else :
    raise ValueError ( "{} is an invalid hash_type. Expected 'md5' or 'sha256'." . format ( hash_type ) )
with open ( file_path , 'rb' ) as f :
    for block in iter ( lambda : f . read ( block_size ) , b'' ) :
        hash_ . update ( block )
return hash_ . hexdigest ( )


signature:
def storage ( self , provider = 'osfstorage' ) :
    """Return storage ` provider ` ."""

implementation:
stores = self . _json ( self . _get ( self . _storages_url ) , 200 )
stores = stores [ 'data' ]
for store in stores :
    provides = self . _get_attribute ( store , 'attributes' , 'provider' )
    if provides == provider :
        return Storage ( store , self . session )
raise RuntimeError ( "Project has no storage " "provider '{}'" . format ( provider ) )


signature:
def storages ( self ) :
    """Iterate over all storages for this projects ."""

implementation:
stores = self . _json ( self . _get ( self . _storages_url ) , 200 )
stores = stores [ 'data' ]
for store in stores :
    yield Storage ( store , self . session )


signature:
def create_file ( self , path , fp , force = False , update = False ) :
    """Store a new file at ` path ` in this storage .
 The contents of the file descriptor ` fp ` ( opened in ' rb ' mode )
 will be uploaded to ` path ` which is the full path at
 which to store the file .
 To force overwrite of an existing file , set ` force = True ` .
 To overwrite an existing file only if the files differ , set ` update = True `"""

implementation:
if 'b' not in fp . mode :
    raise ValueError ( "File has to be opened in binary mode." )
# all paths are assumed to be absolute
path = norm_remote_path ( path )
directory , fname = os . path . split ( path )
directories = directory . split ( os . path . sep )
# navigate to the right parent object for our file
parent = self
for directory in directories : # skip empty directory names
    if directory :
        parent = parent . create_folder ( directory , exist_ok = True )
url = parent . _new_file_url
# When uploading a large file ( > a few MB ) that already exists
# we sometimes get a ConnectionError instead of a status = = 409.
connection_error = False
# peek at the file to check if it is an empty file which needs special
# handling in requests . If we pass a file like object to data that
# turns out to be of length zero then no file is created on the OSF .
# See : https : / / github . com / osfclient / osfclient / pull / 135
if file_empty ( fp ) :
    response = self . _put ( url , params = { 'name' : fname } , data = b'' )
else :
    try :
        response = self . _put ( url , params = { 'name' : fname } , data = fp )
    except ConnectionError :
        connection_error = True
if connection_error or response . status_code == 409 :
    if not force and not update : # one - liner to get file size from file pointer from
    # https : / / stackoverflow . com / a / 283719/2680824
        file_size_bytes = get_local_file_size ( fp )
        large_file_cutoff = 2 ** 20
        # 1 MB in bytes
        if connection_error and file_size_bytes < large_file_cutoff :
            msg = ( "There was a connection error which might mean {} " + "already exists. Try again with the `--force` flag " + "specified." ) . format ( path )
            raise RuntimeError ( msg )
        else : # note in case of connection error , we are making an inference here
            raise FileExistsError ( path )
    else : # find the upload URL for the file we are trying to update
        for file_ in self . files :
            if norm_remote_path ( file_ . path ) == path :
                if not force :
                    if checksum ( path ) == file_ . hashes . get ( 'md5' ) : # If the hashes are equal and force is False ,
                    # we ' re done here
                        break
                # in the process of attempting to upload the file we
                # moved through it - > reset read position to beginning
                # of the file
                fp . seek ( 0 )
                file_ . update ( fp )
                break
        else :
            raise RuntimeError ( "Could not create a new file at " "({}) nor update it." . format ( path ) )


signature:
def copyfileobj ( fsrc , fdst , total , length = 16 * 1024 ) :
    """Copy data from file - like object fsrc to file - like object fdst
 This is like shutil . copyfileobj but with a progressbar ."""

implementation:
with tqdm ( unit = 'bytes' , total = total , unit_scale = True ) as pbar :
    while 1 :
        buf = fsrc . read ( length )
        if not buf :
            break
        fdst . write ( buf )
        pbar . update ( len ( buf ) )


signature:
def write_to ( self , fp ) :
    """Write contents of this file to a local file .
 Pass in a filepointer ` fp ` that has been opened for writing in
 binary mode ."""

implementation:
if 'b' not in fp . mode :
    raise ValueError ( "File has to be opened in binary mode." )
response = self . _get ( self . _download_url , stream = True )
if response . status_code == 200 :
    response . raw . decode_content = True
    copyfileobj ( response . raw , fp , int ( response . headers [ 'Content-Length' ] ) )
else :
    raise RuntimeError ( "Response has status " "code {}." . format ( response . status_code ) )


signature:
def remove ( self ) :
    """Remove this file from the remote storage ."""

implementation:
response = self . _delete ( self . _delete_url )
if response . status_code != 204 :
    raise RuntimeError ( 'Could not delete {}.' . format ( self . path ) )


signature:
def update ( self , fp ) :
    """Update the remote file from a local file .
 Pass in a filepointer ` fp ` that has been opened for writing in
 binary mode ."""

implementation:
if 'b' not in fp . mode :
    raise ValueError ( "File has to be opened in binary mode." )
url = self . _upload_url
# peek at the file to check if it is an ampty file which needs special
# handling in requests . If we pass a file like object to data that
# turns out to be of length zero then no file is created on the OSF
if fp . peek ( 1 ) :
    response = self . _put ( url , data = fp )
else :
    response = self . _put ( url , data = b'' )
if response . status_code != 200 :
    msg = ( 'Could not update {} (status ' 'code: {}).' . format ( self . path , response . status_code ) )
    raise RuntimeError ( msg )


signature:
def _iter_children ( self , url , kind , klass , recurse = None ) :
    """Iterate over all children of ` kind `
 Yield an instance of ` klass ` when a child is of type ` kind ` . Uses
 ` recurse ` as the path of attributes in the JSON returned from ` url `
 to find more children ."""

implementation:
children = self . _follow_next ( url )
while children :
    child = children . pop ( )
    kind_ = child [ 'attributes' ] [ 'kind' ]
    if kind_ == kind :
        yield klass ( child , self . session )
    elif recurse is not None : # recurse into a child and add entries to ` children `
        url = self . _get_attribute ( child , * recurse )
        children . extend ( self . _follow_next ( url ) )


signature:
def might_need_auth ( f ) :
    """Decorate a CLI function that might require authentication .
 Catches any UnauthorizedException raised , prints a helpful message and
 then exits ."""

implementation:
@ wraps ( f )
def wrapper ( cli_args ) :
    try :
        return_value = f ( cli_args )
    except UnauthorizedException as e :
        config = config_from_env ( config_from_file ( ) )
        username = _get_username ( cli_args , config )
        if username is None :
            sys . exit ( "Please set a username (run `osf -h` for details)." )
        else :
            sys . exit ( "You are not authorized to access this project." )
    return return_value
return wrapper


signature:
def init ( args ) :
    """Initialize or edit an existing . osfcli . config file ."""

implementation:
# reading existing config file , convert to configparser object
config = config_from_file ( )
config_ = configparser . ConfigParser ( )
config_ . add_section ( 'osf' )
if 'username' not in config . keys ( ) :
    config_ . set ( 'osf' , 'username' , '' )
else :
    config_ . set ( 'osf' , 'username' , config [ 'username' ] )
if 'project' not in config . keys ( ) :
    config_ . set ( 'osf' , 'project' , '' )
else :
    config_ . set ( 'osf' , 'project' , config [ 'project' ] )
# now we can start asking for new values
print ( 'Provide a username for the config file [current username: {}]:' . format ( config_ . get ( 'osf' , 'username' ) ) )
username = input ( )
if username :
    config_ . set ( 'osf' , 'username' , username )
print ( 'Provide a project for the config file [current project: {}]:' . format ( config_ . get ( 'osf' , 'project' ) ) )
project = input ( )
if project :
    config_ . set ( 'osf' , 'project' , project )
cfgfile = open ( ".osfcli.config" , "w" )
config_ . write ( cfgfile )
cfgfile . close ( )


signature:
def clone ( args ) :
    """Copy all files from all storages of a project .
 The output directory defaults to the current directory .
 If the project is private you need to specify a username .
 If args . update is True , overwrite any existing local files only if local and
 remote files differ ."""

implementation:
osf = _setup_osf ( args )
project = osf . project ( args . project )
output_dir = args . project
if args . output is not None :
    output_dir = args . output
with tqdm ( unit = 'files' ) as pbar :
    for store in project . storages :
        prefix = os . path . join ( output_dir , store . name )
        for file_ in store . files :
            path = file_ . path
            if path . startswith ( '/' ) :
                path = path [ 1 : ]
            path = os . path . join ( prefix , path )
            if os . path . exists ( path ) and args . update :
                if checksum ( path ) == file_ . hashes . get ( 'md5' ) :
                    continue
            directory , _ = os . path . split ( path )
            makedirs ( directory , exist_ok = True )
            with open ( path , "wb" ) as f :
                file_ . write_to ( f )
            pbar . update ( )


signature:
def fetch ( args ) :
    """Fetch an individual file from a project .
 The first part of the remote path is interpreted as the name of the
 storage provider . If there is no match the default ( osfstorage ) is
 used .
 The local path defaults to the name of the remote file .
 If the project is private you need to specify a username .
 If args . force is True , write local file even if that file already exists .
 If args . force is False but args . update is True , overwrite an existing local
 file only if local and remote files differ ."""

implementation:
storage , remote_path = split_storage ( args . remote )
local_path = args . local
if local_path is None :
    _ , local_path = os . path . split ( remote_path )
local_path_exists = os . path . exists ( local_path )
if local_path_exists and not args . force and not args . update :
    sys . exit ( "Local file %s already exists, not overwriting." % local_path )
directory , _ = os . path . split ( local_path )
if directory :
    makedirs ( directory , exist_ok = True )
osf = _setup_osf ( args )
project = osf . project ( args . project )
store = project . storage ( storage )
for file_ in store . files :
    if norm_remote_path ( file_ . path ) == remote_path :
        if local_path_exists and not args . force and args . update :
            if file_ . hashes . get ( 'md5' ) == checksum ( local_path ) :
                print ( "Local file %s already matches remote." % local_path )
                break
        with open ( local_path , 'wb' ) as fp :
            file_ . write_to ( fp )
        # only fetching one file so we are done
        break


signature:
def list_ ( args ) :
    """List all files from all storages for project .
 If the project is private you need to specify a username ."""

implementation:
osf = _setup_osf ( args )
project = osf . project ( args . project )
for store in project . storages :
    prefix = store . name
    for file_ in store . files :
        path = file_ . path
        if path . startswith ( '/' ) :
            path = path [ 1 : ]
        print ( os . path . join ( prefix , path ) )


signature:
def upload ( args ) :
    """Upload a new file to an existing project .
 The first part of the remote path is interpreted as the name of the
 storage provider . If there is no match the default ( osfstorage ) is
 used .
 If the project is private you need to specify a username .
 To upload a whole directory ( and all its sub - directories ) use the ` - r `
 command - line option . If your source directory name ends in a / then
 files will be created directly in the remote directory . If it does not
 end in a slash an extra sub - directory with the name of the local directory
 will be created .
 To place contents of local directory ` foo ` in remote directory ` bar / foo ` :
 $ osf upload - r foo bar
 To place contents of local directory ` foo ` in remote directory ` bar ` :
 $ osf upload - r foo / bar"""

implementation:
osf = _setup_osf ( args )
if osf . username is None or osf . password is None :
    sys . exit ( 'To upload a file you need to provide a username and' ' password.' )
project = osf . project ( args . project )
storage , remote_path = split_storage ( args . destination )
store = project . storage ( storage )
if args . recursive :
    if not os . path . isdir ( args . source ) :
        raise RuntimeError ( "Expected source ({}) to be a directory when " "using recursive mode." . format ( args . source ) )
    # local name of the directory that is being uploaded
    _ , dir_name = os . path . split ( args . source )
    for root , _ , files in os . walk ( args . source ) :
        subdir_path = os . path . relpath ( root , args . source )
        for fname in files :
            local_path = os . path . join ( root , fname )
            with open ( local_path , 'rb' ) as fp : # build the remote path + fname
                name = os . path . join ( remote_path , dir_name , subdir_path , fname )
                store . create_file ( name , fp , force = args . force , update = args . update )
else :
    with open ( args . source , 'rb' ) as fp :
        store . create_file ( remote_path , fp , force = args . force , update = args . update )


signature:
def remove ( args ) :
    """Remove a file from the project ' s storage .
 The first part of the remote path is interpreted as the name of the
 storage provider . If there is no match the default ( osfstorage ) is
 used ."""

implementation:
osf = _setup_osf ( args )
if osf . username is None or osf . password is None :
    sys . exit ( 'To remove a file you need to provide a username and' ' password.' )
project = osf . project ( args . project )
storage , remote_path = split_storage ( args . target )
store = project . storage ( storage )
for f in store . files :
    if norm_remote_path ( f . path ) == remote_path :
        f . remove ( )


signature:
def login ( self , username , password = None , token = None ) :
    """Login user for protected API calls ."""

implementation: self . session . basic_auth ( username , password )


signature:
def project ( self , project_id ) :
    """Fetch project ` project _ id ` ."""

implementation:
type_ = self . guid ( project_id )
url = self . _build_url ( type_ , project_id )
if type_ in Project . _types :
    return Project ( self . _json ( self . _get ( url ) , 200 ) , self . session )
raise OSFException ( '{} is unrecognized type {}. Clone supports projects and registrations' . format ( project_id , type_ ) )


signature:
def guid ( self , guid ) :
    """Determines JSONAPI type for provided GUID"""

implementation: return self . _json ( self . _get ( self . _build_url ( 'guids' , guid ) ) , 200 ) [ 'data' ] [ 'type' ]


signature:
def _json ( self , response , status_code ) :
    """Extract JSON from response if ` status _ code ` matches ."""

implementation:
if isinstance ( status_code , numbers . Integral ) :
    status_code = ( status_code , )
if response . status_code in status_code :
    return response . json ( )
else :
    raise RuntimeError ( "Response has status " "code {} not {}" . format ( response . status_code , status_code ) )


signature:
def _follow_next ( self , url ) :
    """Follow the ' next ' link on paginated results ."""

implementation:
response = self . _json ( self . _get ( url ) , 200 )
data = response [ 'data' ]
next_url = self . _get_attribute ( response , 'links' , 'next' )
while next_url is not None :
    response = self . _json ( self . _get ( next_url ) , 200 )
    data . extend ( response [ 'data' ] )
    next_url = self . _get_attribute ( response , 'links' , 'next' )
return data


signature:
def project ( * descs , root_file = None ) :
    """Make a new project , using recursion and alias resolution .
 Use this function in preference to calling Project ( ) directly ."""

implementation:
load . ROOT_FILE = root_file
desc = merge . merge ( merge . DEFAULT_PROJECT , * descs )
path = desc . get ( 'path' , '' )
if root_file :
    project_path = os . path . dirname ( root_file )
    if path :
        path += ':' + project_path
    else :
        path = project_path
with load . extender ( path ) :
    desc = recurse . recurse ( desc )
project = construct . construct ( ** desc )
project . desc = desc
return project


signature:
def clear ( self ) :
    """Clear description to default values"""

implementation:
self . _desc = { }
for key , value in merge . DEFAULT_PROJECT . items ( ) :
    if key not in self . _HIDDEN :
        self . _desc [ key ] = type ( value ) ( )


signature:
def update ( self , desc = None , ** kwds ) :
    """This method updates the description much like dict . update ( ) , * except * :
 1 . for description which have dictionary values , it uses update
 to alter the existing value and does not replace them .
 2 . ` None ` is a special value that means " clear section to default " or
 " delete field " ."""

implementation: sections . update ( self . _desc , desc , ** kwds )


signature:
def SPI ( ledtype = None , num = 0 , ** kwargs ) :
    """Wrapper function for using SPI device drivers on systems like the
 Raspberry Pi and BeagleBone . This allows using any of the SPI drivers
 from a single entry point instead importing the driver for a specific
 LED type .
 Provides the same parameters of
 : py : class : ` bibliopixel . drivers . SPI . SPIBase ` as
 well as those below :
 : param ledtype : One of : LPD8806 , WS2801 , WS281X , or APA102"""

implementation:
from . . . project . types . ledtype import make
if ledtype is None :
    raise ValueError ( 'Must provide ledtype value!' )
ledtype = make ( ledtype )
if num == 0 :
    raise ValueError ( 'Must provide num value >0!' )
if ledtype not in SPI_DRIVERS . keys ( ) :
    raise ValueError ( '{} is not a valid LED type.' . format ( ledtype ) )
return SPI_DRIVERS [ ledtype ] ( num , ** kwargs )


signature:
def put_edit ( self , f , * args , ** kwds ) :
    """Defer an edit to run on the EditQueue .
 : param callable f : The function to be called
 : param tuple args : Positional arguments to the function
 : param tuple kwds : Keyword arguments to the function
 : throws queue . Full : if the queue is full"""

implementation: self . put_nowait ( functools . partial ( f , * args , ** kwds ) )


signature:
def get_and_run_edits ( self ) :
    """Get all the edits in the queue , then execute them .
 The algorithm gets all edits , and then executes all of them . It does
 * not * pull off one edit , execute , repeat until the queue is empty , and
 that means that the queue might not be empty at the end of
 ` ` run _ edits ` ` , because new edits might have entered the queue
 while the previous edits are being executed .
 This has the advantage that if edits enter the queue faster than they
 can be processed , ` ` get _ and _ run _ edits ` ` won ' t go into an infinite loop ,
 but rather the queue will grow unboundedly , which that can be
 detected , and mitigated and reported on - or if Queue . maxsize is
 set , ` ` bp ` ` will report a fairly clear error and just dump the edits
 on the ground ."""

implementation:
if self . empty ( ) :
    return
edits = [ ]
while True :
    try :
        edits . append ( self . get_nowait ( ) )
    except queue . Empty :
        break
for e in edits :
    try :
        e ( )
    except :
        log . error ( 'Error on edit %s' , e )
        traceback . print_exc ( )


signature:
def error ( self , text ) :
    """SHOULD BE PRIVATE"""

implementation:
msg = 'Error with dev: {}, spi_speed: {} - {}' . format ( self . _dev , self . _spi_speed , text )
log . error ( msg )
raise IOError ( msg )


signature:
def send_packet ( self , data ) :
    """SHOULD BE PRIVATE"""

implementation:
package_size = 4032
# bit smaller than 4096 because of headers
for i in range ( int ( math . ceil ( len ( data ) / package_size ) ) ) :
    start = i * package_size
    end = ( i + 1 ) * package_size
    self . _spi . write ( data [ start : end ] )
    self . _spi . flush ( )


signature:
def find_serial_devices ( self ) :
    """Scan and report all compatible serial devices on system .
 : returns : List of discovered devices"""

implementation:
if self . devices is not None :
    return self . devices
self . devices = { }
hardware_id = "(?i)" + self . hardware_id
# forces case insensitive
for ports in serial . tools . list_ports . grep ( hardware_id ) :
    port = ports [ 0 ]
    try :
        id = self . get_device_id ( port )
        ver = self . _get_device_version ( port )
    except :
        log . debug ( 'Error getting device_id for %s, %s' , port , self . baudrate )
        if True :
            raise
        continue
    if getattr ( ports , '__len__' , lambda : 0 ) ( ) :
        log . debug ( 'Multi-port device %s:%s:%s with %s ports found' , self . hardware_id , id , ver , len ( ports ) )
    if id < 0 :
        log . debug ( 'Serial device %s:%s:%s with id %s < 0' , self . hardware_id , id , ver )
    else :
        self . devices [ id ] = port , ver
return self . devices


signature:
def get_device ( self , id = None ) :
    """Returns details of either the first or specified device
 : param int id : Identifier of desired device . If not given , first device
 found will be returned
 : returns tuple : Device ID , Device Address , Firmware Version"""

implementation:
if id is None :
    if not self . devices :
        raise ValueError ( 'No default device for %s' % self . hardware_id )
    id , ( device , version ) = sorted ( self . devices . items ( ) ) [ 0 ]
elif id in self . devices :
    device , version = self . devices [ id ]
else :
    error = 'Unable to find device with ID %s' % id
    log . error ( error )
    raise ValueError ( error )
log . info ( "Using COM Port: %s, Device ID: %s, Device Ver: %s" , device , id , version )
return id , device , version


signature:
def error ( self , fail = True , action = '' ) :
    """SHOULD BE PRIVATE METHOD"""

implementation:
e = 'There was an unknown error communicating with the device.'
if action :
    e = 'While %s: %s' % ( action , e )
log . error ( e )
if fail :
    raise IOError ( e )


signature:
def set_device_id ( self , dev , id ) :
    """Set device ID to new value .
 : param str dev : Serial device address / path
 : param id : Device ID to set"""

implementation:
if id < 0 or id > 255 :
    raise ValueError ( "ID must be an unsigned byte!" )
com , code , ok = io . send_packet ( CMDTYPE . SETID , 1 , dev , self . baudrate , 5 , id )
if not ok :
    raise_error ( code )


signature:
def get_device_id ( self , dev ) :
    """Get device ID at given address / path .
 : param str dev : Serial device address / path
 : param baudrate : Baudrate to use when connecting ( optional )"""

implementation:
com , code , ok = io . send_packet ( CMDTYPE . GETID , 0 , dev , self . baudrate , 5 )
if code is None :
    self . error ( action = 'get_device_id' )
return code


signature:
def get ( name = None ) :
    """Return a named Palette , or None if no such name exists .
 If ` ` name ` ` is omitted , the default value is used ."""

implementation:
if name is None or name == 'default' :
    return _DEFAULT_PALETTE
if isinstance ( name , str ) :
    return PROJECT_PALETTES . get ( name ) or BUILT_IN_PALETTES . get ( name )


signature:
def crop ( image , top_offset = 0 , left_offset = 0 , bottom_offset = 0 , right_offset = 0 ) :
    """Return an image cropped on top , bottom , left or right ."""

implementation:
if bottom_offset or top_offset or left_offset or right_offset :
    width , height = image . size
    box = ( left_offset , top_offset , width - right_offset , height - bottom_offset )
    image = image . crop ( box = box )
return image


signature:
def resize ( image , x , y , stretch = False , top = None , left = None , mode = 'RGB' , resample = None ) :
    """Return an image resized ."""

implementation:
if x <= 0 :
    raise ValueError ( 'x must be greater than zero' )
if y <= 0 :
    raise ValueError ( 'y must be greater than zero' )
from PIL import Image
resample = Image . ANTIALIAS if resample is None else resample
if not isinstance ( resample , numbers . Number ) :
    try :
        resample = getattr ( Image , resample . upper ( ) )
    except :
        raise ValueError ( "(1) Didn't understand resample=%s" % resample )
    if not isinstance ( resample , numbers . Number ) :
        raise ValueError ( "(2) Didn't understand resample=%s" % resample )
size = x , y
if stretch :
    return image . resize ( size , resample = resample )
result = Image . new ( mode , size )
ratios = [ d1 / d2 for d1 , d2 in zip ( size , image . size ) ]
if ratios [ 0 ] < ratios [ 1 ] :
    new_size = ( size [ 0 ] , int ( image . size [ 1 ] * ratios [ 0 ] ) )
else :
    new_size = ( int ( image . size [ 0 ] * ratios [ 1 ] ) , size [ 1 ] )
image = image . resize ( new_size , resample = resample )
if left is None :
    box_x = int ( ( x - new_size [ 0 ] ) / 2 )
elif left :
    box_x = 0
else :
    box_x = x - new_size [ 0 ]
if top is None :
    box_y = int ( ( y - new_size [ 1 ] ) / 2 )
elif top :
    box_y = 0
else :
    box_y = y - new_size [ 1 ]
result . paste ( image , box = ( box_x , box_y ) )
return result


signature:
def extract ( self , msg ) :
    """Yield an ordered dictionary if msg [ ' type ' ] is in keys _ by _ type ."""

implementation:
def normal ( key ) :
    v = msg . get ( key )
    if v is None :
        return v
    normalizer = self . normalizers . get ( key , lambda x : x )
    return normalizer ( v )
def odict ( keys ) :
    return collections . OrderedDict ( ( k , normal ( k ) ) for k in keys )
def match ( m ) :
    return ( msg . get ( k ) in v for k , v in m . items ( ) ) if m else ( )
accept = all ( match ( self . accept ) )
reject = any ( match ( self . reject ) )
if reject or not accept :
    keys = ( )
elif self . keys_by_type is None :
    keys = [ k for k in msg . keys ( ) if k not in self . omit ]
else :
    keys = self . keys_by_type . get ( msg . get ( 'type' ) )
return odict ( keys )


signature:
def get ( self , x , y ) :
    """Return the pixel color at position ( x , y ) , or Colors . black if that
 position is out - of - bounds ."""

implementation:
try :
    pixel = self . coord_map [ y ] [ x ]
    return self . _get_base ( pixel )
except IndexError :
    return colors . COLORS . Black


signature:
def drawCircle ( self , x0 , y0 , r , color = None ) :
    """Draw a circle in an RGB color , with center x0 , y0 and radius r ."""

implementation: md . draw_circle ( self . set , x0 , y0 , r , color )


signature:
def fillCircle ( self , x0 , y0 , r , color = None ) :
    """Draw a filled circle in an RGB color , with center x0 , y0 and radius r ."""

implementation: md . fill_circle ( self . set , x0 , y0 , r , color )


signature:
def drawLine ( self , x0 , y0 , x1 , y1 , color = None , colorFunc = None , aa = False ) :
    """Draw a between x0 , y0 and x1 , y1 in an RGB color .
 : param colorFunc : a function that takes an integer from x0 to x1 and
 returns a color corresponding to that point
 : param aa : if True , use Bresenham ' s algorithm for line drawing ;
 otherwise use Xiaolin Wu ' s algorithm"""

implementation: md . draw_line ( self . set , x0 , y0 , x1 , y1 , color , colorFunc , aa )


signature:
def bresenham_line ( self , x0 , y0 , x1 , y1 , color = None , colorFunc = None ) :
    """Draw line from point x0 , y0 to x1 , y1 using Bresenham ' s algorithm .
 Will draw beyond matrix bounds ."""

implementation: md . bresenham_line ( self . set , x0 , y0 , x1 , y1 , color , colorFunc )


signature:
def drawRect ( self , x , y , w , h , color = None , aa = False ) :
    """Draw rectangle with top - left corner at x , y , width w and height h
 : param aa : if True , use Bresenham ' s algorithm for line drawing ;
 otherwise use Xiaolin Wu ' s algorithm"""

implementation: md . draw_rect ( self . set , x , y , w , h , color , aa )


signature:
def fillRect ( self , x , y , w , h , color = None , aa = False ) :
    """Draw a solid rectangle with top - left corner at ( x , y ) , width w and
 height h .
 : param aa : if True , use Bresenham ' s algorithm for line drawing ;
 otherwise use Xiaolin Wu ' s algorithm"""

implementation: md . fill_rect ( self . set , x , y , w , h , color , aa )


signature:
def fillScreen ( self , color = None ) :
    """Fill the matrix with the given RGB color"""

implementation: md . fill_rect ( self . set , 0 , 0 , self . width , self . height , color )


signature:
def drawRoundRect ( self , x , y , w , h , r , color = None , aa = False ) :
    """Draw a rounded rectangle with top - left corner at ( x , y ) , width w ,
 height h , and corner radius r
 : param aa : if True , use Bresenham ' s algorithm for line drawing ;
 otherwise use Xiaolin Wu ' s algorithm"""

implementation: md . draw_round_rect ( self . set , x , y , w , h , r , color , aa )


signature:
def fillRoundRect ( self , x , y , w , h , r , color = None , aa = False ) :
    """Draw a rounded rectangle with top - left corner at ( x , y ) , width w ,
 height h , and corner radius r
 : param aa : if True , use Bresenham ' s algorithm for line drawing ;
 otherwise use Xiaolin Wu ' s algorithm"""

implementation: md . fill_round_rect ( self . set , x , y , w , h , r , color , aa )


signature:
def drawTriangle ( self , x0 , y0 , x1 , y1 , x2 , y2 , color = None , aa = False ) :
    """Draw triangle with vertices ( x0 , y0 ) , ( x1 , y1 ) and ( x2 , y2)
 : param aa : if True , use Bresenham ' s algorithm for line drawing ;
 Otherwise use Xiaolin Wu ' s algorithm"""

implementation: md . draw_triangle ( self . set , x0 , y0 , x1 , y1 , x2 , y2 , color , aa )


signature:
def fillTriangle ( self , x0 , y0 , x1 , y1 , x2 , y2 , color = None , aa = False ) :
    """Draw filled triangle with points x0 , y0 - x1 , y1 - x2 , y2
 : param aa : if True , use Bresenham ' s algorithm for line drawing ;
 otherwise use Xiaolin Wu ' s algorithm"""

implementation: md . fill_triangle ( self . set , x0 , y0 , x1 , y1 , x2 , y2 , color , aa )


signature:
def drawChar ( self , x , y , c , color , bg , aa = False , font = font . default_font , font_scale = 1 ) :
    """Draw a single character c at at ( x , y ) in an RGB color ."""

implementation: md . draw_char ( self . fonts , self . set , self . width , self . height , x , y , c , color , bg , aa , font , font_scale )


signature:
def drawText ( self , text , x = 0 , y = 0 , color = None , bg = colors . COLORS . Off , aa = False , font = font . default_font , font_scale = 1 ) :
    """Draw a line of text starting at ( x , y ) in an RGB color .
 : param colorFunc : a function that takes an integer from x0 to x1 and
 returns a color corresponding to that point
 : param aa : if True , use Bresenham ' s algorithm for line drawing ;
 otherwise use Xiaolin Wu ' s algorithm"""

implementation: md . draw_text ( self . fonts , self . set , text , self . width , self . height , x , y , color , bg , aa , font , font_scale )


signature:
def set_project ( self , project ) :
    """Set the base project for routing ."""

implementation:
def visit ( x ) : # Try to set _ project , then recurse through any values ( )
    set_project = getattr ( x , 'set_project' , None )
    if set_project :
        set_project ( project )
    values = getattr ( x , 'values' , lambda : ( ) )
    for v in values ( ) :
        visit ( v )
visit ( self . routing )


signature:
def receive ( self , msg ) :
    """Returns a ( receiver , msg ) pair , where receiver is ` None ` if no route for
 the message was found , or otherwise an object with a ` receive ` method
 that can accept that ` msg ` ."""

implementation:
x = self . routing
while not isinstance ( x , ActionList ) :
    if not x or not msg :
        return None , msg
    if not isinstance ( x , dict ) :
        raise ValueError ( 'Unexpected type %s' % type ( x ) )
    _ , value = msg . popitem ( last = False )
    x = x . get ( str ( value ) )
return x , msg


signature:
def set ( self , ring , angle , color ) :
    """Set pixel to RGB color tuple"""

implementation:
pixel = self . angleToPixel ( angle , ring )
self . _set_base ( pixel , color )


signature:
def get ( self , ring , angle ) :
    """Get RGB color tuple of color at index pixel"""

implementation:
pixel = self . angleToPixel ( angle , ring )
return self . _get_base ( pixel )


signature:
def run ( function , * args , use_subprocess = False , daemon = True , ** kwds ) :
    """Create input , output queues , call ` function ` in a subprocess or a thread .
 ` ` function ` ` is called like this : ` ` function ( input , output , * args , * * kwds ) ` `
 : param use _ subprocess : if true , create a new multiprocess ;
 if false , create a new thread
 : param function : the function to call
 : param daemon : is the thread or subprocess run as a daemon or not ?
 : param args : positional arguments to the function
 : param kwds : keyword arguments to the function
 : returns : a tuple with three elements : the subprocess or thread , an input
 queue , and an output queue ."""

implementation:
if use_subprocess :
    Creator , Queue = multiprocessing . Process , multiprocessing . Queue
else :
    Creator , Queue = threading . Thread , queue . Queue
input , output = Queue ( ) , Queue ( )
args = input , output , function , args
sub = Creator ( target = _run_locally , args = args , kwargs = kwds , daemon = daemon )
sub . start ( )
return sub , input , output


signature:
def color_scale ( color , level ) :
    """Scale RGB tuple by level , 0 - 256"""

implementation: return tuple ( [ int ( i * level ) >> 8 for i in list ( color ) ] )


signature:
def load ( self , project_file = '' ) :
    """Load / reload the description from a YML file . Prompt if no file given ."""

implementation:
self . _request_project_file ( project_file )
self . clear ( )
self . desc . update ( data_file . load ( self . _project_file ) )


signature:
def save ( self , project_file = '' ) :
    """Save the description as a YML file . Prompt if no file given ."""

implementation:
self . _request_project_file ( project_file )
data_file . dump ( self . desc . as_dict ( ) , self . project_file )


signature:
def get ( self , position = 0 ) :
    """Return a color interpolated from the Palette .
 In the case where continuous = False , serpentine = False , scale = 1,
 autoscale = False , and offset = 0 , this is exactly the same as plain old [ ]
 indexing , but with a wrap - around .
 The constructor parameters affect this result as documented in the
 constructor .
 Arguments :
 ` ` position ` ` :
 May be any integer or floating point number"""

implementation:
n = len ( self )
if n == 1 :
    return self [ 0 ]
pos = position
if self . length and self . autoscale :
    pos *= len ( self )
    pos /= self . length
pos *= self . scale
pos += self . offset
if not self . continuous :
    if not self . serpentine :
        return self [ int ( pos % n ) ]
    # We want a color sequence of length 2n - 2
    # e . g . for n = 5 : a b c d | e d c b | a b c d . . .
    m = ( 2 * n ) - 2
    pos %= m
    if pos < n :
        return self [ int ( pos ) ]
    else :
        return self [ int ( m - pos ) ]
if self . serpentine :
    pos %= ( 2 * n )
    if pos > n :
        pos = ( 2 * n ) - pos
else :
    pos %= n
# p is a number in [ 0 , n ) : scale it to be in [ 0 , n - 1)
pos *= n - 1
pos /= n
index = int ( pos )
fade = pos - index
if not fade :
    return self [ index ]
r1 , g1 , b1 = self [ index ]
r2 , g2 , b2 = self [ ( index + 1 ) % len ( self ) ]
dr , dg , db = r2 - r1 , g2 - g1 , b2 - b1
return r1 + fade * dr , g1 + fade * dg , b1 + fade * db


signature:
def run ( self , next_task ) :
    """Wait for the event , run the task , trigger the next task ."""

implementation:
self . event . wait ( )
self . task ( )
self . event . clear ( )
next_task . event . set ( )


signature:
def report ( function , * args , ** kwds ) :
    """Run a function , catch , report and discard exceptions"""

implementation:
try :
    function ( * args , ** kwds )
except Exception :
    traceback . print_exc ( )


signature:
def _receive ( self , msg ) :
    """Receive a message from the input source and perhaps raise an Exception ."""

implementation:
msg = self . _convert ( msg )
if msg is None :
    return
str_msg = self . verbose and self . _msg_to_str ( msg )
if self . verbose and log . is_debug ( ) :
    log . debug ( 'Message %s' , str_msg )
if self . pre_routing :
    self . pre_routing . receive ( msg )
receiver , msg = self . routing . receive ( msg )
if receiver :
    receiver . receive ( msg )
    if self . verbose :
        log . info ( 'Routed message %s (%s) to %s' , str_msg [ : 128 ] , msg , repr ( receiver ) )


signature:
def set_device_brightness ( self , val ) :
    """APA102 & SK9822 support on - chip brightness control , allowing greater
 color depth .
 APA102 superimposes a 440Hz PWM on the 19kHz base PWM to control
 brightness . SK9822 uses a base 4.7kHz PWM but controls brightness with a
 variable current source .
 Because of this SK9822 will have much less flicker at lower levels .
 Either way , this option is better and faster than scaling in
 BiblioPixel ."""

implementation:
# bitshift to scale from 8 bit to 5
self . _chipset_brightness = ( val >> 3 )
self . _brightness_list = [ 0xE0 + self . _chipset_brightness ] * self . numLEDs
self . _packet [ self . _start_frame : self . _pixel_stop : 4 ] = ( self . _brightness_list )


signature:
def _addLoggingLevel ( levelName , levelNum , methodName = None ) :
    """Comprehensively adds a new logging level to the ` logging ` module and the
 currently configured logging class .
 ` levelName ` becomes an attribute of the ` logging ` module with the value
 ` levelNum ` . ` methodName ` becomes a convenience method for both ` logging `
 itself and the class returned by ` logging . getLoggerClass ( ) ` ( usually just
 ` logging . Logger ` ) . If ` methodName ` is not specified , ` levelName . lower ( ) ` is
 used .
 To avoid accidental clobberings of existing attributes , this method will
 raise an ` AttributeError ` if the level name is already an attribute of the
 ` logging ` module or if the method name is already present
 Example
 > > > addLoggingLevel ( ' TRACE ' , logging . DEBUG - 5)
 > > > logging . getLogger ( _ _ name _ _ ) . setLevel ( " TRACE " )
 > > > logging . getLogger ( _ _ name _ _ ) . trace ( ' that worked ' )
 > > > logging . trace ( ' so did this ' )
 > > > logging . TRACE"""

implementation:
if not methodName :
    methodName = levelName . lower ( )
if hasattr ( logging , levelName ) :
    raise AttributeError ( '{} already defined in logging module' . format ( levelName ) )
if hasattr ( logging , methodName ) :
    raise AttributeError ( '{} already defined in logging module' . format ( methodName ) )
if hasattr ( logging . getLoggerClass ( ) , methodName ) :
    raise AttributeError ( '{} already defined in logger class' . format ( methodName ) )
# This method was inspired by the answers to Stack Overflow post
# http : / / stackoverflow . com / q / 2183233/2988730 , especially
# http : / / stackoverflow . com / a / 13638084/2988730
def logForLevel ( self , message , * args , ** kwargs ) :
    if self . isEnabledFor ( levelNum ) :
        self . _log ( levelNum , message , args , ** kwargs )
def logToRoot ( message , * args , ** kwargs ) :
    logging . log ( levelNum , message , * args , ** kwargs )
logging . addLevelName ( levelNum , levelName )
setattr ( logging , levelName , levelNum )
setattr ( logging . getLoggerClass ( ) , methodName , logForLevel )
setattr ( logging , methodName , logToRoot )


signature:
def set_log_level ( level ) :
    """: param level : the level to set - either a string level name from
 ' frame ' , ' debug ' , ' info ' , ' warning ' , ' error '
 or an integer log level from :
 log . FRAME , log . DEBUG , log . INFO , log . WARNING , log . ERROR"""

implementation:
if isinstance ( level , str ) :
    level = LOG_NAMES [ level . lower ( ) ]
logger . setLevel ( level )


signature:
def construct ( cls , project , ** desc ) :
    """Construct a layout .
 SHOULD BE PRIVATE"""

implementation: return cls ( project . drivers , maker = project . maker , ** desc )


signature:
def clone ( self ) :
    """Return an independent copy of this layout with a completely separate
 color _ list and no drivers ."""

implementation:
args = { k : getattr ( self , k ) for k in self . CLONE_ATTRS }
args [ 'color_list' ] = copy . copy ( self . color_list )
return self . __class__ ( [ ] , ** args )


signature:
def set_color_list ( self , color_list , offset = 0 ) :
    """Set the internal colors starting at an optional offset .
 If ` color _ list ` is a list or other 1 - dimensional array , it is reshaped
 into an N x 3 list .
 If ` color _ list ` too long it is truncated ; if it is too short then only
 the initial colors are set ."""

implementation:
if not len ( color_list ) :
    return
color_list = make . colors ( color_list )
size = len ( self . _colors ) - offset
if len ( color_list ) > size :
    color_list = color_list [ : size ]
self . _colors [ offset : offset + len ( color_list ) ] = color_list


signature:
def setRGB ( self , pixel , r , g , b ) :
    """Set single pixel using individual RGB values instead of tuple"""

implementation: self . _set_base ( pixel , ( r , g , b ) )


signature:
def setHSV ( self , pixel , hsv ) :
    """Set single pixel to HSV tuple"""

implementation:
color = conversions . hsv2rgb ( hsv )
self . _set_base ( pixel , color )


signature:
def fill ( self , color , start = 0 , end = - 1 ) :
    """Fill the entire strip with RGB color tuple"""

implementation:
start = max ( start , 0 )
if end < 0 or end >= self . numLEDs :
    end = self . numLEDs - 1
for led in range ( start , end + 1 ) : # since 0 - index include end in range
    self . _set_base ( led , color )


signature:
def fillRGB ( self , r , g , b , start = 0 , end = - 1 ) :
    """Fill entire strip by giving individual RGB values instead of tuple"""

implementation: self . fill ( ( r , g , b ) , start , end )


signature:
def fillHSV ( self , hsv , start = 0 , end = - 1 ) :
    """Fill the entire strip with HSV color tuple"""

implementation: self . fill ( conversions . hsv2rgb ( hsv ) , start , end )


signature:
def set_colors ( self , buf ) :
    """DEPRECATED : use self . color _ list
 Use with extreme caution !
 Directly sets the internal buffer and bypasses all brightness and
 rotation control buf must also be in the exact format required by the
 display type ."""

implementation:
deprecated . deprecated ( 'layout.set_colors' )
if len ( self . _colors ) != len ( buf ) :
    raise IOError ( "Data buffer size incorrect! " "Expected: {} bytes / Received: {} bytes" . format ( len ( self . _colors ) , len ( buf ) ) )
self . _colors [ : ] = buf


signature:
def set_device_brightness ( self , brightness ) :
    """Hardware specific method to set the global brightness for
 this driver ' s output . This method is required to be implemented ,
 however , users should call
 : py : meth : ` . driver _ base . DriverBase . set _ brightness `
 instead of calling this method directly .
 : param int brightness : 0-255 value representing the desired
 brightness level"""

implementation:
packet = util . generate_header ( CMDTYPE . BRIGHTNESS , 1 )
packet . append ( self . _brightness )
s = self . _connect ( )
s . sendall ( packet )
resp = ord ( s . recv ( 1 ) )
return resp == RETURN_CODES . SUCCESS


signature:
def wheel_helper ( pos , length , cycle_step ) :
    """Helper for wheel _ color that distributes colors over length and
 allows shifting position ."""

implementation: return wheel_color ( ( pos * len ( _WHEEL ) / length ) + cycle_step )


signature:
def single ( method ) :
    """Decorator for RestServer methods that take a single address"""

implementation:
@ functools . wraps ( method )
def single ( self , address , value = None ) :
    address = urllib . parse . unquote_plus ( address )
    try :
        error = NO_PROJECT_ERROR
        if not self . project :
            raise ValueError
        error = BAD_ADDRESS_ERROR
        ed = editor . Editor ( address , self . project )
        if value is None :
            error = BAD_GETTER_ERROR
            result = method ( self , ed )
        else :
            error = BAD_SETTER_ERROR
            result = method ( self , ed , value )
        result = { 'value' : result }
    except Exception as e :
        traceback . print_exc ( )
        msg = '%s\n%s' % ( error . format ( ** locals ( ) ) , e )
        result = { 'error' : msg }
    return flask . jsonify ( result )
return single


signature:
def multi ( method ) :
    """Decorator for RestServer methods that take multiple addresses"""

implementation:
@ functools . wraps ( method )
def multi ( self , address = '' ) :
    values = flask . request . values
    address = urllib . parse . unquote_plus ( address )
    if address and values and not address . endswith ( '.' ) :
        address += '.'
    result = { }
    for a in values or '' :
        try :
            if not self . project :
                raise ValueError ( 'No Project is currently loaded' )
            ed = editor . Editor ( address + a , self . project )
            result [ address + a ] = { 'value' : method ( self , ed , a ) }
        except :
            if self . project :
                traceback . print_exc ( )
            result [ address + a ] = { 'error' : 'Could not multi addr %s' % a }
    return flask . jsonify ( result )
return multi


signature:
def advance_permutation ( a , increasing = True , forward = True ) :
    """Advance a list of unique , ordered elements in - place , lexicographically
 increasing or backward , by rightmost or leftmost digit .
 Returns False if the permutation wrapped around - i . e . went from
 lexicographically greatest to least , and True in all other cases .
 If the length of the list is N , then this function will repeat values after
 N ! steps , and will return False exactly once .
 See also https : / / stackoverflow . com / a / 34325140/43839"""

implementation:
if not forward :
    a . reverse ( )
cmp = operator . lt if increasing else operator . gt
try :
    i = next ( i for i in reversed ( range ( len ( a ) - 1 ) ) if cmp ( a [ i ] , a [ i + 1 ] ) )
    j = next ( j for j in reversed ( range ( i + 1 , len ( a ) ) ) if cmp ( a [ i ] , a [ j ] ) )
except StopIteration : # This is the lexicographically last permutation .
    if forward :
        a . reverse ( )
    return False
a [ i ] , a [ j ] = a [ j ] , a [ i ]
a [ i + 1 : ] = reversed ( a [ i + 1 : ] )
if not forward :
    a . reverse ( )
return True


signature:
def _on_index ( self , old_index ) :
    """Override this method to get called right after ` ` self . index ` ` is set .
 : param int old _ index : the previous index , before it was changed ."""

implementation:
if self . animation :
    log . debug ( '%s: %s' , self . __class__ . __name__ , self . current_animation . title )
    self . frames = self . animation . generate_frames ( False )


signature:
def animation ( self ) :
    """: returns : the selected animation based on self . index , or None if
 self . index is out of bounds"""

implementation:
if 0 <= self . _index < len ( self . animations ) :
    return self . animations [ self . _index ]


signature:
def apply ( self , function ) :
    """For each row or column in cuts , read a list of its colors ,
 apply the function to that list of colors , then write it back
 to the layout ."""

implementation:
for cut in self . cuts :
    value = self . read ( cut )
    function ( value )
    self . write ( cut , value )


implementation: self . set ( pixel , ( r , g , b ) )


implementation:
color = conversions . hsv2rgb ( hsv )
self . set ( pixel , color )


signature:
def compose_events ( events , condition = all ) :
    """Compose a sequence of events into one event .
 Arguments :
 events : a sequence of objects looking like threading . Event
 condition : a function taking a sequence of bools and returning a bool ."""

implementation:
events = list ( events )
master_event = threading . Event ( )
def changed ( ) :
    if condition ( e . is_set ( ) for e in events ) :
        master_event . set ( )
    else :
        master_event . clear ( )
def add_changed ( f ) :
    @ functools . wraps ( f )
    def wrapped ( ) :
        f ( )
        changed ( )
    return wrapped
for e in events :
    e . set = add_changed ( e . set )
    e . clear = add_changed ( e . clear )
changed ( )
return master_event


signature:
def _add_redundant_arguments ( parser ) :
    """These arguments are redundant with just using a project , and we should
 encouraging that as you don ' t have to learn any dumb flags !
 For example , instead of
 bp foo . yml - - animation = wombat - - numbers = float
 use
 bp foo . yml + ' { animation : wombat , numbers : float } '"""

implementation:
parser . add_argument ( '-a' , '--animation' , default = None , help = 'Default animation type if no animation is specified' )
if deprecated . allowed ( ) : # pragma : no cover
    parser . add_argument ( '--dimensions' , '--dim' , default = None , help = 'DEPRECATED: x, (x, y) or (x, y, z) dimensions for project' )
parser . add_argument ( '--shape' , default = None , help = 'x, (x, y) or (x, y, z) dimensions for project' )
parser . add_argument ( '-l' , '--layout' , default = None , help = 'Default layout class if no layout is specified' )
parser . add_argument ( '--numbers' , '-n' , default = 'python' , choices = NUMBER_TYPES , help = NUMBERS_HELP )
parser . add_argument ( '-p' , '--path' , default = None , help = PATH_HELP )


signature:
def draw_circle ( setter , x0 , y0 , r , color = None ) :
    """Draws a circle at point x0 , y0 with radius r of the specified RGB color"""

implementation:
f = 1 - r
ddF_x = 1
ddF_y = - 2 * r
x = 0
y = r
setter ( x0 , y0 + r , color )
setter ( x0 , y0 - r , color )
setter ( x0 + r , y0 , color )
setter ( x0 - r , y0 , color )
while x < y :
    if f >= 0 :
        y -= 1
        ddF_y += 2
        f += ddF_y
    x += 1
    ddF_x += 2
    f += ddF_x
    setter ( x0 + x , y0 + y , color )
    setter ( x0 - x , y0 + y , color )
    setter ( x0 + x , y0 - y , color )
    setter ( x0 - x , y0 - y , color )
    setter ( x0 + y , y0 + x , color )
    setter ( x0 - y , y0 + x , color )
    setter ( x0 + y , y0 - x , color )
    setter ( x0 - y , y0 - x , color )


signature:
def fill_circle ( setter , x0 , y0 , r , color = None ) :
    """Draws a filled circle at point x0 , y0 with radius r and specified color"""

implementation:
_draw_fast_vline ( setter , x0 , y0 - r , 2 * r + 1 , color )
_fill_circle_helper ( setter , x0 , y0 , r , 3 , 0 , color )


signature:
def bresenham_line ( setter , x0 , y0 , x1 , y1 , color = None , colorFunc = None ) :
    """Draw line from point x0 , y0 to x , 1 , y1 . Will draw beyond matrix bounds ."""

implementation:
steep = abs ( y1 - y0 ) > abs ( x1 - x0 )
if steep :
    x0 , y0 = y0 , x0
    x1 , y1 = y1 , x1
if x0 > x1 :
    x0 , x1 = x1 , x0
    y0 , y1 = y1 , y0
dx = x1 - x0
dy = abs ( y1 - y0 )
err = dx / 2
if y0 < y1 :
    ystep = 1
else :
    ystep = - 1
count = 0
for x in range ( x0 , x1 + 1 ) :
    if colorFunc :
        color = colorFunc ( count )
        count += 1
    if steep :
        setter ( y0 , x , color )
    else :
        setter ( x , y0 , color )
    err -= dy
    if err < 0 :
        y0 += ystep
        err += dx


signature:
def draw_rect ( setter , x , y , w , h , color = None , aa = False ) :
    """Draw rectangle with top - left corner at x , y , width w and height h"""

implementation:
_draw_fast_hline ( setter , x , y , w , color , aa )
_draw_fast_hline ( setter , x , y + h - 1 , w , color , aa )
_draw_fast_vline ( setter , x , y , h , color , aa )
_draw_fast_vline ( setter , x + w - 1 , y , h , color , aa )


signature:
def fill_rect ( setter , x , y , w , h , color = None , aa = False ) :
    """Draw solid rectangle with top - left corner at x , y , width w and height h"""

implementation:
for i in range ( x , x + w ) :
    _draw_fast_vline ( setter , i , y , h , color , aa )


signature:
def draw_round_rect ( setter , x , y , w , h , r , color = None , aa = False ) :
    """Draw rectangle with top - left corner at x , y , width w , height h ,
 and corner radius r ."""

implementation:
_draw_fast_hline ( setter , x + r , y , w - 2 * r , color , aa )
# Top
_draw_fast_hline ( setter , x + r , y + h - 1 , w - 2 * r , color , aa )
# Bottom
_draw_fast_vline ( setter , x , y + r , h - 2 * r , color , aa )
# Left
_draw_fast_vline ( setter , x + w - 1 , y + r , h - 2 * r , color , aa )
# Right
# draw four corners
_draw_circle_helper ( setter , x + r , y + r , r , 1 , color , aa )
_draw_circle_helper ( setter , x + w - r - 1 , y + r , r , 2 , color , aa )
_draw_circle_helper ( setter , x + w - r - 1 , y + h - r - 1 , r , 4 , color , aa )
_draw_circle_helper ( setter , x + r , y + h - r - 1 , r , 8 , color , aa )


signature:
def fill_round_rect ( setter , x , y , w , h , r , color = None , aa = False ) :
    """Draw solid rectangle with top - left corner at x , y , width w , height h ,
 and corner radius r"""

implementation:
fill_rect ( setter , x + r , y , w - 2 * r , h , color , aa )
_fill_circle_helper ( setter , x + w - r - 1 , y + r , r , 1 , h - 2 * r - 1 , color , aa )
_fill_circle_helper ( setter , x + r , y + r , r , 2 , h - 2 * r - 1 , color , aa )


signature:
def draw_triangle ( setter , x0 , y0 , x1 , y1 , x2 , y2 , color = None , aa = False ) :
    """Draw triangle with points x0 , y0 - x1 , y1 - x2 , y2"""

implementation:
draw_line ( setter , x0 , y0 , x1 , y1 , color , aa )
draw_line ( setter , x1 , y1 , x2 , y2 , color , aa )
draw_line ( setter , x2 , y2 , x0 , y0 , color , aa )


signature:
def fill_triangle ( setter , x0 , y0 , x1 , y1 , x2 , y2 , color = None , aa = False ) :
    """Draw solid triangle with points x0 , y0 - x1 , y1 - x2 , y2"""

implementation:
a = b = y = last = 0
if y0 > y1 :
    y0 , y1 = y1 , y0
    x0 , x1 = x1 , x0
if y1 > y2 :
    y2 , y1 = y1 , y2
    x2 , x1 = x1 , x2
if y0 > y1 :
    y0 , y1 = y1 , y0
    x0 , x1 = x1 , x0
if y0 == y2 : # Handle awkward all - on - same - line case as its own thing
    a = b = x0
    if x1 < a :
        a = x1
    elif x1 > b :
        b = x1
    if x2 < a :
        a = x2
    elif x2 > b :
        b = x2
        _draw_fast_hline ( setter , a , y0 , b - a + 1 , color , aa )
dx01 = x1 - x0
dy01 = y1 - y0
dx02 = x2 - x0
dy02 = y2 - y0
dx12 = x2 - x1
dy12 = y2 - y1
sa = 0
sb = 0
# For upper part of triangle , find scanline crossings for segments
# 0-1 and 0-2 . If y1 = y2 ( flat - bottomed triangle ) , the scanline y1
# is included here ( and second loop will be skipped , avoiding a / 0
# error there ) , otherwise scanline y1 is skipped here and handled
# in the second loop . . . which also avoids a / 0 error here if y0 = y1
# ( flat - topped triangle ) .
if y1 == y2 :
    last = y1
    # include y1 scanline
else :
    last = y1 - 1
# skip it
for y in range ( y , last + 1 ) :
    a = x0 + sa / dy01
    b = x0 + sb / dy02
    sa += dx01
    sb += dx02
    if a > b :
        a , b = b , a
        _draw_fast_hline ( setter , a , y , b - a + 1 , color , aa )
# For lower part of triangle , find scanline crossings for segments
# 0-2 and 1-2 . This loop is skipped if y1 = y2.
sa = dx12 * ( y - y1 )
sb = dx02 * ( y - y0 )
for y in range ( y , y2 + 1 ) :
    a = x1 + sa / dy12
    b = x0 + sb / dy02
    sa += dx12
    sb += dx02
    if a > b :
        a , b = b , a
        _draw_fast_hline ( setter , a , y , b - a + 1 , color , aa )


signature:
def set_colors ( self , colors , pos ) :
    """Use with caution !
 Directly set the pixel buffers .
 : param colors : A list of color tuples
 : param int pos : Position in color list to begin set operation ."""

implementation:
self . _colors = colors
self . _pos = pos
end = self . _pos + self . numLEDs
if end > len ( self . _colors ) :
    raise ValueError ( 'Needed %d colors but found %d' % ( end , len ( self . _colors ) ) )


signature:
def update_colors ( self ) :
    """Apply any corrections to the current color list
 and send the results to the driver output . This function primarily
 provided as a wrapper for each driver ' s implementation of
 : py : func : ` _ compute _ packet ` and : py : func : ` _ send _ packet ` ."""

implementation:
start = self . clock . time ( )
with self . brightness_lock : # Swap in a new brightness .
    brightness , self . _waiting_brightness = ( self . _waiting_brightness , None )
if brightness is not None :
    self . _brightness = brightness
    if self . set_device_brightness :
        self . set_device_brightness ( brightness )
self . _compute_packet ( )
self . _send_packet ( )
self . lastUpdate = self . clock . time ( ) - start


signature:
def _render ( self ) :
    """Typically called from : py : func : ` _ compute _ packet ` this applies
 brightness and gamma correction to the pixels controlled by this
 driver ."""

implementation:
if self . set_device_brightness :
    level = 1.0
else :
    level = self . _brightness / 255.0
gam , ( r , g , b ) = self . gamma . get , self . c_order
for i in range ( min ( self . numLEDs , len ( self . _buf ) / 3 ) ) :
    c = [ int ( level * x ) for x in self . _colors [ i + self . _pos ] ]
    self . _buf [ i * 3 : ( i + 1 ) * 3 ] = gam ( c [ r ] ) , gam ( c [ g ] ) , gam ( c [ b ] )


signature:
def pointOnCircle ( cx , cy , radius , angle ) :
    """Calculates the coordinates of a point on a circle given the center point ,
 radius , and angle ."""

implementation:
angle = math . radians ( angle ) - ( math . pi / 2 )
x = cx + radius * math . cos ( angle )
if x < cx :
    x = math . ceil ( x )
else :
    x = math . floor ( x )
y = cy + radius * math . sin ( angle )
if y < cy :
    y = math . ceil ( y )
else :
    y = math . floor ( y )
return ( int ( x ) , int ( y ) )


signature:
def genVector ( width , height , x_mult = 1 , y_mult = 1 ) :
    """Generates a map of vector lengths from the center point to each coordinate .
 width - width of matrix to generate
 height - height of matrix to generate
 x _ mult - value to scale x - axis by
 y _ mult - value to scale y - axis by"""

implementation:
center_x = ( width - 1 ) / 2
center_y = ( height - 1 ) / 2
def length ( x , y ) :
    dx = math . pow ( x - center_x , 2 * x_mult )
    dy = math . pow ( y - center_y , 2 * y_mult )
    return int ( math . sqrt ( dx + dy ) )
return [ [ length ( x , y ) for x in range ( width ) ] for y in range ( height ) ]


signature:
def all_named_colors ( ) :
    """Return an iteration over all name , color pairs in tables"""

implementation:
yield from _TO_COLOR_USER . items ( )
for name , color in _TO_COLOR . items ( ) :
    if name not in _TO_COLOR_USER :
        yield name , color


signature:
def contains ( x ) :
    """Return true if this string or integer tuple appears in tables"""

implementation:
if isinstance ( x , str ) :
    x = canonical_name ( x )
    return x in _TO_COLOR_USER or x in _TO_COLOR
else :
    x = tuple ( x )
    return x in _TO_NAME_USER or x in _TO_NAME


signature:
def make_segments ( strip , length ) :
    """Return a list of Segments that evenly split the strip ."""

implementation:
if len ( strip ) % length :
    raise ValueError ( 'The length of strip must be a multiple of length' )
s = [ ]
try :
    while True :
        s . append ( s [ - 1 ] . next ( length ) if s else Segment ( strip , length ) )
except ValueError :
    return s


signature:
def next ( self , length ) :
    """Return a new segment starting right after self in the same buffer ."""

implementation: return Segment ( self . strip , length , self . offset + self . length )


signature:
def sender ( address , use_queue = True , ** kwds ) :
    """: param str address : a pair ( ip _ address , port ) to pass to socket . connect
 : param bool use _ queue : if True , run the connection in a different thread
 with a queue"""

implementation: return QueuedSender ( address , ** kwds ) if use_queue else Sender ( address )


signature:
def start ( self , threaded = None ) :
    """Creates and starts the project ."""

implementation:
if threaded is not None :
    self . threaded = threaded
run = { 'run' : { 'threaded' : False } }
self . project = project . project ( self . desc , run , root_file = self . project_file )
self . _run = self . project . run
self . _runner . start ( self . threaded )


signature:
def stop ( self = None ) :
    """Stop the builder if it ' s running ."""

implementation:
if not self :
    instance = getattr ( Runner . instance ( ) , 'builder' , None )
    self = instance and instance ( )
    if not self :
        return
self . _runner . stop ( )
if self . project :
    self . project . stop ( )
    self . project = None


signature:
def simpixel ( new = 0 , autoraise = True ) :
    """Open an instance of simpixel in the browser"""

implementation: simpixel_driver . open_browser ( new = new , autoraise = autoraise )


signature:
def recurse ( desc , pre = 'pre_recursion' , post = None , python_path = None ) :
    """Depth first recursion through a dictionary containing type constructors
 The arguments pre , post and children are independently either :
 * None , which means to do nothing
 * a string , which means to use the static class method of that name on the
 class being constructed , or
 * a callable , to be called at each recursion
 Arguments :
 dictionary - - a project dictionary or one of its subdictionaries
 pre - - called before children are visited node in the recursion
 post - - called after children are visited in the recursion
 python _ path - - relative path to start resolving typenames"""

implementation:
def call ( f , desc ) :
    if isinstance ( f , str ) : # f is the name of a static class method on the datatype .
        f = getattr ( datatype , f , None )
    return f and f ( desc )
# Automatically load strings that look like JSON or Yaml filenames .
desc = load . load_if_filename ( desc ) or desc
desc = construct . to_type_constructor ( desc , python_path )
datatype = desc . get ( 'datatype' )
desc = call ( pre , desc ) or desc
for child_name in getattr ( datatype , 'CHILDREN' , [ ] ) :
    child = desc . get ( child_name )
    if child :
        is_plural = child_name . endswith ( 's' )
        remove_s = is_plural and child_name != 'drivers'
        # This is because it ' s the " drivers " directory , whereas
        # the others are animation , control , layout , project
        # without the s . TODO : rename drivers / to driver / in v4
        cname = child_name [ : - 1 ] if remove_s else child_name
        new_path = python_path or ( 'bibliopixel.' + cname )
        if is_plural :
            if isinstance ( child , ( dict , str ) ) :
                child = [ child ]
            for i , c in enumerate ( child ) :
                child [ i ] = recurse ( c , pre , post , new_path )
            desc [ child_name ] = child
        else :
            desc [ child_name ] = recurse ( child , pre , post , new_path )
d = call ( post , desc )
return desc if d is None else d


signature:
def class_name ( c ) :
    """: param c : either an object or a class
 : return : the classname as a string"""

implementation:
if not isinstance ( c , type ) :
    c = type ( c )
return '%s.%s' % ( c . __module__ , c . __name__ )


signature:
def to_type_constructor ( value , python_path = None ) :
    """Tries to convert a value to a type constructor .
 If value is a string , then it used as the " typename " field .
 If the " typename " field exists , the symbol for that name is imported and
 added to the type constructor as a field " datatype " .
 Throws :
 ImportError - - if " typename " is set but cannot be imported
 ValueError - - if " typename " is malformed"""

implementation:
if not value :
    return value
if callable ( value ) :
    return { 'datatype' : value }
value = to_type ( value )
typename = value . get ( 'typename' )
if typename :
    r = aliases . resolve ( typename )
    try :
        value [ 'datatype' ] = importer . import_symbol ( r , python_path = python_path )
        del value [ 'typename' ]
    except Exception as e :
        value [ '_exception' ] = e
return value


signature:
def fill ( strip , item , start = 0 , stop = None , step = 1 ) :
    """Fill a portion of a strip from start to stop by step with a given item .
 If stop is not given , it defaults to the length of the strip ."""

implementation:
if stop is None :
    stop = len ( strip )
for i in range ( start , stop , step ) :
    strip [ i ] = item


signature:
def pop_legacy_palette ( kwds , * color_defaults ) :
    """Older animations in BPA and other areas use all sorts of different names for
 what we are now representing with palettes .
 This function mutates a kwds dictionary to remove these legacy fields and
 extract a palette from it , which it returns ."""

implementation:
palette = kwds . pop ( 'palette' , None )
if palette :
    legacy = [ k for k , _ in color_defaults if k in kwds ]
    if legacy :
        raise ValueError ( 'Cannot set palette and ' + ', ' . join ( legacy ) )
    return palette
values = [ kwds . pop ( k , v ) for k , v in color_defaults ]
if values and color_defaults [ 0 ] [ 0 ] in ( 'colors' , 'palette' ) :
    values = values [ 0 ]
return make . colors ( values or None )


signature:
def euclidean ( c1 , c2 ) :
    """Square of the euclidean distance"""

implementation:
diffs = ( ( i - j ) for i , j in zip ( c1 , c2 ) )
return sum ( x * x for x in diffs )


signature:
def _write ( self , filename , frames , fps , loop = 0 , palette = 256 ) :
    """Write a series of frames as a single animated GIF .
 : param str filename : the name of the GIF file to write
 : param list frames : a list of filenames , each of which represents a single
 frame of the animation . Each frame must have exactly the same
 dimensions , and the code has only been tested with . gif files .
 : param float fps :
 The number of frames per second .
 : param int loop :
 The number of iterations . Default 0 ( meaning loop indefinitely ) .
 : param int palette :
 The number of colors to quantize the image to . Is rounded to
 the nearest power of two . Default 256."""

implementation:
from PIL import Image
images = [ ]
for f in frames :
    data = open ( f , 'rb' ) . read ( )
    images . append ( Image . open ( io . BytesIO ( data ) ) )
# GIF duration is only measured to a hundredth of a second
duration = round ( 1 / fps , 2 )
im = images . pop ( 0 )
im . save ( filename , save_all = True , append_images = images , duration = duration , loop = loop , palette = palette )


signature:
def dumps ( data , use_yaml = None , safe = True , ** kwds ) :
    """Dumps data into a nicely formatted JSON string .
 : param dict data : a dictionary to dump
 : param kwds : keywords to pass to json . dumps
 : returns : a string with formatted data
 : rtype : str"""

implementation:
if use_yaml is None :
    use_yaml = ALWAYS_DUMP_YAML
if use_yaml :
    dumps = yaml . safe_dump if safe else yaml . dump
else :
    dumps = json . dumps
    kwds . update ( indent = 4 , sort_keys = True )
    if not safe :
        kwds . update ( default = repr )
return dumps ( data , ** kwds )


signature:
def dump ( data , file = sys . stdout , use_yaml = None , ** kwds ) :
    """Dumps data as nicely formatted JSON string to a file or file handle
 : param dict data : a dictionary to dump
 : param file : a filename or file handle to write to
 : param kwds : keywords to pass to json . dump"""

implementation:
if use_yaml is None :
    use_yaml = ALWAYS_DUMP_YAML
def dump ( fp ) :
    if use_yaml :
        yaml . safe_dump ( data , stream = fp , ** kwds )
    else :
        json . dump ( data , fp , indent = 4 , sort_keys = True , ** kwds )
if not isinstance ( file , str ) :
    return dump ( file )
if os . path . isabs ( file ) :
    parent = os . path . dirname ( file )
    if not os . path . exists ( parent ) :
        os . makedirs ( parent , exist_ok = True )
with open ( file , 'w' ) as fp :
    return dump ( fp )


signature:
def load ( file , use_yaml = None ) :
    """Loads not only JSON files but also YAML files ending in . yml .
 : param file : a filename or file handle to read from
 : returns : the data loaded from the JSON or YAML file
 : rtype : dict"""

implementation:
if isinstance ( file , str ) :
    fp = open ( file )
    filename = file
else :
    fp = file
    filename = getattr ( fp , 'name' , '' )
try :
    return loads ( fp . read ( ) , use_yaml , filename )
except Exception as e :
    e . args = ( 'There was a error in the data file' , filename ) + e . args
    raise


signature:
def load_if ( s ) :
    """Load either a filename , or a string representation of yml / json ."""

implementation:
is_data_file = s . endswith ( '.json' ) or s . endswith ( '.yml' )
return load ( s ) if is_data_file else loads ( s )


signature:
def adapt_animation_layout ( animation ) :
    """Adapt the setter in an animation ' s layout so that Strip animations can run
 on on Matrix , Cube , or Circle layout , and Matrix or Cube animations can run
 on a Strip layout ."""

implementation:
layout = animation . layout
required = getattr ( animation , 'LAYOUT_CLASS' , None )
if not required or isinstance ( layout , required ) :
    return
msg = LAYOUT_WARNING % ( type ( animation ) . __name__ , required . __name__ , type ( layout ) . __name__ )
setter = layout . set
adaptor = None
if required is strip . Strip :
    if isinstance ( layout , matrix . Matrix ) :
        width = layout . width
        def adaptor ( pixel , color = None ) :
            y , x = divmod ( pixel , width )
            setter ( x , y , color or BLACK )
    elif isinstance ( layout , cube . Cube ) :
        lx , ly = layout . x , layout . y
        def adaptor ( pixel , color = None ) :
            yz , x = divmod ( pixel , lx )
            z , y = divmod ( yz , ly )
            setter ( x , y , z , color or BLACK )
    elif isinstance ( layout , circle . Circle ) :
        def adaptor ( pixel , color = None ) :
            layout . _set_base ( pixel , color or BLACK )
elif required is matrix . Matrix :
    if isinstance ( layout , strip . Strip ) :
        width = animation . width
        def adaptor ( x , y , color = None ) :
            setter ( x + y * width , color or BLACK )
if not adaptor :
    raise ValueError ( msg )
log . warning ( msg )
animation . layout . set = adaptor


signature:
def hsv2rgb_raw ( hsv ) :
    """Converts an HSV tuple to RGB . Intended for internal use .
 You should use hsv2rgb _ spectrum or hsv2rgb _ rainbow instead ."""

implementation:
HSV_SECTION_3 = 0x40
h , s , v = hsv
# The brightness floor is minimum number that all of
# R , G , and B will be set to .
invsat = 255 - s
brightness_floor = ( v * invsat ) // 256
# The color amplitude is the maximum amount of R , G , and B
# that will be added on top of the brightness _ floor to
# create the specific hue desired .
color_amplitude = v - brightness_floor
# figure out which section of the hue wheel we ' re in ,
# and how far offset we are within that section
section = h // HSV_SECTION_3
# 0 . . 2
offset = h % HSV_SECTION_3
# 0 . . 63
rampup = offset
rampdown = ( HSV_SECTION_3 - 1 ) - offset
# compute color - amplitude - scaled - down versions of rampup and rampdown
rampup_amp_adj = ( rampup * color_amplitude ) // ( 256 // 4 )
rampdown_amp_adj = ( rampdown * color_amplitude ) // ( 256 // 4 )
# add brightness _ floor offset to everything
rampup_adj_with_floor = rampup_amp_adj + brightness_floor
rampdown_adj_with_floor = rampdown_amp_adj + brightness_floor
r , g , b = ( 0 , 0 , 0 )
if section :
    if section == 1 : # section 1 : 0x40 . . 0x7F
        r = brightness_floor
        g = rampdown_adj_with_floor
        b = rampup_adj_with_floor
    else : # section 2 ; 0x80 . . 0xBF
        r = rampup_adj_with_floor
        g = brightness_floor
        b = rampdown_adj_with_floor
else : # section 0 : 0x00 . . 0x3F
    r = rampdown_adj_with_floor
    g = rampup_adj_with_floor
    b = brightness_floor
return ( r , g , b )


signature:
def hsv2rgb_spectrum ( hsv ) :
    """Generates RGB values from HSV values in line with a typical light
 spectrum ."""

implementation:
h , s , v = hsv
return hsv2rgb_raw ( ( ( h * 192 ) >> 8 , s , v ) )


signature:
def hsv2rgb_rainbow ( hsv ) :
    """Generates RGB values from HSV that have an even visual
 distribution . Be careful as this method is only have as fast as
 hsv2rgb _ spectrum ."""

implementation:
def nscale8x3_video ( r , g , b , scale ) :
    nonzeroscale = 0
    if scale != 0 :
        nonzeroscale = 1
    if r != 0 :
        r = ( ( r * scale ) >> 8 ) + nonzeroscale
    if g != 0 :
        g = ( ( g * scale ) >> 8 ) + nonzeroscale
    if b != 0 :
        b = ( ( b * scale ) >> 8 ) + nonzeroscale
    return ( r , g , b )
def scale8_video_LEAVING_R1_DIRTY ( i , scale ) :
    nonzeroscale = 0
    if scale != 0 :
        nonzeroscale = 1
    if i != 0 :
        i = ( ( i * scale ) >> 8 ) + nonzeroscale
    return i
h , s , v = hsv
offset = h & 0x1F
# 0 . . 31
offset8 = offset * 8
third = ( offset8 * ( 256 // 3 ) ) >> 8
r , g , b = ( 0 , 0 , 0 )
if not ( h & 0x80 ) :
    if not ( h & 0x40 ) :
        if not ( h & 0x20 ) :
            r = 255 - third
            g = third
            b = 0
        else :
            r = 171
            g = 85 + third
            b = 0x00
    else :
        if not ( h & 0x20 ) :
            twothirds = ( third << 1 )
            r = 171 - twothirds
            g = 171 + third
            b = 0
        else :
            r = 0
            g = 255 - third
            b = third
else :
    if not ( h & 0x40 ) :
        if not ( h & 0x20 ) :
            r = 0x00
            twothirds = ( third << 1 )
            g = 171 - twothirds
            b = 85 + twothirds
        else :
            r = third
            g = 0
            b = 255 - third
    else :
        if not ( h & 0x20 ) :
            r = 85 + third
            g = 0
            b = 171 - third
        else :
            r = 171 + third
            g = 0x00
            b = 85 - third
if s != 255 :
    r , g , b = nscale8x3_video ( r , g , b , s )
    desat = 255 - s
    desat = ( desat * desat ) >> 8
    brightness_floor = desat
    r = r + brightness_floor
    g = g + brightness_floor
    b = b + brightness_floor
if v != 255 :
    v = scale8_video_LEAVING_R1_DIRTY ( v , v )
    r , g , b = nscale8x3_video ( r , g , b , v )
return ( r , g , b )


signature:
def hsv2rgb_360 ( hsv ) :
    """Python default hsv to rgb conversion for when hue values in the
 range 0-359 are preferred . Due to requiring float math , this method
 is slower than hsv2rgb _ rainbow and hsv2rgb _ spectrum ."""


signature:
def color_cmp ( a , b ) :
    """Order colors by hue , saturation and value , in that order .
 Returns - 1 if a < b , 0 if a = = b and 1 if a < b ."""

implementation:
if a == b :
    return 0
a , b = rgb_to_hsv ( a ) , rgb_to_hsv ( b )
return - 1 if a < b else 1


signature:
def get_server ( self , key , ** kwds ) :
    """Get a new or existing server for this key .
 : param int key : key for the server to use"""

implementation:
kwds = dict ( self . kwds , ** kwds )
server = self . servers . get ( key )
if server : # Make sure it ' s the right server .
    server . check_keywords ( self . constructor , kwds )
else : # Make a new server
    server = _CachedServer ( self . constructor , key , kwds )
    self . servers [ key ] = server
return server


signature:
def set_one ( desc , name , value ) :
    """Set one section in a Project description"""

implementation:
old_value = desc . get ( name )
if old_value is None :
    raise KeyError ( 'No section "%s"' % name )
if value is None :
    value = type ( old_value ) ( )
elif name in CLASS_SECTIONS :
    if isinstance ( value , str ) :
        value = { 'typename' : aliases . resolve ( value ) }
    elif isinstance ( value , type ) :
        value = { 'typename' : class_name . class_name ( value ) }
    elif not isinstance ( value , dict ) :
        raise TypeError ( 'Expected dict, str or type, got "%s"' % value )
    typename = value . get ( 'typename' )
    if typename :
        s = 's' if name == 'driver' else ''
        path = 'bibliopixel.' + name + s
        importer . import_symbol ( typename , path )
elif name == 'shape' :
    if not isinstance ( value , ( list , int , tuple , str ) ) :
        raise TypeError ( 'Expected shape, got "%s"' % value )
elif type ( old_value ) is not type ( value ) :
    raise TypeError ( 'Expected %s but got "%s" of type %s' % ( type ( old_value ) , value , type ( value ) ) )
desc [ name ] = value


signature:
def update ( desc , other = None , ** kwds ) :
    """Update sections in a Project description"""

implementation:
other = other and _as_dict ( other ) or { }
for i in other , kwds :
    for k , v in i . items ( ) :
        if isinstance ( v , dict ) : # Only for dicts , merge instead of overwriting
            old_v = desc [ k ]
            for k2 , v2 in v . items ( ) :
                if v2 is None :
                    old_v . pop ( k2 , None )
                else :
                    old_v [ k2 ] = v2
        else :
            set_one ( desc , k , v )


signature:
def name_to_color ( name ) :
    """: param str name : a string identifying a color . It might be a color name
 from the two lists of color names , juce and classic ; or
 it might be a list of numeric r , g , b values separated by
 commas .
 : returns : a color as an RGB 3 - tuple"""

implementation:
def to_color ( name ) :
    name = name . lower ( )
    if ',' in name :
        if name . startswith ( '(' ) and name . endswith ( ')' ) :
            name = name [ 1 : - 1 ]
        if name . startswith ( '[' ) and name . endswith ( ']' ) :
            name = name [ 1 : - 1 ]
        r , g , b = name . split ( ',' )
        return _from_number ( r ) , _from_number ( g ) , _from_number ( b )
    try :
        n = _from_number ( name )
    except :
        color = tables . get_color ( name )
        if color :
            return color
        raise ValueError
    return tables . to_triplet ( n )
try :
    color = to_color ( name )
except :
    raise ValueError ( 'Unknown color name %s' % str ( name ) )
if not all ( 0 <= i <= 255 for i in color ) :
    raise ValueError ( 'Component out of range: %s' % color )
return color


signature:
def color_to_name ( color , use_hex = False ) :
    """: param tuple color : an RGB 3 - tuple of integer colors
 : returns : a string name for this color
 ` ` name _ to _ color ( color _ to _ name ( c ) ) = = c ` ` is guaranteed to be true ( but the
 reverse is not true , because name _ to _ color is a many - to - one function ) ."""

implementation:
if isinstance ( color , list ) :
    color = tuple ( color )
elif not isinstance ( color , tuple ) :
    raise ValueError ( 'Not a color' )
if use_hex :
    return '#%02x%02x%02x' % color
return tables . get_name ( color ) or str ( color )


signature:
def toggle ( s ) :
    """Toggle back and forth between a name and a tuple representation .
 : param str s : a string which is either a text name , or a tuple - string :
 a string with three numbers separated by commas
 : returns : if the string was a text name , return a tuple . If it ' s a
 tuple - string and it corresponds to a text name , return the text
 name , else return the original tuple - string ."""

implementation:
is_numeric = ',' in s or s . startswith ( '0x' ) or s . startswith ( '#' )
c = name_to_color ( s )
return color_to_name ( c ) if is_numeric else str ( c )


signature:
def to_color ( c ) :
    """Try to coerce the argument into a color - a 3 - tuple of numbers -"""

implementation:
if isinstance ( c , numbers . Number ) :
    return c , c , c
if not c :
    raise ValueError ( 'Cannot create color from empty "%s"' % c )
if isinstance ( c , str ) :
    return name_to_color ( c )
if isinstance ( c , list ) :
    c = tuple ( c )
if isinstance ( c , tuple ) :
    if len ( c ) > 3 :
        return c [ : 3 ]
    while len ( c ) < 3 :
        c += ( c [ - 1 ] , )
    return c
raise ValueError ( 'Cannot create color from "%s"' % c )


signature:
def step ( self , amt = 1 ) :
    """This may seem silly , but on a Receiver step ( ) need not do anything .
 Instead , receive the data on the receive thread and set it on the buffer
 then call self . _ hold _ for _ data . set ( )"""

implementation:
if not self . _stop_event . isSet ( ) :
    self . _hold_for_data . wait ( )
    self . _hold_for_data . clear ( )