signature:
def deemph ( self ) :
    '''Apply Compact Disc ( IEC 60908 ) de - emphasis ( a treble attenuation
 shelving filter ) . Pre - emphasis was applied in the mastering of some
 CDs issued in the early 1980s . These included many classical music
 albums , as well as now sought - after issues of albums by The Beatles ,
 Pink Floyd and others . Pre - emphasis should be removed at playback time
 by a de - emphasis filter in the playback device . However , not all modern
 CD players have this filter , and very few PC CD drives have it ; playing
 pre - emphasised audio without the correct de - emphasis filter results in
 audio that sounds harsh and is far from what its creators intended .
 The de - emphasis filter is implemented as a biquad and requires the
 input audio sample rate to be either 44.1kHz or 48kHz . Maximum
 deviation from the ideal response is only 0.06dB ( up to 20kHz ) .
 See Also
 bass , treble'''

implementation:
effect_args = [ 'deemph' ]
self . effects . extend ( effect_args )
self . effects_log . append ( 'deemph' )
return self


signature:
def delay ( self , positions ) :
    '''Delay one or more audio channels such that they start at the given
 positions .
 Parameters
 positions : list of floats
 List of times ( in seconds ) to delay each audio channel .
 If fewer positions are given than the number of channels , the
 remaining channels will be unaffected .'''

implementation:
if not isinstance ( positions , list ) :
    raise ValueError ( "positions must be a a list of numbers" )
if not all ( ( is_number ( p ) and p >= 0 ) for p in positions ) :
    raise ValueError ( "positions must be positive nubmers" )
effect_args = [ 'delay' ]
effect_args . extend ( [ '{:f}' . format ( p ) for p in positions ] )
self . effects . extend ( effect_args )
self . effects_log . append ( 'delay' )
return self


signature:
def downsample ( self , factor = 2 ) :
    '''Downsample the signal by an integer factor . Only the first out of
 each factor samples is retained , the others are discarded .
 No decimation filter is applied . If the input is not a properly
 bandlimited baseband signal , aliasing will occur . This may be desirable
 e . g . , for frequency translation .
 For a general resampling effect with anti - aliasing , see rate .
 Parameters
 factor : int , default = 2
 Downsampling factor .
 See Also
 rate , upsample'''

implementation:
if not isinstance ( factor , int ) or factor < 1 :
    raise ValueError ( 'factor must be a positive integer.' )
effect_args = [ 'downsample' , '{}' . format ( factor ) ]
self . effects . extend ( effect_args )
self . effects_log . append ( 'downsample' )
return self


signature:
def earwax ( self ) :
    '''Makes audio easier to listen to on headphones . Adds ‘ cues ’ to 44.1kHz
 stereo audio so that when listened to on headphones the stereo image is
 moved from inside your head ( standard for headphones ) to outside and in
 front of the listener ( standard for speakers ) .
 Warning : Will only work properly on 44.1kHz stereo audio !'''

implementation:
effect_args = [ 'earwax' ]
self . effects . extend ( effect_args )
self . effects_log . append ( 'earwax' )
return self


signature:
def echo ( self , gain_in = 0.8 , gain_out = 0.9 , n_echos = 1 , delays = [ 60 ] , decays = [ 0.4 ] ) :
    '''Add echoing to the audio .
 Echoes are reflected sound and can occur naturally amongst mountains
 ( and sometimes large buildings ) when talking or shouting ; digital echo
 effects emulate this behav - iour and are often used to help fill out
 the sound of a single instrument or vocal . The time differ - ence
 between the original signal and the reflection is the ' delay ' ( time ) ,
 and the loudness of the reflected signal is the ' decay ' . Multiple
 echoes can have different delays and decays .
 Parameters
 gain _ in : float , default = 0.8
 Input volume , between 0 and 1
 gain _ out : float , default = 0.9
 Output volume , between 0 and 1
 n _ echos : int , default = 1
 Number of reflections
 delays : list , default = [ 60]
 List of delays in miliseconds
 decays : list , default = [ 0.4]
 List of decays , relative to gain in between 0 and 1
 See Also
 echos , reverb , chorus'''

implementation:
if not is_number ( gain_in ) or gain_in <= 0 or gain_in > 1 :
    raise ValueError ( "gain_in must be a number between 0 and 1." )
if not is_number ( gain_out ) or gain_out <= 0 or gain_out > 1 :
    raise ValueError ( "gain_out must be a number between 0 and 1." )
if not isinstance ( n_echos , int ) or n_echos <= 0 :
    raise ValueError ( "n_echos must be a positive integer." )
# validate delays
if not isinstance ( delays , list ) :
    raise ValueError ( "delays must be a list" )
if len ( delays ) != n_echos :
    raise ValueError ( "the length of delays must equal n_echos" )
if any ( ( not is_number ( p ) or p <= 0 ) for p in delays ) :
    raise ValueError ( "the elements of delays must be numbers > 0" )
# validate decays
if not isinstance ( decays , list ) :
    raise ValueError ( "decays must be a list" )
if len ( decays ) != n_echos :
    raise ValueError ( "the length of decays must equal n_echos" )
if any ( ( not is_number ( p ) or p <= 0 or p > 1 ) for p in decays ) :
    raise ValueError ( "the elements of decays must be between 0 and 1" )
effect_args = [ 'echo' , '{:f}' . format ( gain_in ) , '{:f}' . format ( gain_out ) ]
for i in range ( n_echos ) :
    effect_args . extend ( [ '{}' . format ( delays [ i ] ) , '{}' . format ( decays [ i ] ) ] )
self . effects . extend ( effect_args )
self . effects_log . append ( 'echo' )
return self


signature:
def equalizer ( self , frequency , width_q , gain_db ) :
    '''Apply a two - pole peaking equalisation ( EQ ) filter to boost or
 reduce around a given frequency .
 This effect can be applied multiple times to produce complex EQ curves .
 Parameters
 frequency : float
 The filter ' s central frequency in Hz .
 width _ q : float
 The filter ' s width as a Q - factor .
 gain _ db : float
 The filter ' s gain in dB .
 See Also
 bass , treble'''

implementation:
if not is_number ( frequency ) or frequency <= 0 :
    raise ValueError ( "frequency must be a positive number." )
if not is_number ( width_q ) or width_q <= 0 :
    raise ValueError ( "width_q must be a positive number." )
if not is_number ( gain_db ) :
    raise ValueError ( "gain_db must be a number." )
effect_args = [ 'equalizer' , '{:f}' . format ( frequency ) , '{:f}q' . format ( width_q ) , '{:f}' . format ( gain_db ) ]
self . effects . extend ( effect_args )
self . effects_log . append ( 'equalizer' )
return self


signature:
def fade ( self , fade_in_len = 0.0 , fade_out_len = 0.0 , fade_shape = 'q' ) :
    '''Add a fade in and / or fade out to an audio file .
 Default fade shape is 1/4 sine wave .
 Parameters
 fade _ in _ len : float , default = 0.0
 Length of fade - in ( seconds ) . If fade _ in _ len = 0,
 no fade in is applied .
 fade _ out _ len : float , defaut = 0.0
 Length of fade - out ( seconds ) . If fade _ out _ len = 0,
 no fade in is applied .
 fade _ shape : str , default = ' q '
 Shape of fade . Must be one of
 * ' q ' for quarter sine ( default ) ,
 * ' h ' for half sine ,
 * ' t ' for linear ,
 * ' l ' for logarithmic
 * ' p ' for inverted parabola .
 See Also
 splice'''

implementation:
fade_shapes = [ 'q' , 'h' , 't' , 'l' , 'p' ]
if fade_shape not in fade_shapes :
    raise ValueError ( "Fade shape must be one of {}" . format ( " " . join ( fade_shapes ) ) )
if not is_number ( fade_in_len ) or fade_in_len < 0 :
    raise ValueError ( "fade_in_len must be a nonnegative number." )
if not is_number ( fade_out_len ) or fade_out_len < 0 :
    raise ValueError ( "fade_out_len must be a nonnegative number." )
effect_args = [ ]
if fade_in_len > 0 :
    effect_args . extend ( [ 'fade' , '{}' . format ( fade_shape ) , '{:f}' . format ( fade_in_len ) ] )
if fade_out_len > 0 :
    effect_args . extend ( [ 'reverse' , 'fade' , '{}' . format ( fade_shape ) , '{:f}' . format ( fade_out_len ) , 'reverse' ] )
if len ( effect_args ) > 0 :
    self . effects . extend ( effect_args )
    self . effects_log . append ( 'fade' )
return self


signature:
def fir ( self , coefficients ) :
    '''Use SoX ’ s FFT convolution engine with given FIR filter coefficients .
 Parameters
 coefficients : list
 fir filter coefficients'''

implementation:
if not isinstance ( coefficients , list ) :
    raise ValueError ( "coefficients must be a list." )
if not all ( [ is_number ( c ) for c in coefficients ] ) :
    raise ValueError ( "coefficients must be numbers." )
effect_args = [ 'fir' ]
effect_args . extend ( [ '{:f}' . format ( c ) for c in coefficients ] )
self . effects . extend ( effect_args )
self . effects_log . append ( 'fir' )
return self


signature:
def flanger ( self , delay = 0 , depth = 2 , regen = 0 , width = 71 , speed = 0.5 , shape = 'sine' , phase = 25 , interp = 'linear' ) :
    '''Apply a flanging effect to the audio .
 Parameters
 delay : float , default = 0
 Base delay ( in miliseconds ) between 0 and 30.
 depth : float , default = 2
 Added swept delay ( in miliseconds ) between 0 and 10.
 regen : float , default = 0
 Percentage regeneration between - 95 and 95.
 width : float , default = 71,
 Percentage of delayed signal mixed with original between 0 and 100.
 speed : float , default = 0.5
 Sweeps per second ( in Hz ) between 0.1 and 10.
 shape : ' sine ' or ' triangle ' , default = ' sine '
 Swept wave shape
 phase : float , default = 25
 Swept wave percentage phase - shift for multi - channel flange between
 0 and 100 . 0 = 100 = same phase on each channel
 interp : ' linear ' or ' quadratic ' , default = ' linear '
 Digital delay - line interpolation type .
 See Also
 tremolo'''

implementation:
if not is_number ( delay ) or delay < 0 or delay > 30 :
    raise ValueError ( "delay must be a number between 0 and 30." )
if not is_number ( depth ) or depth < 0 or depth > 10 :
    raise ValueError ( "depth must be a number between 0 and 10." )
if not is_number ( regen ) or regen < - 95 or regen > 95 :
    raise ValueError ( "regen must be a number between -95 and 95." )
if not is_number ( width ) or width < 0 or width > 100 :
    raise ValueError ( "width must be a number between 0 and 100." )
if not is_number ( speed ) or speed < 0.1 or speed > 10 :
    raise ValueError ( "speed must be a number between 0.1 and 10." )
if shape not in [ 'sine' , 'triangle' ] :
    raise ValueError ( "shape must be one of 'sine' or 'triangle'." )
if not is_number ( phase ) or phase < 0 or phase > 100 :
    raise ValueError ( "phase must be a number between 0 and 100." )
if interp not in [ 'linear' , 'quadratic' ] :
    raise ValueError ( "interp must be one of 'linear' or 'quadratic'." )
effect_args = [ 'flanger' , '{:f}' . format ( delay ) , '{:f}' . format ( depth ) , '{:f}' . format ( regen ) , '{:f}' . format ( width ) , '{:f}' . format ( speed ) , '{}' . format ( shape ) , '{:f}' . format ( phase ) , '{}' . format ( interp ) ]
self . effects . extend ( effect_args )
self . effects_log . append ( 'flanger' )
return self


signature:
def gain ( self , gain_db = 0.0 , normalize = True , limiter = False , balance = None ) :
    '''Apply amplification or attenuation to the audio signal .
 Parameters
 gain _ db : float , default = 0.0
 Gain adjustment in decibels ( dB ) .
 normalize : bool , default = True
 If True , audio is normalized to gain _ db relative to full scale .
 If False , simply adjusts the audio power level by gain _ db .
 limiter : bool , default = False
 If True , a simple limiter is invoked to prevent clipping .
 balance : str or None , default = None
 Balance gain across channels . Can be one of :
 * None applies no balancing ( default )
 * ' e ' applies gain to all channels other than that with the
 highest peak level , such that all channels attain the same
 peak level
 * ' B ' applies gain to all channels other than that with the
 highest RMS level , such that all channels attain the same
 RMS level
 * ' b ' applies gain with clipping protection to all channels other
 than that with the highest RMS level , such that all channels
 attain the same RMS level
 If normalize = True , ' B ' and ' b ' are equivalent .
 See Also
 loudness'''

implementation:
if not is_number ( gain_db ) :
    raise ValueError ( "gain_db must be a number." )
if not isinstance ( normalize , bool ) :
    raise ValueError ( "normalize must be a boolean." )
if not isinstance ( limiter , bool ) :
    raise ValueError ( "limiter must be a boolean." )
if balance not in [ None , 'e' , 'B' , 'b' ] :
    raise ValueError ( "balance must be one of None, 'e', 'B', or 'b'." )
effect_args = [ 'gain' ]
if balance is not None :
    effect_args . append ( '-{}' . format ( balance ) )
if normalize :
    effect_args . append ( '-n' )
if limiter :
    effect_args . append ( '-l' )
effect_args . append ( '{:f}' . format ( gain_db ) )
self . effects . extend ( effect_args )
self . effects_log . append ( 'gain' )
return self


signature:
def highpass ( self , frequency , width_q = 0.707 , n_poles = 2 ) :
    '''Apply a high - pass filter with 3dB point frequency . The filter can be
 either single - pole or double - pole . The filters roll off at 6dB per pole
 per octave ( 20dB per pole per decade ) .
 Parameters
 frequency : float
 The filter ' s cutoff frequency in Hz .
 width _ q : float , default = 0.707
 The filter ' s width as a Q - factor . Applies only when n _ poles = 2.
 The default gives a Butterworth response .
 n _ poles : int , default = 2
 The number of poles in the filter . Must be either 1 or 2
 See Also
 lowpass , equalizer , sinc , allpass'''

implementation:
if not is_number ( frequency ) or frequency <= 0 :
    raise ValueError ( "frequency must be a positive number." )
if not is_number ( width_q ) or width_q <= 0 :
    raise ValueError ( "width_q must be a positive number." )
if n_poles not in [ 1 , 2 ] :
    raise ValueError ( "n_poles must be 1 or 2." )
effect_args = [ 'highpass' , '-{}' . format ( n_poles ) , '{:f}' . format ( frequency ) ]
if n_poles == 2 :
    effect_args . append ( '{:f}q' . format ( width_q ) )
self . effects . extend ( effect_args )
self . effects_log . append ( 'highpass' )
return self


signature:
def hilbert ( self , num_taps = None ) :
    '''Apply an odd - tap Hilbert transform filter , phase - shifting the signal
 by 90 degrees . This is used in many matrix coding schemes and for
 analytic signal generation . The process is often written as a
 multiplication by i ( or j ) , the imaginary unit . An odd - tap Hilbert
 transform filter has a bandpass characteristic , attenuating the lowest
 and highest frequencies .
 Parameters
 num _ taps : int or None , default = None
 Number of filter taps - must be odd . If none , it is chosen to have
 a cutoff frequency of about 75 Hz .'''

implementation:
if num_taps is not None and not isinstance ( num_taps , int ) :
    raise ValueError ( "num taps must be None or an odd integer." )
if num_taps is not None and num_taps % 2 == 0 :
    raise ValueError ( "num_taps must an odd integer." )
effect_args = [ 'hilbert' ]
if num_taps is not None :
    effect_args . extend ( [ '-n' , '{}' . format ( num_taps ) ] )
self . effects . extend ( effect_args )
self . effects_log . append ( 'hilbert' )
return self


signature:
def loudness ( self , gain_db = - 10.0 , reference_level = 65.0 ) :
    '''Loudness control . Similar to the gain effect , but provides
 equalisation for the human auditory system .
 The gain is adjusted by gain _ db and the signal is equalised according
 to ISO 226 w . r . t . reference _ level .
 Parameters
 gain _ db : float , default = - 10.0
 Loudness adjustment amount ( in dB )
 reference _ level : float , default = 65.0
 Reference level ( in dB ) according to which the signal is equalized .
 Must be between 50 and 75 ( dB )
 See Also
 gain'''

implementation:
if not is_number ( gain_db ) :
    raise ValueError ( 'gain_db must be a number.' )
if not is_number ( reference_level ) :
    raise ValueError ( 'reference_level must be a number' )
if reference_level > 75 or reference_level < 50 :
    raise ValueError ( 'reference_level must be between 50 and 75' )
effect_args = [ 'loudness' , '{:f}' . format ( gain_db ) , '{:f}' . format ( reference_level ) ]
self . effects . extend ( effect_args )
self . effects_log . append ( 'loudness' )
return self


signature:
def mcompand ( self , n_bands = 2 , crossover_frequencies = [ 1600 ] , attack_time = [ 0.005 , 0.000625 ] , decay_time = [ 0.1 , 0.0125 ] , soft_knee_db = [ 6.0 , None ] , tf_points = [ [ ( - 47 , - 40 ) , ( - 34 , - 34 ) , ( - 17 , - 33 ) , ( 0 , 0 ) ] , [ ( - 47 , - 40 ) , ( - 34 , - 34 ) , ( - 15 , - 33 ) , ( 0 , 0 ) ] ] , gain = [ None , None ] ) :
    '''The multi - band compander is similar to the single - band compander but
 the audio is first divided into bands using Linkwitz - Riley cross - over
 filters and a separately specifiable compander run on each band .
 When used with n _ bands = 1 , this effect is identical to compand .
 When using n _ bands > 1 , the first set of arguments applies a single
 band compander , and each subsequent set of arugments is applied on
 each of the crossover frequencies .
 Parameters
 n _ bands : int , default = 2
 The number of bands .
 crossover _ frequencies : list of float , default = [ 1600]
 A list of crossover frequencies in Hz of length n _ bands - 1.
 The first band is always the full spectrum , followed by the bands
 specified by crossover _ frequencies .
 attack _ time : list of float , default = [ 0.005 , 0.000625]
 A list of length n _ bands , where each element is the time in seconds
 over which the instantaneous level of the input signal is averaged
 to determine increases in volume over the current band .
 decay _ time : list of float , default = [ 0.1 , 0.0125]
 A list of length n _ bands , where each element is the time in seconds
 over which the instantaneous level of the input signal is averaged
 to determine decreases in volume over the current band .
 soft _ knee _ db : list of float or None , default = [ 6.0 , None ]
 A list of length n _ bands , where each element is the ammount ( in dB )
 for which the points at where adjacent line segments on the
 transfer function meet will be rounded over the current band .
 If None , no soft _ knee is applied .
 tf _ points : list of list of tuples , default = [
 [ ( - 47 , - 40 ) , ( - 34 , - 34 ) , ( - 17 , - 33 ) , ( 0 , 0 ) ] ,
 [ ( - 47 , - 40 ) , ( - 34 , - 34 ) , ( - 15 , - 33 ) , ( 0 , 0 ) ] ]
 A list of length n _ bands , where each element is the transfer
 function points as a list of tuples corresponding to points in
 ( dB , dB ) defining the compander ' s transfer function over the
 current band .
 gain : list of floats or None
 A list of gain values for each frequency band .
 If None , no gain is applied .
 See Also
 compand , contrast'''

implementation:
if not isinstance ( n_bands , int ) or n_bands < 1 :
    raise ValueError ( "n_bands must be a positive integer." )
if ( not isinstance ( crossover_frequencies , list ) or len ( crossover_frequencies ) != n_bands - 1 ) :
    raise ValueError ( "crossover_frequences must be a list of length n_bands - 1" )
if any ( [ not is_number ( f ) or f < 0 for f in crossover_frequencies ] ) :
    raise ValueError ( "crossover_frequencies elements must be positive floats." )
if not isinstance ( attack_time , list ) or len ( attack_time ) != n_bands :
    raise ValueError ( "attack_time must be a list of length n_bands" )
if any ( [ not is_number ( a ) or a <= 0 for a in attack_time ] ) :
    raise ValueError ( "attack_time elements must be positive numbers." )
if not isinstance ( decay_time , list ) or len ( decay_time ) != n_bands :
    raise ValueError ( "decay_time must be a list of length n_bands" )
if any ( [ not is_number ( d ) or d <= 0 for d in decay_time ] ) :
    raise ValueError ( "decay_time elements must be positive numbers." )
if any ( [ a > d for a , d in zip ( attack_time , decay_time ) ] ) :
    logger . warning ( "Elements of attack_time are larger than decay_time.\n" "For most situations, attack_time should be shorter than " "decay time because the human ear is more sensitive to sudden " "loud music than sudden soft music." )
if not isinstance ( soft_knee_db , list ) or len ( soft_knee_db ) != n_bands :
    raise ValueError ( "soft_knee_db must be a list of length n_bands." )
if any ( [ ( not is_number ( d ) and d is not None ) for d in soft_knee_db ] ) :
    raise ValueError ( "elements of soft_knee_db must be a number or None." )
if not isinstance ( tf_points , list ) or len ( tf_points ) != n_bands :
    raise ValueError ( "tf_points must be a list of length n_bands." )
if any ( [ not isinstance ( t , list ) or len ( t ) == 0 for t in tf_points ] ) :
    raise ValueError ( "tf_points must be a list with at least one point." )
for tfp in tf_points :
    if any ( not isinstance ( pair , tuple ) for pair in tfp ) :
        raise ValueError ( "elements of tf_points lists must be pairs" )
    if any ( len ( pair ) != 2 for pair in tfp ) :
        raise ValueError ( "Tuples in tf_points lists must be length 2" )
    if any ( not ( is_number ( p [ 0 ] ) and is_number ( p [ 1 ] ) ) for p in tfp ) :
        raise ValueError ( "Tuples in tf_points lists must be pairs of numbers." )
    if any ( ( p [ 0 ] > 0 or p [ 1 ] > 0 ) for p in tfp ) :
        raise ValueError ( "Tuple values in tf_points lists must be <= 0 (dB)." )
    if len ( tf_points ) > len ( set ( [ p [ 0 ] for p in tfp ] ) ) :
        raise ValueError ( "Found duplicate x-value in tf_points list." )
if not isinstance ( gain , list ) or len ( gain ) != n_bands :
    raise ValueError ( "gain must be a list of length n_bands" )
if any ( [ not ( is_number ( g ) or g is None ) for g in gain ] ) :
    print ( gain )
    raise ValueError ( "gain elements must be numbers or None." )
effect_args = [ 'mcompand' ]
for i in range ( n_bands ) :
    if i > 0 :
        effect_args . append ( '{:f}' . format ( crossover_frequencies [ i - 1 ] ) )
    intermed_args = [ "{:f},{:f}" . format ( attack_time [ i ] , decay_time [ i ] ) ]
    tf_points_band = tf_points [ i ]
    tf_points_band = sorted ( tf_points_band , key = lambda tf_points_band : tf_points_band [ 0 ] )
    transfer_list = [ ]
    for point in tf_points_band :
        transfer_list . extend ( [ "{:f}" . format ( point [ 0 ] ) , "{:f}" . format ( point [ 1 ] ) ] )
    if soft_knee_db [ i ] is not None :
        intermed_args . append ( "{:f}:{}" . format ( soft_knee_db [ i ] , "," . join ( transfer_list ) ) )
    else :
        intermed_args . append ( "," . join ( transfer_list ) )
    if gain [ i ] is not None :
        intermed_args . append ( "{:f}" . format ( gain [ i ] ) )
    effect_args . append ( ' ' . join ( intermed_args ) )
self . effects . extend ( effect_args )
self . effects_log . append ( 'mcompand' )
return self


signature:
def noiseprof ( self , input_filepath , profile_path ) :
    '''Calculate a profile of the audio for use in noise reduction .
 Running this command does not effect the Transformer effects
 chain . When this function is called , the calculated noise profile
 file is saved to the ` profile _ path ` .
 Parameters
 input _ filepath : str
 Path to audiofile from which to compute a noise profile .
 profile _ path : str
 Path to save the noise profile file .
 See Also
 noisered'''

implementation:
if os . path . isdir ( profile_path ) :
    raise ValueError ( "profile_path {} is a directory." . format ( profile_path ) )
if os . path . dirname ( profile_path ) == '' and profile_path != '' :
    _abs_profile_path = os . path . join ( os . getcwd ( ) , profile_path )
else :
    _abs_profile_path = profile_path
if not os . access ( os . path . dirname ( _abs_profile_path ) , os . W_OK ) :
    raise IOError ( "profile_path {} is not writeable." . format ( _abs_profile_path ) )
effect_args = [ 'noiseprof' , profile_path ]
self . build ( input_filepath , None , extra_args = effect_args )
return None


signature:
def noisered ( self , profile_path , amount = 0.5 ) :
    '''Reduce noise in the audio signal by profiling and filtering .
 This effect is moderately effective at removing consistent
 background noise such as hiss or hum .
 Parameters
 profile _ path : str
 Path to a noise profile file .
 This file can be generated using the ` noiseprof ` effect .
 amount : float , default = 0.5
 How much noise should be removed is specified by amount . Should
 be between 0 and 1 . Higher numbers will remove more noise but
 present a greater likelihood of removing wanted components of
 the audio signal .
 See Also
 noiseprof'''

implementation:
if not os . path . exists ( profile_path ) :
    raise IOError ( "profile_path {} does not exist." . format ( profile_path ) )
if not is_number ( amount ) or amount < 0 or amount > 1 :
    raise ValueError ( "amount must be a number between 0 and 1." )
effect_args = [ 'noisered' , profile_path , '{:f}' . format ( amount ) ]
self . effects . extend ( effect_args )
self . effects_log . append ( 'noisered' )
return self


signature:
def norm ( self , db_level = - 3.0 ) :
    '''Normalize an audio file to a particular db level .
 This behaves identically to the gain effect with normalize = True .
 Parameters
 db _ level : float , default = - 3.0
 Output volume ( db )
 See Also
 gain , loudness'''

implementation:
if not is_number ( db_level ) :
    raise ValueError ( 'db_level must be a number.' )
effect_args = [ 'norm' , '{:f}' . format ( db_level ) ]
self . effects . extend ( effect_args )
self . effects_log . append ( 'norm' )
return self


signature:
def oops ( self ) :
    '''Out Of Phase Stereo effect . Mixes stereo to twin - mono where each
 mono channel contains the difference between the left and right stereo
 channels . This is sometimes known as the ' karaoke ' effect as it often
 has the effect of removing most or all of the vocals from a recording .'''

implementation:
effect_args = [ 'oops' ]
self . effects . extend ( effect_args )
self . effects_log . append ( 'oops' )
return self


signature:
def overdrive ( self , gain_db = 20.0 , colour = 20.0 ) :
    '''Apply non - linear distortion .
 Parameters
 gain _ db : float , default = 20
 Controls the amount of distortion ( dB ) .
 colour : float , default = 20
 Controls the amount of even harmonic content in the output ( dB ) .'''

implementation:
if not is_number ( gain_db ) :
    raise ValueError ( 'db_level must be a number.' )
if not is_number ( colour ) :
    raise ValueError ( 'colour must be a number.' )
effect_args = [ 'overdrive' , '{:f}' . format ( gain_db ) , '{:f}' . format ( colour ) ]
self . effects . extend ( effect_args )
self . effects_log . append ( 'overdrive' )
return self


signature:
def pad ( self , start_duration = 0.0 , end_duration = 0.0 ) :
    '''Add silence to the beginning or end of a file .
 Calling this with the default arguments has no effect .
 Parameters
 start _ duration : float
 Number of seconds of silence to add to beginning .
 end _ duration : float
 Number of seconds of silence to add to end .
 See Also
 delay'''

implementation:
if not is_number ( start_duration ) or start_duration < 0 :
    raise ValueError ( "Start duration must be a positive number." )
if not is_number ( end_duration ) or end_duration < 0 :
    raise ValueError ( "End duration must be positive." )
effect_args = [ 'pad' , '{:f}' . format ( start_duration ) , '{:f}' . format ( end_duration ) ]
self . effects . extend ( effect_args )
self . effects_log . append ( 'pad' )
return self


signature:
def phaser ( self , gain_in = 0.8 , gain_out = 0.74 , delay = 3 , decay = 0.4 , speed = 0.5 , modulation_shape = 'sinusoidal' ) :
    '''Apply a phasing effect to the audio .
 Parameters
 gain _ in : float , default = 0.8
 Input volume between 0 and 1
 gain _ out : float , default = 0.74
 Output volume between 0 and 1
 delay : float , default = 3
 Delay in miliseconds between 0 and 5
 decay : float , default = 0.4
 Decay relative to gain _ in , between 0.1 and 0.5.
 speed : float , default = 0.5
 Modulation speed in Hz , between 0.1 and 2
 modulation _ shape : str , defaul = ' sinusoidal '
 Modulation shpae . One of ' sinusoidal ' or ' triangular '
 See Also
 flanger , tremolo'''

implementation:
if not is_number ( gain_in ) or gain_in <= 0 or gain_in > 1 :
    raise ValueError ( "gain_in must be a number between 0 and 1." )
if not is_number ( gain_out ) or gain_out <= 0 or gain_out > 1 :
    raise ValueError ( "gain_out must be a number between 0 and 1." )
if not is_number ( delay ) or delay <= 0 or delay > 5 :
    raise ValueError ( "delay must be a positive number." )
if not is_number ( decay ) or decay < 0.1 or decay > 0.5 :
    raise ValueError ( "decay must be a number between 0.1 and 0.5." )
if not is_number ( speed ) or speed < 0.1 or speed > 2 :
    raise ValueError ( "speed must be a positive number." )
if modulation_shape not in [ 'sinusoidal' , 'triangular' ] :
    raise ValueError ( "modulation_shape must be one of 'sinusoidal', 'triangular'." )
effect_args = [ 'phaser' , '{:f}' . format ( gain_in ) , '{:f}' . format ( gain_out ) , '{:f}' . format ( delay ) , '{:f}' . format ( decay ) , '{:f}' . format ( speed ) ]
if modulation_shape == 'sinusoidal' :
    effect_args . append ( '-s' )
elif modulation_shape == 'triangular' :
    effect_args . append ( '-t' )
self . effects . extend ( effect_args )
self . effects_log . append ( 'phaser' )
return self


signature:
def pitch ( self , n_semitones , quick = False ) :
    '''Pitch shift the audio without changing the tempo .
 This effect uses the WSOLA algorithm . The audio is chopped up into
 segments which are then shifted in the time domain and overlapped
 ( cross - faded ) at points where their waveforms are most similar as
 determined by measurement of least squares .
 Parameters
 n _ semitones : float
 The number of semitones to shift . Can be positive or negative .
 quick : bool , default = False
 If True , this effect will run faster but with lower sound quality .
 See Also
 bend , speed , tempo'''

implementation:
if not is_number ( n_semitones ) :
    raise ValueError ( "n_semitones must be a positive number" )
if n_semitones < - 12 or n_semitones > 12 :
    logger . warning ( "Using an extreme pitch shift. " "Quality of results will be poor" )
if not isinstance ( quick , bool ) :
    raise ValueError ( "quick must be a boolean." )
effect_args = [ 'pitch' ]
if quick :
    effect_args . append ( '-q' )
effect_args . append ( '{:f}' . format ( n_semitones * 100. ) )
self . effects . extend ( effect_args )
self . effects_log . append ( 'pitch' )
return self


signature:
def rate ( self , samplerate , quality = 'h' ) :
    '''Change the audio sampling rate ( i . e . resample the audio ) to any
 given ` samplerate ` . Better the resampling quality = slower runtime .
 Parameters
 samplerate : float
 Desired sample rate .
 quality : str
 Resampling quality . One of :
 * q : Quick - very low quality ,
 * l : Low ,
 * m : Medium ,
 * h : High ( default ) ,
 * v : Very high
 See Also
 upsample , downsample , convert'''

implementation:
quality_vals = [ 'q' , 'l' , 'm' , 'h' , 'v' ]
if not is_number ( samplerate ) or samplerate <= 0 :
    raise ValueError ( "Samplerate must be a positive number." )
if quality not in quality_vals :
    raise ValueError ( "Quality must be one of {}." . format ( ' ' . join ( quality_vals ) ) )
effect_args = [ 'rate' , '-{}' . format ( quality ) , '{:f}' . format ( samplerate ) ]
self . effects . extend ( effect_args )
self . effects_log . append ( 'rate' )
return self


signature:
def remix ( self , remix_dictionary = None , num_output_channels = None ) :
    '''Remix the channels of an audio file .
 Note : volume options are not yet implemented
 Parameters
 remix _ dictionary : dict or None
 Dictionary mapping output channel to list of input channel ( s ) .
 Empty lists indicate the corresponding output channel should be
 empty . If None , mixes all channels down to a single mono file .
 num _ output _ channels : int or None
 The number of channels in the output file . If None , the number of
 output channels is equal to the largest key in remix _ dictionary .
 If remix _ dictionary is None , this variable is ignored .
 Examples
 Remix a 4 - channel input file . The output file will have
 input channel 2 in channel 1 , a mixdown of input channels 1 an 3 in
 channel 2 , an empty channel 3 , and a copy of input channel 4 in
 channel 4.
 > > > import sox
 > > > tfm = sox . Transformer ( )
 > > > remix _ dictionary = { 1 : [ 2 ] , 2 : [ 1 , 3 ] , 4 : [ 4 ] }
 > > > tfm . remix ( remix _ dictionary )'''

implementation:
if not ( isinstance ( remix_dictionary , dict ) or remix_dictionary is None ) :
    raise ValueError ( "remix_dictionary must be a dictionary or None." )
if remix_dictionary is not None :
    if not all ( [ isinstance ( i , int ) and i > 0 for i in remix_dictionary . keys ( ) ] ) :
        raise ValueError ( "remix dictionary must have positive integer keys." )
    if not all ( [ isinstance ( v , list ) for v in remix_dictionary . values ( ) ] ) :
        raise ValueError ( "remix dictionary values must be lists." )
    for v_list in remix_dictionary . values ( ) :
        if not all ( [ isinstance ( v , int ) and v > 0 for v in v_list ] ) :
            raise ValueError ( "elements of remix dictionary values must " "be positive integers" )
if not ( ( isinstance ( num_output_channels , int ) and num_output_channels > 0 ) or num_output_channels is None ) :
    raise ValueError ( "num_output_channels must be a positive integer or None." )
effect_args = [ 'remix' ]
if remix_dictionary is None :
    effect_args . append ( '-' )
else :
    if num_output_channels is None :
        num_output_channels = max ( remix_dictionary . keys ( ) )
    for channel in range ( 1 , num_output_channels + 1 ) :
        if channel in remix_dictionary . keys ( ) :
            out_channel = ',' . join ( [ str ( i ) for i in remix_dictionary [ channel ] ] )
        else :
            out_channel = '0'
        effect_args . append ( out_channel )
self . effects . extend ( effect_args )
self . effects_log . append ( 'remix' )
return self


signature:
def repeat ( self , count = 1 ) :
    '''Repeat the entire audio count times .
 Parameters
 count : int , default = 1
 The number of times to repeat the audio .'''

implementation:
if not isinstance ( count , int ) or count < 1 :
    raise ValueError ( "count must be a postive integer." )
effect_args = [ 'repeat' , '{}' . format ( count ) ]
self . effects . extend ( effect_args )
self . effects_log . append ( 'repeat' )


signature:
def reverb ( self , reverberance = 50 , high_freq_damping = 50 , room_scale = 100 , stereo_depth = 100 , pre_delay = 0 , wet_gain = 0 , wet_only = False ) :
    '''Add reverberation to the audio using the ‘ freeverb ’ algorithm .
 A reverberation effect is sometimes desirable for concert halls that
 are too small or contain so many people that the hall ’ s natural
 reverberance is diminished . Applying a small amount of stereo reverb
 to a ( dry ) mono signal will usually make it sound more natural .
 Parameters
 reverberance : float , default = 50
 Percentage of reverberance
 high _ freq _ damping : float , default = 50
 Percentage of high - frequency damping .
 room _ scale : float , default = 100
 Scale of the room as a percentage .
 stereo _ depth : float , default = 100
 Stereo depth as a percentage .
 pre _ delay : float , default = 0
 Pre - delay in milliseconds .
 wet _ gain : float , default = 0
 Amount of wet gain in dB
 wet _ only : bool , default = False
 If True , only outputs the wet signal .
 See Also
 echo'''

implementation:
if ( not is_number ( reverberance ) or reverberance < 0 or reverberance > 100 ) :
    raise ValueError ( "reverberance must be between 0 and 100" )
if ( not is_number ( high_freq_damping ) or high_freq_damping < 0 or high_freq_damping > 100 ) :
    raise ValueError ( "high_freq_damping must be between 0 and 100" )
if ( not is_number ( room_scale ) or room_scale < 0 or room_scale > 100 ) :
    raise ValueError ( "room_scale must be between 0 and 100" )
if ( not is_number ( stereo_depth ) or stereo_depth < 0 or stereo_depth > 100 ) :
    raise ValueError ( "stereo_depth must be between 0 and 100" )
if not is_number ( pre_delay ) or pre_delay < 0 :
    raise ValueError ( "pre_delay must be a positive number" )
if not is_number ( wet_gain ) :
    raise ValueError ( "wet_gain must be a number" )
if not isinstance ( wet_only , bool ) :
    raise ValueError ( "wet_only must be a boolean." )
effect_args = [ 'reverb' ]
if wet_only :
    effect_args . append ( '-w' )
effect_args . extend ( [ '{:f}' . format ( reverberance ) , '{:f}' . format ( high_freq_damping ) , '{:f}' . format ( room_scale ) , '{:f}' . format ( stereo_depth ) , '{:f}' . format ( pre_delay ) , '{:f}' . format ( wet_gain ) ] )
self . effects . extend ( effect_args )
self . effects_log . append ( 'reverb' )
return self


signature:
def reverse ( self ) :
    '''Reverse the audio completely'''

implementation:
effect_args = [ 'reverse' ]
self . effects . extend ( effect_args )
self . effects_log . append ( 'reverse' )
return self


signature:
def silence ( self , location = 0 , silence_threshold = 0.1 , min_silence_duration = 0.1 , buffer_around_silence = False ) :
    '''Removes silent regions from an audio file .
 Parameters
 location : int , default = 0
 Where to remove silence . One of :
 * 0 to remove silence throughout the file ( default ) ,
 * 1 to remove silence from the beginning ,
 * - 1 to remove silence from the end ,
 silence _ threshold : float , default = 0.1
 Silence threshold as percentage of maximum sample amplitude .
 Must be between 0 and 100.
 min _ silence _ duration : float , default = 0.1
 The minimum ammount of time in seconds required for a region to be
 considered non - silent .
 buffer _ around _ silence : bool , default = False
 If True , leaves a buffer of min _ silence _ duration around removed
 silent regions .
 See Also
 vad'''

implementation:
if location not in [ - 1 , 0 , 1 ] :
    raise ValueError ( "location must be one of -1, 0, 1." )
if not is_number ( silence_threshold ) or silence_threshold < 0 :
    raise ValueError ( "silence_threshold must be a number between 0 and 100" )
elif silence_threshold >= 100 :
    raise ValueError ( "silence_threshold must be a number between 0 and 100" )
if not is_number ( min_silence_duration ) or min_silence_duration <= 0 :
    raise ValueError ( "min_silence_duration must be a positive number." )
if not isinstance ( buffer_around_silence , bool ) :
    raise ValueError ( "buffer_around_silence must be a boolean." )
effect_args = [ ]
if location == - 1 :
    effect_args . append ( 'reverse' )
if buffer_around_silence :
    effect_args . extend ( [ 'silence' , '-l' ] )
else :
    effect_args . append ( 'silence' )
effect_args . extend ( [ '1' , '{:f}' . format ( min_silence_duration ) , '{:f}%' . format ( silence_threshold ) ] )
if location == 0 :
    effect_args . extend ( [ '-1' , '{:f}' . format ( min_silence_duration ) , '{:f}%' . format ( silence_threshold ) ] )
if location == - 1 :
    effect_args . append ( 'reverse' )
self . effects . extend ( effect_args )
self . effects_log . append ( 'silence' )
return self


signature:
def sinc ( self , filter_type = 'high' , cutoff_freq = 3000 , stop_band_attenuation = 120 , transition_bw = None , phase_response = None ) :
    '''Apply a sinc kaiser - windowed low - pass , high - pass , band - pass , or
 band - reject filter to the signal .
 Parameters
 filter _ type : str , default = ' high '
 Type of filter . One of :
 - ' high ' for a high - pass filter
 - ' low ' for a low - pass filter
 - ' pass ' for a band - pass filter
 - ' reject ' for a band - reject filter
 cutoff _ freq : float or list , default = 3000
 A scalar or length 2 list indicating the filter ' s critical
 frequencies . The critical frequencies are given in Hz and must be
 positive . For a high - pass or low - pass filter , cutoff _ freq
 must be a scalar . For a band - pass or band - reject filter , it must be
 a length 2 list .
 stop _ band _ attenuation : float , default = 120
 The stop band attenuation in dB
 transition _ bw : float , list or None , default = None
 The transition band - width in Hz .
 If None , sox ' s default of 5 % of the total bandwith is used .
 If a float , the given transition bandwith is used for both the
 upper and lower bands ( if applicable ) .
 If a list , the first argument is used for the lower band and the
 second for the upper band .
 phase _ response : float or None
 The filter ' s phase response between 0 ( minimum ) and 100 ( maximum ) .
 If None , sox ' s default phase repsonse is used .
 See Also
 band , bandpass , bandreject , highpass , lowpass'''

implementation:
filter_types = [ 'high' , 'low' , 'pass' , 'reject' ]
if filter_type not in filter_types :
    raise ValueError ( "filter_type must be one of {}" . format ( ', ' . join ( filter_types ) ) )
if not ( is_number ( cutoff_freq ) or isinstance ( cutoff_freq , list ) ) :
    raise ValueError ( "cutoff_freq must be a number or a list" )
if filter_type in [ 'high' , 'low' ] and isinstance ( cutoff_freq , list ) :
    raise ValueError ( "For filter types 'high' and 'low', " "cutoff_freq must be a float, not a list" )
if filter_type in [ 'pass' , 'reject' ] and is_number ( cutoff_freq ) :
    raise ValueError ( "For filter types 'pass' and 'reject', " "cutoff_freq must be a list, not a float" )
if is_number ( cutoff_freq ) and cutoff_freq <= 0 :
    raise ValueError ( "cutoff_freq must be a postive number" )
if isinstance ( cutoff_freq , list ) :
    if len ( cutoff_freq ) != 2 :
        raise ValueError ( "If cutoff_freq is a list it may only have 2 elements." )
    if any ( [ not is_number ( f ) or f <= 0 for f in cutoff_freq ] ) :
        raise ValueError ( "elements of cutoff_freq must be positive numbers" )
    cutoff_freq = sorted ( cutoff_freq )
if not is_number ( stop_band_attenuation ) or stop_band_attenuation < 0 :
    raise ValueError ( "stop_band_attenuation must be a positive number" )
if not ( is_number ( transition_bw ) or isinstance ( transition_bw , list ) or transition_bw is None ) :
    raise ValueError ( "transition_bw must be a number, a list or None." )
if filter_type in [ 'high' , 'low' ] and isinstance ( transition_bw , list ) :
    raise ValueError ( "For filter types 'high' and 'low', " "transition_bw must be a float, not a list" )
if is_number ( transition_bw ) and transition_bw <= 0 :
    raise ValueError ( "transition_bw must be a postive number" )
if isinstance ( transition_bw , list ) :
    if any ( [ not is_number ( f ) or f <= 0 for f in transition_bw ] ) :
        raise ValueError ( "elements of transition_bw must be positive numbers" )
    if len ( transition_bw ) != 2 :
        raise ValueError ( "If transition_bw is a list it may only have 2 elements." )
if phase_response is not None and not is_number ( phase_response ) :
    raise ValueError ( "phase_response must be a number or None." )
if ( is_number ( phase_response ) and ( phase_response < 0 or phase_response > 100 ) ) :
    raise ValueError ( "phase response must be between 0 and 100" )
effect_args = [ 'sinc' ]
effect_args . extend ( [ '-a' , '{:f}' . format ( stop_band_attenuation ) ] )
if phase_response is not None :
    effect_args . extend ( [ '-p' , '{:f}' . format ( phase_response ) ] )
if filter_type == 'high' :
    if transition_bw is not None :
        effect_args . extend ( [ '-t' , '{:f}' . format ( transition_bw ) ] )
    effect_args . append ( '{:f}' . format ( cutoff_freq ) )
elif filter_type == 'low' :
    effect_args . append ( '-{:f}' . format ( cutoff_freq ) )
    if transition_bw is not None :
        effect_args . extend ( [ '-t' , '{:f}' . format ( transition_bw ) ] )
else :
    if is_number ( transition_bw ) :
        effect_args . extend ( [ '-t' , '{:f}' . format ( transition_bw ) ] )
    elif isinstance ( transition_bw , list ) :
        effect_args . extend ( [ '-t' , '{:f}' . format ( transition_bw [ 0 ] ) ] )
if filter_type == 'pass' :
    effect_args . append ( '{:f}-{:f}' . format ( cutoff_freq [ 0 ] , cutoff_freq [ 1 ] ) )
elif filter_type == 'reject' :
    effect_args . append ( '{:f}-{:f}' . format ( cutoff_freq [ 1 ] , cutoff_freq [ 0 ] ) )
if isinstance ( transition_bw , list ) :
    effect_args . extend ( [ '-t' , '{:f}' . format ( transition_bw [ 1 ] ) ] )
self . effects . extend ( effect_args )
self . effects_log . append ( 'sinc' )
return self


signature:
def speed ( self , factor ) :
    '''Adjust the audio speed ( pitch and tempo together ) .
 Technically , the speed effect only changes the sample rate information ,
 leaving the samples themselves untouched . The rate effect is invoked
 automatically to resample to the output sample rate , using its default
 quality / speed . For higher quality or higher speed resampling , in
 addition to the speed effect , specify the rate effect with the desired
 quality option .
 Parameters
 factor : float
 The ratio of the new speed to the old speed .
 For ex . 1.1 speeds up the audio by 10 % ; 0.9 slows it down by 10 % .
 Note - this argument is the inverse of what is passed to the sox
 stretch effect for consistency with speed .
 See Also
 rate , tempo , pitch'''

implementation:
if not is_number ( factor ) or factor <= 0 :
    raise ValueError ( "factor must be a positive number" )
if factor < 0.5 or factor > 2 :
    logger . warning ( "Using an extreme factor. Quality of results will be poor" )
effect_args = [ 'speed' , '{:f}' . format ( factor ) ]
self . effects . extend ( effect_args )
self . effects_log . append ( 'speed' )
return self


signature:
def stat ( self , input_filepath , scale = None , rms = False ) :
    '''Display time and frequency domain statistical information about the
 audio . Audio is passed unmodified through the SoX processing chain .
 Unlike other Transformer methods , this does not modify the transformer
 effects chain . Instead it computes statistics on the output file that
 would be created if the build command were invoked .
 Note : The file is downmixed to mono prior to computation .
 Parameters
 input _ filepath : str
 Path to input file to compute stats on .
 scale : float or None , default = None
 If not None , scales the input by the given scale factor .
 rms : bool , default = False
 If True , scales all values by the average rms amplitude .
 Returns
 stat _ dict : dict
 Dictionary of statistics .
 See Also
 stats , power _ spectrum , sox . file _ info'''

implementation:
effect_args = [ 'channels' , '1' , 'stat' ]
if scale is not None :
    if not is_number ( scale ) or scale <= 0 :
        raise ValueError ( "scale must be a positive number." )
    effect_args . extend ( [ '-s' , '{:f}' . format ( scale ) ] )
if rms :
    effect_args . append ( '-rms' )
_ , _ , stat_output = self . build ( input_filepath , None , extra_args = effect_args , return_output = True )
stat_dict = { }
lines = stat_output . split ( '\n' )
for line in lines :
    split_line = line . split ( )
    if len ( split_line ) == 0 :
        continue
    value = split_line [ - 1 ]
    key = ' ' . join ( split_line [ : - 1 ] )
    stat_dict [ key . strip ( ':' ) ] = value
return stat_dict


signature:
def power_spectrum ( self , input_filepath ) :
    '''Calculates the power spectrum ( 4096 point DFT ) . This method
 internally invokes the stat command with the - freq option .
 Note : The file is downmixed to mono prior to computation .
 Parameters
 input _ filepath : str
 Path to input file to compute stats on .
 Returns
 power _ spectrum : list
 List of frequency ( Hz ) , amplitude pairs .
 See Also
 stat , stats , sox . file _ info'''

implementation:
effect_args = [ 'channels' , '1' , 'stat' , '-freq' ]
_ , _ , stat_output = self . build ( input_filepath , None , extra_args = effect_args , return_output = True )
power_spectrum = [ ]
lines = stat_output . split ( '\n' )
for line in lines :
    split_line = line . split ( )
    if len ( split_line ) != 2 :
        continue
    freq , amp = split_line
    power_spectrum . append ( [ float ( freq ) , float ( amp ) ] )
return power_spectrum


signature:
def stats ( self , input_filepath ) :
    '''Display time domain statistical information about the audio
 channels . Audio is passed unmodified through the SoX processing chain .
 Statistics are calculated and displayed for each audio channel
 Unlike other Transformer methods , this does not modify the transformer
 effects chain . Instead it computes statistics on the output file that
 would be created if the build command were invoked .
 Note : The file is downmixed to mono prior to computation .
 Parameters
 input _ filepath : str
 Path to input file to compute stats on .
 Returns
 stats _ dict : dict
 List of frequency ( Hz ) , amplitude pairs .
 See Also
 stat , sox . file _ info'''

implementation:
effect_args = [ 'channels' , '1' , 'stats' ]
_ , _ , stats_output = self . build ( input_filepath , None , extra_args = effect_args , return_output = True )
stats_dict = { }
lines = stats_output . split ( '\n' )
for line in lines :
    split_line = line . split ( )
    if len ( split_line ) == 0 :
        continue
    value = split_line [ - 1 ]
    key = ' ' . join ( split_line [ : - 1 ] )
    stats_dict [ key ] = value
return stats_dict


signature:
def stretch ( self , factor , window = 20 ) :
    '''Change the audio duration ( but not its pitch ) .
 * * Unless factor is close to 1 , use the tempo effect instead . * *
 This effect is broadly equivalent to the tempo effect with search set
 to zero , so in general , its results are comparatively poor ; it is
 retained as it can sometimes out - perform tempo for small factors .
 Parameters
 factor : float
 The ratio of the new tempo to the old tempo .
 For ex . 1.1 speeds up the tempo by 10 % ; 0.9 slows it down by 10 % .
 Note - this argument is the inverse of what is passed to the sox
 stretch effect for consistency with tempo .
 window : float , default = 20
 Window size in miliseconds
 See Also
 tempo , speed , pitch'''

implementation:
if not is_number ( factor ) or factor <= 0 :
    raise ValueError ( "factor must be a positive number" )
if factor < 0.5 or factor > 2 :
    logger . warning ( "Using an extreme time stretching factor. " "Quality of results will be poor" )
if abs ( factor - 1.0 ) > 0.1 :
    logger . warning ( "For this stretch factor, " "the tempo effect has better performance." )
if not is_number ( window ) or window <= 0 :
    raise ValueError ( "window must be a positive number." )
effect_args = [ 'stretch' , '{:f}' . format ( factor ) , '{:f}' . format ( window ) ]
self . effects . extend ( effect_args )
self . effects_log . append ( 'stretch' )
return self


signature:
def swap ( self ) :
    '''Swap stereo channels . If the input is not stereo , pairs of channels
 are swapped , and a possible odd last channel passed through .
 E . g . , for seven channels , the output order will be 2 , 1 , 4 , 3 , 6 , 5 , 7.
 See Also
 remix'''

implementation:
effect_args = [ 'swap' ]
self . effects . extend ( effect_args )
self . effects_log . append ( 'swap' )
return self


signature:
def tempo ( self , factor , audio_type = None , quick = False ) :
    '''Time stretch audio without changing pitch .
 This effect uses the WSOLA algorithm . The audio is chopped up into
 segments which are then shifted in the time domain and overlapped
 ( cross - faded ) at points where their waveforms are most similar as
 determined by measurement of least squares .
 Parameters
 factor : float
 The ratio of new tempo to the old tempo .
 For ex . 1.1 speeds up the tempo by 10 % ; 0.9 slows it down by 10 % .
 audio _ type : str
 Type of audio , which optimizes algorithm parameters . One of :
 * m : Music ,
 * s : Speech ,
 * l : Linear ( useful when factor is close to 1 ) ,
 quick : bool , default = False
 If True , this effect will run faster but with lower sound quality .
 See Also
 stretch , speed , pitch'''

implementation:
if not is_number ( factor ) or factor <= 0 :
    raise ValueError ( "factor must be a positive number" )
if factor < 0.5 or factor > 2 :
    logger . warning ( "Using an extreme time stretching factor. " "Quality of results will be poor" )
if abs ( factor - 1.0 ) <= 0.1 :
    logger . warning ( "For this stretch factor, " "the stretch effect has better performance." )
if audio_type not in [ None , 'm' , 's' , 'l' ] :
    raise ValueError ( "audio_type must be one of None, 'm', 's', or 'l'." )
if not isinstance ( quick , bool ) :
    raise ValueError ( "quick must be a boolean." )
effect_args = [ 'tempo' ]
if quick :
    effect_args . append ( '-q' )
if audio_type is not None :
    effect_args . append ( '-{}' . format ( audio_type ) )
effect_args . append ( '{:f}' . format ( factor ) )
self . effects . extend ( effect_args )
self . effects_log . append ( 'tempo' )
return self


signature:
def treble ( self , gain_db , frequency = 3000.0 , slope = 0.5 ) :
    '''Boost or cut the treble ( lower ) frequencies of the audio using a
 two - pole shelving filter with a response similar to that of a standard
 hi - fi ’ s tone - controls . This is also known as shelving equalisation .
 The filters are described in detail in
 http : / / musicdsp . org / files / Audio - EQ - Cookbook . txt
 Parameters
 gain _ db : float
 The gain at the Nyquist frequency .
 For a large cut use - 20 , for a large boost use 20.
 frequency : float , default = 100.0
 The filter ' s cutoff frequency in Hz .
 slope : float , default = 0.5
 The steepness of the filter ' s shelf transition .
 For a gentle slope use 0.3 , and use 1.0 for a steep slope .
 See Also
 bass , equalizer'''

implementation:
if not is_number ( gain_db ) :
    raise ValueError ( "gain_db must be a number" )
if not is_number ( frequency ) or frequency <= 0 :
    raise ValueError ( "frequency must be a positive number." )
if not is_number ( slope ) or slope <= 0 or slope > 1.0 :
    raise ValueError ( "width_q must be a positive number." )
effect_args = [ 'treble' , '{:f}' . format ( gain_db ) , '{:f}' . format ( frequency ) , '{:f}s' . format ( slope ) ]
self . effects . extend ( effect_args )
self . effects_log . append ( 'treble' )
return self


signature:
def tremolo ( self , speed = 6.0 , depth = 40.0 ) :
    '''Apply a tremolo ( low frequency amplitude modulation ) effect to the
 audio . The tremolo frequency in Hz is giv en by speed , and the depth
 as a percentage by depth ( default 40 ) .
 Parameters
 speed : float
 Tremolo speed in Hz .
 depth : float
 Tremolo depth as a percentage of the total amplitude .
 See Also
 flanger
 Examples
 > > > tfm = sox . Transformer ( )
 For a growl - type effect
 > > > tfm . tremolo ( speed = 100.0)'''

implementation:
if not is_number ( speed ) or speed <= 0 :
    raise ValueError ( "speed must be a positive number." )
if not is_number ( depth ) or depth <= 0 or depth > 100 :
    raise ValueError ( "depth must be a positive number less than 100." )
effect_args = [ 'tremolo' , '{:f}' . format ( speed ) , '{:f}' . format ( depth ) ]
self . effects . extend ( effect_args )
self . effects_log . append ( 'tremolo' )
return self


signature:
def trim ( self , start_time , end_time = None ) :
    '''Excerpt a clip from an audio file , given the start timestamp and end timestamp of the clip within the file , expressed in seconds . If the end timestamp is set to ` None ` or left unspecified , it defaults to the duration of the audio file .
 Parameters
 start _ time : float
 Start time of the clip ( seconds )
 end _ time : float or None , default = None
 End time of the clip ( seconds )'''

implementation:
if not is_number ( start_time ) or start_time < 0 :
    raise ValueError ( "start_time must be a positive number." )
effect_args = [ 'trim' , '{:f}' . format ( start_time ) ]
if end_time is not None :
    if not is_number ( end_time ) or end_time < 0 :
        raise ValueError ( "end_time must be a positive number." )
    if start_time >= end_time :
        raise ValueError ( "start_time must be smaller than end_time." )
    effect_args . append ( '{:f}' . format ( end_time - start_time ) )
self . effects . extend ( effect_args )
self . effects_log . append ( 'trim' )
return self


signature:
def vad ( self , location = 1 , normalize = True , activity_threshold = 7.0 , min_activity_duration = 0.25 , initial_search_buffer = 1.0 , max_gap = 0.25 , initial_pad = 0.0 ) :
    '''Voice Activity Detector . Attempts to trim silence and quiet
 background sounds from the ends of recordings of speech . The algorithm
 currently uses a simple cepstral power measurement to detect voice , so
 may be fooled by other things , especially music .
 The effect can trim only from the front of the audio , so in order to
 trim from the back , the reverse effect must also be used .
 Parameters
 location : 1 or - 1 , default = 1
 If 1 , trims silence from the beginning
 If - 1 , trims silence from the end
 normalize : bool , default = True
 If true , normalizes audio before processing .
 activity _ threshold : float , default = 7.0
 The measurement level used to trigger activity detection . This may
 need to be cahnged depending on the noise level , signal level , and
 other characteristics of the input audio .
 min _ activity _ duration : float , default = 0.25
 The time constant ( in seconds ) used to help ignore short bursts of
 sound .
 initial _ search _ buffer : float , default = 1.0
 The amount of audio ( in seconds ) to search for quieter / shorter
 bursts of audio to include prior to the detected trigger point .
 max _ gap : float , default = 0.25
 The allowed gap ( in seconds ) between quiteter / shorter bursts of
 audio to include prior to the detected trigger point
 initial _ pad : float , default = 0.0
 The amount of audio ( in seconds ) to preserve before the trigger
 point and any found quieter / shorter bursts .
 See Also
 silence
 Examples
 > > > tfm = sox . Transformer ( )
 Remove silence from the beginning of speech
 > > > tfm . vad ( initial _ pad = 0.3)
 Remove silence from the end of speech
 > > > tfm . vad ( location = - 1 , initial _ pad = 0.2)'''

implementation:
if location not in [ - 1 , 1 ] :
    raise ValueError ( "location must be -1 or 1." )
if not isinstance ( normalize , bool ) :
    raise ValueError ( "normalize muse be a boolean." )
if not is_number ( activity_threshold ) :
    raise ValueError ( "activity_threshold must be a number." )
if not is_number ( min_activity_duration ) or min_activity_duration < 0 :
    raise ValueError ( "min_activity_duration must be a positive number" )
if not is_number ( initial_search_buffer ) or initial_search_buffer < 0 :
    raise ValueError ( "initial_search_buffer must be a positive number" )
if not is_number ( max_gap ) or max_gap < 0 :
    raise ValueError ( "max_gap must be a positive number." )
if not is_number ( initial_pad ) or initial_pad < 0 :
    raise ValueError ( "initial_pad must be a positive number." )
effect_args = [ ]
if normalize :
    effect_args . append ( 'norm' )
if location == - 1 :
    effect_args . append ( 'reverse' )
effect_args . extend ( [ 'vad' , '-t' , '{:f}' . format ( activity_threshold ) , '-T' , '{:f}' . format ( min_activity_duration ) , '-s' , '{:f}' . format ( initial_search_buffer ) , '-g' , '{:f}' . format ( max_gap ) , '-p' , '{:f}' . format ( initial_pad ) ] )
if location == - 1 :
    effect_args . append ( 'reverse' )
self . effects . extend ( effect_args )
self . effects_log . append ( 'vad' )
return self


signature:
def vol ( self , gain , gain_type = 'amplitude' , limiter_gain = None ) :
    '''Apply an amplification or an attenuation to the audio signal .
 Parameters
 gain : float
 Interpreted according to the given ` gain _ type ` .
 If ` gain _ type ' = ' amplitude ' , ` gain ' is a positive amplitude ratio .
 If ` gain _ type ' = ' power ' , ` gain ' is a power ( voltage squared ) .
 If ` gain _ type ' = ' db ' , ` gain ' is in decibels .
 gain _ type : string , default = ' amplitude '
 Type of gain . One of :
 - ' amplitude '
 - ' power '
 - ' db '
 limiter _ gain : float or None , default = None
 If specified , a limiter is invoked on peaks greater than
 ` limiter _ gain ' to prevent clipping .
 ` limiter _ gain ` should be a positive value much less than 1.
 See Also
 gain , compand'''

implementation:
if not is_number ( gain ) :
    raise ValueError ( 'gain must be a number.' )
if limiter_gain is not None :
    if ( not is_number ( limiter_gain ) or limiter_gain <= 0 or limiter_gain >= 1 ) :
        raise ValueError ( 'limiter gain must be a positive number less than 1' )
if gain_type in [ 'amplitude' , 'power' ] and gain < 0 :
    raise ValueError ( "If gain_type = amplitude or power, gain must be positive." )
effect_args = [ 'vol' ]
effect_args . append ( '{:f}' . format ( gain ) )
if gain_type == 'amplitude' :
    effect_args . append ( 'amplitude' )
elif gain_type == 'power' :
    effect_args . append ( 'power' )
elif gain_type == 'db' :
    effect_args . append ( 'dB' )
else :
    raise ValueError ( 'gain_type must be one of amplitude power or db' )
if limiter_gain is not None :
    if gain_type in [ 'amplitude' , 'power' ] and gain > 1 :
        effect_args . append ( '{:f}' . format ( limiter_gain ) )
    elif gain_type == 'db' and gain > 0 :
        effect_args . append ( '{:f}' . format ( limiter_gain ) )
self . effects . extend ( effect_args )
self . effects_log . append ( 'vol' )
return self


signature:
def join ( self , room ) :
    """Lets a user join a room on a specific Namespace ."""

implementation: self . socket . rooms . add ( self . _get_room_name ( room ) )


signature:
def leave ( self , room ) :
    """Lets a user leave a room on a specific Namespace ."""

implementation: self . socket . rooms . remove ( self . _get_room_name ( room ) )


signature:
def socketio_manage ( environ , namespaces , request = None , error_handler = None , json_loads = None , json_dumps = None ) :
    """Main SocketIO management function , call from within your Framework of
 choice ' s view .
 The ` ` environ ` ` variable is the WSGI ` ` environ ` ` . It is used to extract
 Socket object from the underlying server ( as the ' socketio ' key ) , and will
 be attached to both the ` ` Socket ` ` and ` ` Namespace ` ` objects .
 The ` ` namespaces ` ` parameter is a dictionary of the namespace string
 representation as key , and the BaseNamespace namespace class descendant as
 a value . The empty string ( ' ' ) namespace is the global namespace . You can
 use Socket . GLOBAL _ NS to be more explicit . So it would look like :
 . . code - block : : python
 namespaces = { ' ' : GlobalNamespace ,
 ' / chat ' : ChatNamespace }
 The ` ` request ` ` object is not required , but will probably be useful to pass
 framework - specific things into your Socket and Namespace functions . It will
 simply be attached to the Socket and Namespace object ( accessible through
 ` ` self . request ` ` in both cases ) , and it is not accessed in any case by the
 ` ` gevent - socketio ` ` library .
 Pass in an ` ` error _ handler ` ` if you want to override the default
 error _ handler ( which is : func : ` socketio . virtsocket . default _ error _ handler ` .
 The callable you pass in should have the same signature as the default
 error handler .
 The ` ` json _ loads ` ` and ` ` json _ dumps ` ` are overrides for the default
 ` ` json . loads ` ` and ` ` json . dumps ` ` function calls . Override these at
 the top - most level here . This will affect all sockets created by this
 socketio manager , and all namespaces inside .
 This function will block the current " view " or " controller " in your
 framework to do the recv / send on the socket , and dispatch incoming messages
 to your namespaces .
 This is a simple example using Pyramid :
 . . code - block : : python
 def my _ view ( request ) :
 socketio _ manage ( request . environ , { ' ' : GlobalNamespace } , request )
 NOTE : You must understand that this function is going to be called
 * only once * per socket opening , * even though * you are using a long
 polling mechanism . The subsequent calls ( for long polling ) will
 be hooked directly at the server - level , to interact with the
 active ` ` Socket ` ` instance . This means you will * not * get access
 to the future ` ` request ` ` or ` ` environ ` ` objects . This is of
 particular importance regarding sessions ( like Beaker ) . The
 session will be opened once at the opening of the Socket , and not
 closed until the socket is closed . You are responsible for
 opening and closing the cookie - based session yourself if you want
 to keep its data in sync with the rest of your GET / POST calls ."""

implementation:
socket = environ [ 'socketio' ]
socket . _set_environ ( environ )
socket . _set_namespaces ( namespaces )
if request :
    socket . _set_request ( request )
if error_handler :
    socket . _set_error_handler ( error_handler )
if json_loads :
    socket . _set_json_loads ( json_loads )
if json_dumps :
    socket . _set_json_dumps ( json_dumps )
receiver_loop = socket . _spawn_receiver_loop ( )
gevent . joinall ( [ receiver_loop ] )
# TODO : double check , what happens to the WSGI request here ? it vanishes ?
return


signature:
def default_error_handler ( socket , error_name , error_message , endpoint , msg_id , quiet ) :
    """This is the default error handler , you can override this when
 calling : func : ` socketio . socketio _ manage ` .
 It basically sends an event through the socket with the ' error ' name .
 See documentation for : meth : ` Socket . error ` .
 : param quiet : if quiet , this handler will not send a packet to the
 user , but only log for the server developer ."""

implementation:
pkt = dict ( type = 'event' , name = 'error' , args = [ error_name , error_message ] , endpoint = endpoint )
if msg_id :
    pkt [ 'id' ] = msg_id
# Send an error event through the Socket
if not quiet :
    socket . send_packet ( pkt )
# Log that error somewhere for debugging . . .
log . error ( u"default_error_handler: {}, {} (endpoint={}, msg_id={})" . format ( error_name , error_message , endpoint , msg_id ) )


signature:
def _save_ack_callback ( self , msgid , callback ) :
    """Keep a reference of the callback on this socket ."""

implementation:
if msgid in self . ack_callbacks :
    return False
self . ack_callbacks [ msgid ] = callback


signature:
def _pop_ack_callback ( self , msgid ) :
    """Fetch the callback for a given msgid , if it exists , otherwise ,
 return None"""

implementation:
if msgid not in self . ack_callbacks :
    return None
return self . ack_callbacks . pop ( msgid )


signature:
def kill ( self , detach = False ) :
    """This function must / will be called when a socket is to be completely
 shut down , closed by connection timeout , connection error or explicit
 disconnection from the client .
 It will call all of the Namespace ' s
 : meth : ` ~ socketio . namespace . BaseNamespace . disconnect ` methods
 so that you can shut - down things properly ."""

implementation:
# Clear out the callbacks
self . ack_callbacks = { }
if self . connected :
    self . state = self . STATE_DISCONNECTING
    self . server_queue . put_nowait ( None )
    self . client_queue . put_nowait ( None )
    if len ( self . active_ns ) > 0 :
        log . debug ( "Calling disconnect() on %s" % self )
        self . disconnect ( )
if detach :
    self . detach ( )
gevent . killall ( self . jobs )


signature:
def detach ( self ) :
    """Detach this socket from the server . This should be done in
 conjunction with kill ( ) , once all the jobs are dead , detach the
 socket for garbage collection ."""

implementation:
log . debug ( "Removing %s from server sockets" % self )
if self . sessid in self . server . sockets :
    self . server . sockets . pop ( self . sessid )


signature:
def get_multiple_client_msgs ( self , ** kwargs ) :
    """Get multiple messages , in case we ' re going through the various
 XHR - polling methods , on which we can pack more than one message if the
 rate is high , and encode the payload for the HTTP channel ."""

implementation:
client_queue = self . client_queue
msgs = [ client_queue . get ( ** kwargs ) ]
while client_queue . qsize ( ) :
    msgs . append ( client_queue . get ( ) )
return msgs


signature:
def error ( self , error_name , error_message , endpoint = None , msg_id = None , quiet = False ) :
    """Send an error to the user , using the custom or default
 ErrorHandler configured on the [ TODO : Revise this ] Socket / Handler
 object .
 : param error _ name : is a simple string , for easy association on
 the client side
 : param error _ message : is a human readable message , the user
 will eventually see
 : param endpoint : set this if you have a message specific to an
 end point
 : param msg _ id : set this if your error is relative to a
 specific message
 : param quiet : way to make the error handler quiet . Specific to
 the handler . The default handler will only log ,
 with quiet ."""

implementation:
handler = self . error_handler
return handler ( self , error_name , error_message , endpoint , msg_id , quiet )


signature:
def disconnect ( self , silent = False ) :
    """Calling this method will call the
 : meth : ` ~ socketio . namespace . BaseNamespace . disconnect ` method on
 all the active Namespaces that were open , killing all their
 jobs and sending ' disconnect ' packets for each of them .
 Normally , the Global namespace ( endpoint = ' ' ) has special meaning ,
 as it represents the whole connection ,
 : param silent : when True , pass on the ` ` silent ` ` flag to the Namespace
 : meth : ` ~ socketio . namespace . BaseNamespace . disconnect `
 calls ."""

implementation:
for ns_name , ns in list ( six . iteritems ( self . active_ns ) ) :
    ns . recv_disconnect ( )


signature:
def remove_namespace ( self , namespace ) :
    """This removes a Namespace object from the socket .
 This is usually called by
 : meth : ` ~ socketio . namespace . BaseNamespace . disconnect ` ."""

implementation:
if namespace in self . active_ns :
    del self . active_ns [ namespace ]
if len ( self . active_ns ) == 0 and self . connected :
    self . kill ( detach = True )


signature:
def send_packet ( self , pkt ) :
    """Low - level interface to queue a packet on the wire ( encoded as wire
 protocol"""

implementation: self . put_client_msg ( packet . encode ( pkt , self . json_dumps ) )


signature:
def spawn ( self , fn , * args , ** kwargs ) :
    """Spawn a new Greenlet , attached to this Socket instance .
 It will be monitored by the " watcher " method"""

implementation:
log . debug ( "Spawning sub-Socket Greenlet: %s" % fn . __name__ )
job = gevent . spawn ( fn , * args , ** kwargs )
self . jobs . append ( job )
return job


signature:
def _receiver_loop ( self ) :
    """This is the loop that takes messages from the queue for the server
 to consume , decodes them and dispatches them .
 It is the main loop for a socket . We join on this process before
 returning control to the web framework .
 This process is not tracked by the socket itself , it is not going
 to be killed by the ` ` gevent . killall ( socket . jobs ) ` ` , so it must
 exit gracefully itself ."""

implementation:
while True :
    rawdata = self . get_server_msg ( )
    if not rawdata :
        continue
        # or close the connection ?
    try :
        pkt = packet . decode ( rawdata , self . json_loads )
    except ( ValueError , KeyError , Exception ) as e :
        self . error ( 'invalid_packet' , "There was a decoding error when dealing with packet " "with event: %s... (%s)" % ( rawdata [ : 20 ] , e ) )
        continue
    if pkt [ 'type' ] == 'heartbeat' : # This is already dealth with in put _ server _ msg ( ) when
    # any incoming raw data arrives .
        continue
    if pkt [ 'type' ] == 'disconnect' and pkt [ 'endpoint' ] == '' : # On global namespace , we kill everything .
        self . kill ( detach = True )
        continue
    endpoint = pkt [ 'endpoint' ]
    if endpoint not in self . namespaces :
        self . error ( "no_such_namespace" , "The endpoint you tried to connect to " "doesn't exist: %s" % endpoint , endpoint = endpoint )
        continue
    elif endpoint in self . active_ns :
        pkt_ns = self . active_ns [ endpoint ]
    else :
        new_ns_class = self . namespaces [ endpoint ]
        pkt_ns = new_ns_class ( self . environ , endpoint , request = self . request )
        # This calls initialize ( ) on all the classes and mixins , etc . .
        # in the order of the MRO
        for cls in type ( pkt_ns ) . __mro__ :
            if hasattr ( cls , 'initialize' ) :
                cls . initialize ( pkt_ns )
        # use this instead of _ _ init _ _ ,
        # for less confusion
        self . active_ns [ endpoint ] = pkt_ns
    retval = pkt_ns . process_packet ( pkt )
    # Has the client requested an ' ack ' with the reply parameters ?
    if pkt . get ( 'ack' ) == "data" and pkt . get ( 'id' ) :
        if type ( retval ) is tuple :
            args = list ( retval )
        else :
            args = [ retval ]
        returning_ack = dict ( type = 'ack' , ackId = pkt [ 'id' ] , args = args , endpoint = pkt . get ( 'endpoint' , '' ) )
        self . send_packet ( returning_ack )
    # Now , are we still connected ?
    if not self . connected :
        self . kill ( detach = True )
        # ? ? what , s the best clean - up
        # when its not a
        # user - initiated disconnect
        return


signature:
def _spawn_receiver_loop ( self ) :
    """Spawns the reader loop . This is called internall by
 socketio _ manage ( ) ."""

implementation:
job = gevent . spawn ( self . _receiver_loop )
self . jobs . append ( job )
return job


signature:
def _watcher ( self ) :
    """Watch out if we ' ve been disconnected , in that case , kill
 all the jobs ."""

implementation:
while True :
    gevent . sleep ( 1.0 )
    if not self . connected :
        for ns_name , ns in list ( six . iteritems ( self . active_ns ) ) :
            ns . recv_disconnect ( )
        # Killing Socket - level jobs
        gevent . killall ( self . jobs )
        break


signature:
def _heartbeat ( self ) :
    """Start the heartbeat Greenlet to check connection health ."""

implementation:
interval = self . config [ 'heartbeat_interval' ]
while self . connected :
    gevent . sleep ( interval )
    # TODO : this process could use a timeout object like the disconnect
    # timeout thing , and ONLY send packets when none are sent !
    # We would do that by calling timeout . set ( ) for a " sending "
    # timeout . If we ' re sending 100 messages a second , there is
    # no need to push some heartbeats in there also .
    self . put_client_msg ( "2::" )


signature:
def _spawn_heartbeat ( self ) :
    """This functions returns a list of jobs"""

implementation:
self . spawn ( self . _heartbeat )
self . spawn ( self . _heartbeat_timeout )


signature:
def encode ( data , json_dumps = default_json_dumps ) :
    """Encode an attribute dict into a byte string ."""

implementation:
payload = ''
msg = str ( MSG_TYPES [ data [ 'type' ] ] )
if msg in [ '0' , '1' ] : # '1 : : ' [ path ] [ query ]
    msg += '::' + data [ 'endpoint' ]
    if 'qs' in data and data [ 'qs' ] != '' :
        msg += ':' + data [ 'qs' ]
elif msg == '2' : # heartbeat
    msg += '::'
elif msg in [ '3' , '4' , '5' ] : # '3 : ' [ id ( ' + ' ) ] ' : ' [ endpoint ] ' : ' [ data ]
# '4 : ' [ id ( ' + ' ) ] ' : ' [ endpoint ] ' : ' [ json ]
# '5 : ' [ id ( ' + ' ) ] ' : ' [ endpoint ] ' : ' [ json encoded event ]
# The message id is an incremental integer , required for ACKs .
# If the message id is followed by a + , the ACK is not handled by
# socket . io , but by the user instead .
    if msg == '3' :
        payload = data [ 'data' ]
    if msg == '4' :
        payload = json_dumps ( data [ 'data' ] )
    if msg == '5' :
        d = { }
        d [ 'name' ] = data [ 'name' ]
        if 'args' in data and data [ 'args' ] != [ ] :
            d [ 'args' ] = data [ 'args' ]
        payload = json_dumps ( d )
    if 'id' in data :
        msg += ':' + str ( data [ 'id' ] )
        if data [ 'ack' ] == 'data' :
            msg += '+'
        msg += ':'
    else :
        msg += '::'
    if 'endpoint' not in data :
        data [ 'endpoint' ] = ''
    if payload != '' :
        msg += data [ 'endpoint' ] + ':' + payload
    else :
        msg += data [ 'endpoint' ]
elif msg == '6' : # '6 : : : ' [ id ] ' + ' [ data ]
    msg += '::' + data . get ( 'endpoint' , '' ) + ':' + str ( data [ 'ackId' ] )
    if 'args' in data and data [ 'args' ] != [ ] :
        msg += '+' + json_dumps ( data [ 'args' ] )
elif msg == '7' : # '7 : : ' [ endpoint ] ' : ' [ reason ] ' + ' [ advice ]
    msg += ':::'
    if 'reason' in data and data [ 'reason' ] != '' :
        msg += str ( ERROR_REASONS [ data [ 'reason' ] ] )
    if 'advice' in data and data [ 'advice' ] != '' :
        msg += '+' + str ( ERROR_ADVICES [ data [ 'advice' ] ] )
    msg += data [ 'endpoint' ]
# NoOp , used to close a poll after the polling duration time
elif msg == '8' :
    msg += '::'
return msg


signature:
def decode ( rawstr , json_loads = default_json_loads ) :
    """Decode a rawstr packet arriving from the socket into a dict ."""

implementation:
decoded_msg = { }
try : # Handle decoding in Python < 3.
    rawstr = rawstr . decode ( 'utf-8' )
except AttributeError :
    pass
split_data = rawstr . split ( ":" , 3 )
msg_type = split_data [ 0 ]
msg_id = split_data [ 1 ]
endpoint = split_data [ 2 ]
data = ''
if msg_id != '' :
    if "+" in msg_id :
        msg_id = msg_id . split ( '+' ) [ 0 ]
        decoded_msg [ 'id' ] = int ( msg_id )
        decoded_msg [ 'ack' ] = 'data'
    else :
        decoded_msg [ 'id' ] = int ( msg_id )
        decoded_msg [ 'ack' ] = True
# common to every message
msg_type_id = int ( msg_type )
if msg_type_id in MSG_VALUES :
    decoded_msg [ 'type' ] = MSG_VALUES [ int ( msg_type ) ]
else :
    raise Exception ( "Unknown message type: %s" % msg_type )
decoded_msg [ 'endpoint' ] = endpoint
if len ( split_data ) > 3 :
    data = split_data [ 3 ]
if msg_type == "0" : # disconnect
    pass
elif msg_type == "1" : # connect
    decoded_msg [ 'qs' ] = data
elif msg_type == "2" : # heartbeat
    pass
elif msg_type == "3" : # message
    decoded_msg [ 'data' ] = data
elif msg_type == "4" : # json msg
    decoded_msg [ 'data' ] = json_loads ( data )
elif msg_type == "5" : # event
    try :
        data = json_loads ( data )
    except ValueError :
        print ( "Invalid JSON event message" , data )
        decoded_msg [ 'args' ] = [ ]
    else :
        decoded_msg [ 'name' ] = data . pop ( 'name' )
        if 'args' in data :
            decoded_msg [ 'args' ] = data [ 'args' ]
        else :
            decoded_msg [ 'args' ] = [ ]
elif msg_type == "6" : # ack
    if '+' in data :
        ackId , data = data . split ( '+' )
        decoded_msg [ 'ackId' ] = int ( ackId )
        decoded_msg [ 'args' ] = json_loads ( data )
    else :
        decoded_msg [ 'ackId' ] = int ( data )
        decoded_msg [ 'args' ] = [ ]
elif msg_type == "7" : # error
    if '+' in data :
        reason , advice = data . split ( '+' )
        decoded_msg [ 'reason' ] = REASONS_VALUES [ int ( reason ) ]
        decoded_msg [ 'advice' ] = ADVICES_VALUES [ int ( advice ) ]
    else :
        decoded_msg [ 'advice' ] = ''
        if data != '' :
            decoded_msg [ 'reason' ] = REASONS_VALUES [ int ( data ) ]
        else :
            decoded_msg [ 'reason' ] = ''
elif msg_type == "8" : # noop
    pass
return decoded_msg


signature:
def add_acl_method ( self , method_name ) :
    """ACL system : make the method _ name accessible to the current socket"""

implementation:
if isinstance ( self . allowed_methods , set ) :
    self . allowed_methods . add ( method_name )
else :
    self . allowed_methods = set ( [ method_name ] )


signature:
def del_acl_method ( self , method_name ) :
    """ACL system : ensure the user will not have access to that method ."""

implementation:
if self . allowed_methods is None :
    raise ValueError ( "Trying to delete an ACL method, but none were" + " defined yet! Or: No ACL restrictions yet, why would you" + " delete one?" )
self . allowed_methods . remove ( method_name )


signature:
def process_packet ( self , packet ) :
    """If you override this , NONE of the functions in this class
 will be called . It is responsible for dispatching to
 : meth : ` process _ event ` ( which in turn calls ` ` on _ * ( ) ` ` and
 ` ` recv _ * ( ) ` ` methods ) .
 If the packet arrived here , it is because it belongs to this endpoint .
 For each packet arriving , the only possible path of execution , that is ,
 the only methods that * can * be called are the following :
 * recv _ connect ( )
 * recv _ message ( )
 * recv _ json ( )
 * recv _ error ( )
 * recv _ disconnect ( )
 * on _ * ( )"""

implementation:
packet_type = packet [ 'type' ]
if packet_type == 'event' :
    return self . process_event ( packet )
elif packet_type == 'message' :
    return self . call_method_with_acl ( 'recv_message' , packet , packet [ 'data' ] )
elif packet_type == 'json' :
    return self . call_method_with_acl ( 'recv_json' , packet , packet [ 'data' ] )
elif packet_type == 'connect' :
    self . socket . send_packet ( packet )
    return self . call_method_with_acl ( 'recv_connect' , packet )
elif packet_type == 'error' :
    return self . call_method_with_acl ( 'recv_error' , packet )
elif packet_type == 'ack' :
    callback = self . socket . _pop_ack_callback ( packet [ 'ackId' ] )
    if not callback :
        print ( "ERROR: No such callback for ackId %s" % packet [ 'ackId' ] )
        return
    return callback ( * ( packet [ 'args' ] ) )
elif packet_type == 'disconnect' : # Force a disconnect on the namespace .
    return self . call_method_with_acl ( 'recv_disconnect' , packet )
else :
    print ( "Unprocessed packet" , packet )


signature:
def process_event ( self , packet ) :
    """This function dispatches ` ` event ` ` messages to the correct
 functions . You should override this method only if you are not
 satisfied with the automatic dispatching to
 ` ` on _ ` ` - prefixed methods . You could then implement your own dispatch .
 See the source code for inspiration .
 There are two ways to deal with callbacks from the client side
 ( meaning , the browser has a callback waiting for data that this
 server will be sending back ) :
 The first one is simply to return an object . If the incoming
 packet requested has an ' ack ' field set , meaning the browser is
 waiting for callback data , it will automatically be packaged
 and sent , associated with the ' ackId ' from the browser . The
 return value must be a * sequence * of elements , that will be
 mapped to the positional parameters of the callback function
 on the browser side .
 If you want to * know * that you ' re dealing with a packet
 that requires a return value , you can do those things manually
 by inspecting the ` ` ack ` ` and ` ` id ` ` keys from the ` ` packet ` `
 object . Your callback will behave specially if the name of
 the argument to your method is ` ` packet ` ` . It will fill it
 with the unprocessed ` ` packet ` ` object for your inspection ,
 like this :
 . . code - block : : python
 def on _ my _ callback ( self , packet ) :
 if ' ack ' in packet :
 self . emit ( ' go _ back ' , ' param1 ' , id = packet [ ' id ' ] )"""

implementation:
args = packet [ 'args' ]
name = packet [ 'name' ]
if not allowed_event_name_regex . match ( name ) :
    self . error ( "unallowed_event_name" , "name must only contains alpha numerical characters" )
    return
method_name = 'on_' + name . replace ( ' ' , '_' )
# This means the args , passed as a list , will be expanded to
# the method arg and if you passed a dict , it will be a dict
# as the first parameter .
return self . call_method_with_acl ( method_name , packet , * args )


signature:
def call_method_with_acl ( self , method_name , packet , * args ) :
    """You should always use this function to call the methods ,
 as it checks if the user is allowed according to the ACLs .
 If you override : meth : ` process _ packet ` or
 : meth : ` process _ event ` , you should definitely want to use this
 instead of ` ` getattr ( self , ' my _ method ' ) ( ) ` `"""

implementation:
if not self . is_method_allowed ( method_name ) :
    self . error ( 'method_access_denied' , 'You do not have access to method "%s"' % method_name )
    return
return self . call_method ( method_name , packet , * args )


signature:
def call_method ( self , method_name , packet , * args ) :
    """This function is used to implement the two behaviors on dispatched
 ` ` on _ * ( ) ` ` and ` ` recv _ * ( ) ` ` method calls .
 Those are the two behaviors :
 * If there is only one parameter on the dispatched method and
 it is named ` ` packet ` ` , then pass in the packet dict as the
 sole parameter .
 * Otherwise , pass in the arguments as specified by the
 different ` ` recv _ * ( ) ` ` methods args specs , or the
 : meth : ` process _ event ` documentation .
 This method will also consider the
 ` ` exception _ handler _ decorator ` ` . See Namespace documentation
 for details and examples ."""

implementation:
method = getattr ( self , method_name , None )
if method is None :
    self . error ( 'no_such_method' , 'The method "%s" was not found' % method_name )
    return
specs = inspect . getargspec ( method )
func_args = specs . args
if not len ( func_args ) or func_args [ 0 ] != 'self' :
    self . error ( "invalid_method_args" , "The server-side method is invalid, as it doesn't " "have 'self' as its first argument" )
    return
# Check if we need to decorate to handle exceptions
if hasattr ( self , 'exception_handler_decorator' ) :
    method = self . exception_handler_decorator ( method )
if len ( func_args ) == 2 and func_args [ 1 ] == 'packet' :
    return method ( packet )
else :
    return method ( * args )


signature:
def error ( self , error_name , error_message , msg_id = None , quiet = False ) :
    """Use this to use the configured ` ` error _ handler ` ` yield an
 error message to your application .
 : param error _ name : is a short string , to associate messages to recovery
 methods
 : param error _ message : is some human - readable text , describing the error
 : param msg _ id : is used to associate with a request
 : param quiet : specific to error _ handlers . The default doesn ' t send a
 message to the user , but shows a debug message on the
 developer console ."""

implementation: self . socket . error ( error_name , error_message , endpoint = self . ns_name , msg_id = msg_id , quiet = quiet )


signature:
def send ( self , message , json = False , callback = None ) :
    """Use send to send a simple string message .
 If ` ` json ` ` is True , the message will be encoded as a JSON object
 on the wire , and decoded on the other side .
 This is mostly for backwards compatibility . ` ` emit ( ) ` ` is more fun .
 : param callback : This is a callback function that will be
 called automatically by the client upon
 reception . It does not verify that the
 listener over there was completed with
 success . It just tells you that the browser
 got a hold of the packet .
 : type callback : callable"""

implementation:
pkt = dict ( type = "message" , data = message , endpoint = self . ns_name )
if json :
    pkt [ 'type' ] = "json"
if callback : # By passing ack = True , we use the old behavior of being returned
# an ' ack ' packet , automatically triggered by the client - side
# with no user - code being run . The emit ( ) version of the
# callback is more useful I think : ) So migrate your code .
    pkt [ 'ack' ] = True
    pkt [ 'id' ] = msgid = self . socket . _get_next_msgid ( )
    self . socket . _save_ack_callback ( msgid , callback )
self . socket . send_packet ( pkt )


signature:
def emit ( self , event , * args , ** kwargs ) :
    """Use this to send a structured event , with a name and arguments , to
 the client .
 By default , it uses this namespace ' s endpoint . You can send messages on
 other endpoints with something like :
 ` ` self . socket [ ' / other _ endpoint ' ] . emit ( ) ` ` .
 However , it is possible that the ` ` ' / other _ endpoint ' ` ` was not
 initialized yet , and that would yield a ` ` KeyError ` ` .
 The only supported ` ` kwargs ` ` is ` ` callback ` ` . All other parameters
 must be passed positionally .
 : param event : The name of the event to trigger on the other end .
 : param callback : Pass in the callback keyword argument to define a
 call - back that will be called when the client acks .
 This callback is slightly different from the one from
 ` ` send ( ) ` ` , as this callback will receive parameters
 from the explicit call of the ` ` ack ( ) ` ` function
 passed to the listener on the client side .
 The remote listener will need to explicitly ack ( by
 calling its last argument , a function which is
 usually called ' ack ' ) with some parameters indicating
 success or error . The ' ack ' packet coming back here
 will then trigger the callback function with the
 returned values .
 : type callback : callable"""

implementation:
callback = kwargs . pop ( 'callback' , None )
if kwargs :
    raise ValueError ( "emit() only supports positional argument, to stay " "compatible with the Socket.IO protocol. You can " "however pass in a dictionary as the first argument" )
pkt = dict ( type = "event" , name = event , args = args , endpoint = self . ns_name )
if callback : # By passing ' data ' , we indicate that we * want * an explicit ack
# by the client code , not an automatic as with send ( ) .
    pkt [ 'ack' ] = 'data'
    pkt [ 'id' ] = msgid = self . socket . _get_next_msgid ( )
    self . socket . _save_ack_callback ( msgid , callback )
self . socket . send_packet ( pkt )


signature:
def spawn ( self , fn , * args , ** kwargs ) :
    """Spawn a new process , attached to this Namespace .
 It will be monitored by the " watcher " process in the Socket . If the
 socket disconnects , all these greenlets are going to be killed , after
 calling BaseNamespace . disconnect ( )
 This method uses the ` ` exception _ handler _ decorator ` ` . See
 Namespace documentation for more information ."""

implementation:
# self . log . debug ( " Spawning sub - Namespace Greenlet : % s " % fn . _ _ name _ _ )
if hasattr ( self , 'exception_handler_decorator' ) :
    fn = self . exception_handler_decorator ( fn )
new = gevent . spawn ( fn , * args , ** kwargs )
self . jobs . append ( new )
return new


signature:
def disconnect ( self , silent = False ) :
    """Send a ' disconnect ' packet , so that the user knows it has been
 disconnected ( booted actually ) . This will trigger an onDisconnect ( )
 call on the client side .
 Over here , we will kill all ` ` spawn ` ` ed processes and remove the
 namespace from the Socket object .
 : param silent : do not actually send the packet ( if they asked for a
 disconnect for example ) , but just kill all jobs spawned
 by this Namespace , and remove it from the Socket ."""

implementation:
if not silent :
    packet = { "type" : "disconnect" , "endpoint" : self . ns_name }
    self . socket . send_packet ( packet )
# remove _ namespace might throw GreenletExit so
# kill _ local _ jobs must be in finally
try :
    self . socket . remove_namespace ( self . ns_name )
finally :
    self . kill_local_jobs ( )


signature:
def get_socket ( self , sessid = '' ) :
    """Return an existing or new client Socket ."""

implementation:
socket = self . sockets . get ( sessid )
if sessid and not socket :
    return None
    # you ask for a session that doesn ' t exist !
if socket is None :
    socket = Socket ( self , self . config )
    self . sockets [ socket . sessid ] = socket
else :
    socket . incr_hits ( )
return socket


signature:
def create ( ) :
    """Handles post from the " Add room " form on the homepage , and
 redirects to the new room ."""

implementation:
name = request . form . get ( "name" )
if name :
    room , created = get_or_create ( ChatRoom , name = name )
    return redirect ( url_for ( 'room' , slug = room . slug ) )
return redirect ( url_for ( 'rooms' ) )


signature:
def autodiscover ( ) :
    """Auto - discover INSTALLED _ APPS sockets . py modules and fail silently when
 not present . NOTE : socketio _ autodiscover was inspired / copied from
 django . contrib . admin autodiscover"""

implementation:
global LOADING_SOCKETIO
if LOADING_SOCKETIO :
    return
LOADING_SOCKETIO = True
import imp
from django . conf import settings
for app in settings . INSTALLED_APPS :
    try :
        app_path = import_module ( app ) . __path__
    except AttributeError :
        continue
    try :
        imp . find_module ( 'sockets' , app_path )
    except ImportError :
        continue
    import_module ( "%s.sockets" % app )
LOADING_SOCKETIO = False


signature:
def handle_one_response ( self ) :
    """This function deals with * ONE INCOMING REQUEST * from the web .
 It will wire and exchange message to the queues for long - polling
 methods , otherwise , will stay alive for websockets ."""

implementation:
path = self . environ . get ( 'PATH_INFO' )
# Kick non - socket . io requests to our superclass
if not path . lstrip ( '/' ) . startswith ( self . server . resource + '/' ) :
    return super ( SocketIOHandler , self ) . handle_one_response ( )
self . status = None
self . headers_sent = False
self . result = None
self . response_length = 0
self . response_use_chunked = False
# This is analyzed for each and every HTTP requests involved
# in the Socket . IO protocol , whether long - running or long - polling
# ( read : websocket or xhr - polling methods )
request_method = self . environ . get ( "REQUEST_METHOD" )
request_tokens = self . RE_REQUEST_URL . match ( path )
handshake_tokens = self . RE_HANDSHAKE_URL . match ( path )
disconnect_tokens = self . RE_DISCONNECT_URL . match ( path )
if handshake_tokens : # Deal with first handshake here , create the Socket and push
# the config up .
    return self . _do_handshake ( handshake_tokens . groupdict ( ) )
elif disconnect_tokens : # it ' s a disconnect request via XHR
    tokens = disconnect_tokens . groupdict ( )
elif request_tokens :
    tokens = request_tokens . groupdict ( )
    # and continue . . .
else : # This is no socket . io request . Let the WSGI app handle it .
    return super ( SocketIOHandler , self ) . handle_one_response ( )
# Setup socket
sessid = tokens [ "sessid" ]
socket = self . server . get_socket ( sessid )
if not socket :
    self . handle_bad_request ( )
    return [ ]
# Do not say the session is not found , just bad request
# so they don ' t start brute forcing to find open sessions
if self . environ [ 'QUERY_STRING' ] . startswith ( 'disconnect' ) : # according to socket . io specs disconnect requests
# have a ` disconnect ` query string
# https : / / github . com / LearnBoost / socket . io - spec # forced - socket - disconnection
    socket . disconnect ( )
    self . handle_disconnect_request ( )
    return [ ]
# Setup transport
transport = self . handler_types . get ( tokens [ "transport_id" ] )
# In case this is WebSocket request , switch to the WebSocketHandler
# FIXME : fix this ugly class change
old_class = None
if issubclass ( transport , ( transports . WebsocketTransport , transports . FlashSocketTransport ) ) :
    old_class = self . __class__
    self . __class__ = self . server . ws_handler_class
    self . prevent_wsgi_call = True
    # thank you
    # TODO : any errors , treat them ? ?
    self . handle_one_response ( )
# does the Websocket dance before we continue
# Make the socket object available for WSGI apps
self . environ [ 'socketio' ] = socket
# Create a transport and handle the request likewise
self . transport = transport ( self , self . config )
# transports register their own spawn ' d jobs now
self . transport . do_exchange ( socket , request_method )
if not socket . connection_established : # This is executed only on the * first * packet of the establishment
# of the virtual Socket connection .
    socket . connection_established = True
    socket . state = socket . STATE_CONNECTED
    socket . _spawn_heartbeat ( )
    socket . _spawn_watcher ( )
    try : # We ' ll run the WSGI app if it wasn ' t already done .
        if socket . wsgi_app_greenlet is None : # TODO : why don ' t we spawn a call to handle _ one _ response here ?
        # why call directly the WSGI machinery ?
            start_response = lambda status , headers , exc = None : None
            socket . wsgi_app_greenlet = gevent . spawn ( self . application , self . environ , start_response )
    except :
        self . handle_error ( * sys . exc_info ( ) )
# we need to keep the connection open if we are an open socket
if tokens [ 'transport_id' ] in [ 'flashsocket' , 'websocket' ] : # wait here for all jobs to finished , when they are done
    gevent . joinall ( socket . jobs )
# Switch back to the old class so references to this don ' t use the
# incorrect class . Useful for debugging .
if old_class :
    self . __class__ = old_class
# Clean up circular references so they can be garbage collected .
if hasattr ( self , 'websocket' ) and self . websocket :
    if hasattr ( self . websocket , 'environ' ) :
        del self . websocket . environ
    del self . websocket
if self . environ :
    del self . environ


signature:
def get_messages_payload ( self , socket , timeout = None ) :
    """This will fetch the messages from the Socket ' s queue , and if
 there are many messes , pack multiple messages in one payload and return"""

implementation:
try :
    msgs = socket . get_multiple_client_msgs ( timeout = timeout )
    data = self . encode_payload ( msgs )
except Empty :
    data = ""
return data


signature:
def encode_payload ( self , messages ) :
    """Encode list of messages . Expects messages to be unicode .
 ` ` messages ` ` - List of raw messages to encode , if necessary"""

implementation:
if not messages or messages [ 0 ] is None :
    return ''
if len ( messages ) == 1 :
    return messages [ 0 ] . encode ( 'utf-8' )
payload = u'' . join ( [ ( u'\ufffd%d\ufffd%s' % ( len ( p ) , p ) ) for p in messages if p is not None ] )
# FIXME : why is it so that we must filter None from here ? How
# is it even possible that a None gets in there ?
return payload . encode ( 'utf-8' )


signature:
def decode_payload ( self , payload ) :
    """This function can extract multiple messages from one HTTP payload .
 Some times , the XHR / JSONP / . . transports can pack more than one message
 on a single packet . They are encoding following the WebSocket
 semantics , which need to be reproduced here to unwrap the messages .
 The semantics are :
 \ufffd + [ length as a string ] + \ufffd + [ payload as a unicode string ]
 This function returns a list of messages , even though there is only
 one .
 Inspired by socket . io / lib / transports / http . js"""

implementation:
payload = payload . decode ( 'utf-8' )
if payload [ 0 ] == u"\ufffd" :
    ret = [ ]
    while len ( payload ) != 0 :
        len_end = payload . find ( u"\ufffd" , 1 )
        length = int ( payload [ 1 : len_end ] )
        msg_start = len_end + 1
        msg_end = length + msg_start
        message = payload [ msg_start : msg_end ]
        ret . append ( message )
        payload = payload [ msg_end : ]
    return ret
return [ payload ]