signature:
def values ( ns_key ) :
    '''Return the allowed values for an enumerated namespace .
 Parameters
 ns _ key : str
 Namespace key identifier
 Returns
 values : list
 Raises
 NamespaceError
 If ` ns _ key ` is not found , or does not have enumerated values
 Examples
 > > > jams . schema . values ( ' tag _ gtzan ' )
 [ ' blues ' , ' classical ' , ' country ' , ' disco ' , ' hip - hop ' , ' jazz ' ,
 ' metal ' , ' pop ' , ' reggae ' , ' rock ' ]'''

implementation:
if ns_key not in __NAMESPACE__ :
    raise NamespaceError ( 'Unknown namespace: {:s}' . format ( ns_key ) )
if 'enum' not in __NAMESPACE__ [ ns_key ] [ 'value' ] :
    raise NamespaceError ( 'Namespace {:s} is not enumerated' . format ( ns_key ) )
return copy . copy ( __NAMESPACE__ [ ns_key ] [ 'value' ] [ 'enum' ] )


signature:
def get_dtypes ( ns_key ) :
    '''Get the dtypes associated with the value and confidence fields
 for a given namespace .
 Parameters
 ns _ key : str
 The namespace key in question
 Returns
 value _ dtype , confidence _ dtype : numpy . dtype
 Type identifiers for value and confidence fields .'''

implementation:
# First , get the schema
if ns_key not in __NAMESPACE__ :
    raise NamespaceError ( 'Unknown namespace: {:s}' . format ( ns_key ) )
value_dtype = __get_dtype ( __NAMESPACE__ [ ns_key ] . get ( 'value' , { } ) )
confidence_dtype = __get_dtype ( __NAMESPACE__ [ ns_key ] . get ( 'confidence' , { } ) )
return value_dtype , confidence_dtype


signature:
def list_namespaces ( ) :
    '''Print out a listing of available namespaces'''

implementation:
print ( '{:30s}\t{:40s}' . format ( 'NAME' , 'DESCRIPTION' ) )
print ( '-' * 78 )
for sch in sorted ( __NAMESPACE__ ) :
    desc = __NAMESPACE__ [ sch ] [ 'description' ]
    desc = ( desc [ : 44 ] + '..' ) if len ( desc ) > 46 else desc
    print ( '{:30s}\t{:40s}' . format ( sch , desc ) )


signature:
def __get_dtype ( typespec ) :
    '''Get the dtype associated with a jsonschema type definition
 Parameters
 typespec : dict
 The schema definition
 Returns
 dtype : numpy . dtype
 The associated dtype'''

implementation:
if 'type' in typespec :
    return __TYPE_MAP__ . get ( typespec [ 'type' ] , np . object_ )
elif 'enum' in typespec : # Enums map to objects
    return np . object_
elif 'oneOf' in typespec : # Recurse
    types = [ __get_dtype ( v ) for v in typespec [ 'oneOf' ] ]
    # If they ' re not all equal , return object
    if all ( [ t == types [ 0 ] for t in types ] ) :
        return types [ 0 ]
return np . object_


signature:
def __load_jams_schema ( ) :
    '''Load the schema file from the package .'''

implementation:
schema_file = os . path . join ( SCHEMA_DIR , 'jams_schema.json' )
jams_schema = None
with open ( resource_filename ( __name__ , schema_file ) , mode = 'r' ) as fdesc :
    jams_schema = json . load ( fdesc )
if jams_schema is None :
    raise JamsError ( 'Unable to load JAMS schema' )
return jams_schema


signature:
def import_lab ( namespace , filename , infer_duration = True , ** parse_options ) :
    r'''Load a . lab file as an Annotation object .
 . lab files are assumed to have the following format :
 ` ` TIME _ START \ tTIME _ END \ tANNOTATION ` `
 By default , . lab files are assumed to have columns separated by one
 or more white - space characters , and have no header or index column
 information .
 If the . lab file contains only two columns , then an empty duration
 field is inferred .
 If the . lab file contains more than three columns , each row ' s
 annotation value is assigned the contents of last non - empty column .
 Parameters
 namespace : str
 The namespace for the new annotation
 filename : str
 Path to the . lab file
 infer _ duration : bool
 If ` True ` , interval durations are inferred from ` ( start , end ) ` columns ,
 or difference between successive times .
 If ` False ` , interval durations are assumed to be explicitly coded as
 ` ( start , duration ) ` columns . If only one time column is given , then
 durations are set to 0.
 For instantaneous event annotations ( e . g . , beats or onsets ) , this
 should be set to ` False ` .
 parse _ options : additional keyword arguments
 Passed to ` ` pandas . DataFrame . read _ csv ` `
 Returns
 annotation : Annotation
 The newly constructed annotation object
 See Also
 pandas . DataFrame . read _ csv'''

implementation:
# Create a new annotation object
annotation = core . Annotation ( namespace )
parse_options . setdefault ( 'sep' , r'\s+' )
parse_options . setdefault ( 'engine' , 'python' )
parse_options . setdefault ( 'header' , None )
parse_options . setdefault ( 'index_col' , False )
# This is a hack to handle potentially ragged . lab data
parse_options . setdefault ( 'names' , range ( 20 ) )
data = pd . read_csv ( filename , ** parse_options )
# Drop all - nan columns
data = data . dropna ( how = 'all' , axis = 1 )
# Do we need to add a duration column ?
# This only applies to event annotations
if len ( data . columns ) == 2 : # Insert a column of zeros after the timing
    data . insert ( 1 , 'duration' , 0 )
    if infer_duration :
        data [ 'duration' ] [ : - 1 ] = data . loc [ : , 0 ] . diff ( ) [ 1 : ] . values
else : # Convert from time to duration
    if infer_duration :
        data . loc [ : , 1 ] -= data [ 0 ]
for row in data . itertuples ( ) :
    time , duration = row [ 1 : 3 ]
    value = [ x for x in row [ 3 : ] if x is not None ] [ - 1 ]
    annotation . append ( time = time , duration = duration , confidence = 1.0 , value = value )
return annotation


signature:
def expand_filepaths ( base_dir , rel_paths ) :
    """Expand a list of relative paths to a give base directory .
 Parameters
 base _ dir : str
 The target base directory
 rel _ paths : list ( or list - like )
 Collection of relative path strings
 Returns
 expanded _ paths : list
 ` rel _ paths ` rooted at ` base _ dir `
 Examples
 > > > jams . util . expand _ filepaths ( ' / data ' , [ ' audio ' , ' beat ' , ' seglab ' ] )
 [ ' / data / audio ' , ' / data / beat ' , ' / data / seglab ' ]"""

implementation: return [ os . path . join ( base_dir , os . path . normpath ( rp ) ) for rp in rel_paths ]


signature:
def smkdirs ( dpath , mode = 0o777 ) :
    """Safely make a full directory path if it doesn ' t exist .
 Parameters
 dpath : str
 Path of directory / directories to create
 mode : int [ default = 0777]
 Permissions for the new directories
 See also
 os . makedirs"""

implementation:
if not os . path . exists ( dpath ) :
    os . makedirs ( dpath , mode = mode )


signature:
def find_with_extension ( in_dir , ext , depth = 3 , sort = True ) :
    """Naive depth - search into a directory for files with a given extension .
 Parameters
 in _ dir : str
 Path to search .
 ext : str
 File extension to match .
 depth : int
 Depth of directories to search .
 sort : bool
 Sort the list alphabetically
 Returns
 matched : list
 Collection of matching file paths .
 Examples
 > > > jams . util . find _ with _ extension ( ' Audio ' , ' wav ' )
 [ ' Audio / LizNelson _ Rainfall / LizNelson _ Rainfall _ MIX . wav ' ,
 ' Audio / LizNelson _ Rainfall / LizNelson _ Rainfall _ RAW / LizNelson _ Rainfall _ RAW _ 01_01 . wav ' ,
 ' Audio / LizNelson _ Rainfall / LizNelson _ Rainfall _ RAW / LizNelson _ Rainfall _ RAW _ 02_01 . wav ' ,
 ' Audio / Phoenix _ ScotchMorris / Phoenix _ ScotchMorris _ STEMS / Phoenix _ ScotchMorris _ STEM _ 02 . wav ' ,
 ' Audio / Phoenix _ ScotchMorris / Phoenix _ ScotchMorris _ STEMS / Phoenix _ ScotchMorris _ STEM _ 03 . wav ' ,
 ' Audio / Phoenix _ ScotchMorris / Phoenix _ ScotchMorris _ STEMS / Phoenix _ ScotchMorris _ STEM _ 04 . wav ' ]"""

implementation:
assert depth >= 1
ext = ext . strip ( os . extsep )
match = list ( )
for n in range ( 1 , depth + 1 ) :
    wildcard = os . path . sep . join ( [ "*" ] * n )
    search_path = os . path . join ( in_dir , os . extsep . join ( [ wildcard , ext ] ) )
    match += glob . glob ( search_path )
if sort :
    match . sort ( )
return match


signature:
def get_comments ( jam , ann ) :
    '''Get the metadata from a jam and an annotation , combined as a string .
 Parameters
 jam : JAMS
 The jams object
 ann : Annotation
 An annotation object
 Returns
 comments : str
 The jam . file _ metadata and ann . annotation _ metadata , combined and serialized'''

implementation:
jam_comments = jam . file_metadata . __json__
ann_comments = ann . annotation_metadata . __json__
return json . dumps ( { 'metadata' : jam_comments , 'annotation metadata' : ann_comments } , indent = 2 )


signature:
def lab_dump ( ann , comment , filename , sep , comment_char ) :
    '''Save an annotation as a lab / csv .
 Parameters
 ann : Annotation
 The annotation object
 comment : str
 The comment string header
 filename : str
 The output filename
 sep : str
 The separator string for output
 comment _ char : str
 The character used to denote comments'''

implementation:
intervals , values = ann . to_interval_values ( )
frame = pd . DataFrame ( columns = [ 'Time' , 'End Time' , 'Label' ] , data = { 'Time' : intervals [ : , 0 ] , 'End Time' : intervals [ : , 1 ] , 'Label' : values } )
with open ( filename , 'w' ) as fdesc :
    for line in comment . split ( '\n' ) :
        fdesc . write ( '{:s}  {:s}\n' . format ( comment_char , line ) )
    frame . to_csv ( path_or_buf = fdesc , index = False , sep = sep )


signature:
def convert_jams ( jams_file , output_prefix , csv = False , comment_char = '#' , namespaces = None ) :
    '''Convert jams to labs .
 Parameters
 jams _ file : str
 The path on disk to the jams file in question
 output _ prefix : str
 The file path prefix of the outputs
 csv : bool
 Whether to output in csv ( True ) or lab ( False ) format
 comment _ char : str
 The character used to denote comments
 namespaces : list - like
 The set of namespace patterns to match for output'''

implementation:
if namespaces is None :
    raise ValueError ( 'No namespaces provided. Try ".*" for all namespaces.' )
jam = jams . load ( jams_file )
# Get all the annotations
# Filter down to the unique ones
# For each annotation
# generate the comment string
# generate the output filename
# dump to csv
# Make a counter object for each namespace type
counter = collections . Counter ( )
annotations = [ ]
for query in namespaces :
    annotations . extend ( jam . search ( namespace = query ) )
if csv :
    suffix = 'csv'
    sep = ','
else :
    suffix = 'lab'
    sep = '\t'
for ann in annotations :
    index = counter [ ann . namespace ]
    counter [ ann . namespace ] += 1
    filename = os . path . extsep . join ( [ get_output_name ( output_prefix , ann . namespace , index ) , suffix ] )
    comment = get_comments ( jam , ann )
    # Dump to disk
    lab_dump ( ann , comment , filename , sep , comment_char )


signature:
def parse_arguments ( args ) :
    '''Parse arguments from the command line'''

implementation:
parser = argparse . ArgumentParser ( description = 'Convert JAMS to .lab files' )
parser . add_argument ( '-c' , '--comma-separated' , dest = 'csv' , action = 'store_true' , default = False , help = 'Output in .csv instead of .lab' )
parser . add_argument ( '--comment' , dest = 'comment_char' , type = str , default = '#' , help = 'Comment character' )
parser . add_argument ( '-n' , '--namespace' , dest = 'namespaces' , nargs = '+' , default = [ '.*' ] , help = 'One or more namespaces to output.  Default is all.' )
parser . add_argument ( 'jams_file' , help = 'Path to the input jams file' )
parser . add_argument ( 'output_prefix' , help = 'Prefix for output files' )
return vars ( parser . parse_args ( args ) )


signature:
def convert ( annotation , target_namespace ) :
    '''Convert a given annotation to the target namespace .
 Parameters
 annotation : jams . Annotation
 An annotation object
 target _ namespace : str
 The target namespace
 Returns
 mapped _ annotation : jams . Annotation
 if ` annotation ` already belongs to ` target _ namespace ` , then
 it is returned directly .
 otherwise , ` annotation ` is copied and automatically converted
 to the target namespace .
 Raises
 SchemaError
 if the input annotation fails to validate
 NamespaceError
 if no conversion is possible
 Examples
 Convert frequency measurements in Hz to MIDI
 > > > ann _ midi = jams . convert ( ann _ hz , ' note _ midi ' )
 And back to Hz
 > > > ann _ hz2 = jams . convert ( ann _ midi , ' note _ hz ' )'''

implementation:
# First , validate the input . If this fails , we can ' t auto - convert .
annotation . validate ( strict = True )
# If we ' re already in the target namespace , do nothing
if annotation . namespace == target_namespace :
    return annotation
if target_namespace in __CONVERSION__ : # Otherwise , make a copy to mangle
    annotation = deepcopy ( annotation )
    # Look for a way to map this namespace to the target
    for source in __CONVERSION__ [ target_namespace ] :
        if annotation . search ( namespace = source ) :
            return __CONVERSION__ [ target_namespace ] [ source ] ( annotation )
# No conversion possible
raise NamespaceError ( 'Unable to convert annotation from namespace=' '"{0}" to "{1}"' . format ( annotation . namespace , target_namespace ) )


signature:
def can_convert ( annotation , target_namespace ) :
    '''Test if an annotation can be mapped to a target namespace
 Parameters
 annotation : jams . Annotation
 An annotation object
 target _ namespace : str
 The target namespace
 Returns
 True
 if ` annotation ` can be automatically converted to
 ` target _ namespace `
 False
 otherwise'''

implementation:
# If we ' re already in the target namespace , do nothing
if annotation . namespace == target_namespace :
    return True
if target_namespace in __CONVERSION__ : # Look for a way to map this namespace to the target
    for source in __CONVERSION__ [ target_namespace ] :
        if annotation . search ( namespace = source ) :
            return True
return False


signature:
def pitch_hz_to_contour ( annotation ) :
    '''Convert a pitch _ hz annotation to a contour'''

implementation:
annotation . namespace = 'pitch_contour'
data = annotation . pop_data ( )
for obs in data :
    annotation . append ( time = obs . time , duration = obs . duration , confidence = obs . confidence , value = dict ( index = 0 , frequency = np . abs ( obs . value ) , voiced = obs . value > 0 ) )
return annotation


signature:
def note_hz_to_midi ( annotation ) :
    '''Convert a pitch _ hz annotation to pitch _ midi'''

implementation:
annotation . namespace = 'note_midi'
data = annotation . pop_data ( )
for obs in data :
    annotation . append ( time = obs . time , duration = obs . duration , confidence = obs . confidence , value = 12 * ( np . log2 ( obs . value ) - np . log2 ( 440.0 ) ) + 69 )
return annotation


signature:
def scaper_to_tag ( annotation ) :
    '''Convert scaper annotations to tag _ open'''

implementation:
annotation . namespace = 'tag_open'
data = annotation . pop_data ( )
for obs in data :
    annotation . append ( time = obs . time , duration = obs . duration , confidence = obs . confidence , value = obs . value [ 'label' ] )
return annotation


signature:
def _open ( name_or_fdesc , mode = 'r' , fmt = 'auto' ) :
    '''An intelligent wrapper for ` ` open ` ` .
 Parameters
 name _ or _ fdesc : string - type or open file descriptor
 If a string type , refers to the path to a file on disk .
 If an open file descriptor , it is returned as - is .
 mode : string
 The mode with which to open the file .
 See ` ` open ` ` for details .
 fmt : string [ ' auto ' , ' jams ' , ' json ' , ' jamz ' ]
 The encoding for the input / output stream .
 If ` auto ` , the format is inferred from the filename extension .
 Otherwise , use the specified coding .
 See Also
 open
 gzip . open'''

implementation:
open_map = { 'jams' : open , 'json' : open , 'jamz' : gzip . open , 'gz' : gzip . open }
# If we ' ve been given an open descriptor , do the right thing
if hasattr ( name_or_fdesc , 'read' ) or hasattr ( name_or_fdesc , 'write' ) :
    yield name_or_fdesc
elif isinstance ( name_or_fdesc , six . string_types ) : # Infer the opener from the extension
    if fmt == 'auto' :
        _ , ext = os . path . splitext ( name_or_fdesc )
        # Pull off the extension separator
        ext = ext [ 1 : ]
    else :
        ext = fmt
    try :
        ext = ext . lower ( )
        # Force text mode if we ' re using gzip
        if ext in [ 'jamz' , 'gz' ] and 't' not in mode :
            mode = '{:s}t' . format ( mode )
        with open_map [ ext ] ( name_or_fdesc , mode = mode ) as fdesc :
            yield fdesc
    except KeyError :
        raise ParameterError ( 'Unknown JAMS extension ' 'format: "{:s}"' . format ( ext ) )
else : # Don ' t know how to handle this . Raise a parameter error
    raise ParameterError ( 'Invalid filename or ' 'descriptor: {}' . format ( name_or_fdesc ) )


signature:
def load ( path_or_file , validate = True , strict = True , fmt = 'auto' ) :
    r"""Load a JAMS Annotation from a file .
 Parameters
 path _ or _ file : str or file - like
 Path to the JAMS file to load
 OR
 An open file handle to load from .
 validate : bool
 Attempt to validate the JAMS object
 strict : bool
 if ` validate = = True ` , enforce strict schema validation
 fmt : str [ ' auto ' , ' jams ' , ' jamz ' ]
 The encoding format of the input
 If ` auto ` , encoding is inferred from the file name .
 If the input is an open file handle , ` jams ` encoding
 is used .
 Returns
 jam : JAMS
 The loaded JAMS object
 Raises
 SchemaError
 if ` validate = = True ` , ` strict = = True ` , and validation fails
 See also
 JAMS . validate
 JAMS . save
 Examples
 > > > # Load a jams object from a file name
 > > > J = jams . load ( ' data . jams ' )
 > > > # Or from an open file descriptor
 > > > with open ( ' data . jams ' , ' r ' ) as fdesc :
 . . . J = jams . load ( fdesc )
 > > > # Non - strict validation
 > > > J = jams . load ( ' data . jams ' , strict = False )
 > > > # No validation at all
 > > > J = jams . load ( ' data . jams ' , validate = False )"""

implementation:
with _open ( path_or_file , mode = 'r' , fmt = fmt ) as fdesc :
    jam = JAMS ( ** json . load ( fdesc ) )
if validate :
    jam . validate ( strict = strict )
return jam


signature:
def query_pop ( query , prefix , sep = '.' ) :
    '''Pop a prefix from a query string .
 Parameters
 query : str
 The query string
 prefix : str
 The prefix string to pop , if it exists
 sep : str
 The string to separate fields
 Returns
 popped : str
 ` query ` with a ` prefix ` removed from the front ( if found )
 or ` query ` if the prefix was not found
 Examples
 > > > query _ pop ( ' Annotation . namespace ' , ' Annotation ' )
 ' namespace '
 > > > query _ pop ( ' namespace ' , ' Annotation ' )
 ' namespace ' '''

implementation:
terms = query . split ( sep )
if terms [ 0 ] == prefix :
    terms = terms [ 1 : ]
return sep . join ( terms )


signature:
def match_query ( string , query ) :
    '''Test if a string matches a query .
 Parameters
 string : str
 The string to test
 query : string , callable , or object
 Either a regular expression , callable function , or object .
 Returns
 match : bool
 ` True ` if :
 - ` query ` is a callable and ` query ( string ) = = True `
 - ` query ` is a regular expression and ` re . match ( query , string ) `
 - or ` string = = query ` for any other query
 ` False ` otherwise'''

implementation:
if six . callable ( query ) :
    return query ( string )
elif ( isinstance ( query , six . string_types ) and isinstance ( string , six . string_types ) ) :
    return re . match ( query , string ) is not None
else :
    return query == string


signature:
def serialize_obj ( obj ) :
    '''Custom serialization functionality for working with advanced data types .
 - numpy arrays are converted to lists
 - lists are recursively serialized element - wise'''

implementation:
if isinstance ( obj , np . integer ) :
    return int ( obj )
elif isinstance ( obj , np . floating ) :
    return float ( obj )
elif isinstance ( obj , np . ndarray ) :
    return obj . tolist ( )
elif isinstance ( obj , list ) :
    return [ serialize_obj ( x ) for x in obj ]
elif isinstance ( obj , Observation ) :
    return { k : serialize_obj ( v ) for k , v in six . iteritems ( obj . _asdict ( ) ) }
return obj


signature:
def summary ( obj , indent = 0 ) :
    '''Helper function to format repr strings for JObjects and friends .
 Parameters
 obj
 The object to repr
 indent : int > = 0
 indent each new line by ` indent ` spaces
 Returns
 r : str
 If ` obj ` has a ` _ _ summary _ _ ` method , it is used .
 If ` obj ` is a ` SortedKeyList ` , then it returns a description
 of the length of the list .
 Otherwise , ` repr ( obj ) ` .'''

implementation:
if hasattr ( obj , '__summary__' ) :
    rep = obj . __summary__ ( )
elif isinstance ( obj , SortedKeyList ) :
    rep = '<{:d} observations>' . format ( len ( obj ) )
else :
    rep = repr ( obj )
return rep . replace ( '\n' , '\n' + ' ' * indent )


signature:
def update ( self , ** kwargs ) :
    '''Update the attributes of a JObject .
 Parameters
 kwargs
 Keyword arguments of the form ` attribute = new _ value `
 Examples
 > > > J = jams . JObject ( foo = 5)
 > > > J . dumps ( )
 ' { " foo " : 5 } '
 > > > J . update ( bar = ' baz ' )
 > > > J . dumps ( )
 ' { " foo " : 5 , " bar " : " baz " } ' '''

implementation:
for name , value in six . iteritems ( kwargs ) :
    setattr ( self , name , value )


signature:
def search ( self , ** kwargs ) :
    '''Query this object ( and its descendants ) .
 Parameters
 kwargs
 Each ` ( key , value ) ` pair encodes a search field in ` key `
 and a target value in ` value ` .
 ` key ` must be a string , and should correspond to a property in
 the JAMS object hierarchy , e . g . , ' Annotation . namespace ` or ` email `
 ` value ` must be either an object ( tested for equality ) , a
 string describing a search pattern ( regular expression ) , or a
 lambda function which evaluates to ` True ` if the candidate
 object matches the search criteria and ` False ` otherwise .
 Returns
 match : bool
 ` True ` if any of the search keys match the specified value ,
 ` False ` otherwise , or if the search keys do not exist
 within the object .
 Examples
 > > > J = jams . JObject ( foo = 5 , needle = ' quick brown fox ' )
 > > > J . search ( needle = ' . * brown . * ' )
 True
 > > > J . search ( needle = ' . * orange . * ' )
 False
 > > > J . search ( badger = ' . * brown . * ' )
 False
 > > > J . search ( foo = 5)
 True
 > > > J . search ( foo = 10)
 False
 > > > J . search ( foo = lambda x : x < 10)
 True
 > > > J . search ( foo = lambda x : x > 10)
 False'''

implementation:
match = False
r_query = { }
myself = self . __class__ . __name__
# Pop this object name off the query
for k , value in six . iteritems ( kwargs ) :
    k_pop = query_pop ( k , myself )
    if k_pop :
        r_query [ k_pop ] = value
if not r_query :
    return False
for key in r_query :
    if hasattr ( self , key ) :
        match |= match_query ( getattr ( self , key ) , r_query [ key ] )
if not match :
    for attr in dir ( self ) :
        obj = getattr ( self , attr )
        if isinstance ( obj , JObject ) :
            match |= obj . search ( ** r_query )
return match


signature:
def validate ( self , strict = True ) :
    '''Validate a JObject against its schema
 Parameters
 strict : bool
 Enforce strict schema validation
 Returns
 valid : bool
 True if the jam validates
 False if not , and ` strict = = False `
 Raises
 SchemaError
 If ` strict = = True ` and ` jam ` fails validation'''

implementation:
valid = True
try :
    jsonschema . validate ( self . __json__ , self . __schema__ )
except jsonschema . ValidationError as invalid :
    if strict :
        raise SchemaError ( str ( invalid ) )
    else :
        warnings . warn ( str ( invalid ) )
    valid = False
return valid


signature:
def append ( self , time = None , duration = None , value = None , confidence = None ) :
    '''Append an observation to the data field
 Parameters
 time : float > = 0
 duration : float > = 0
 The time and duration of the new observation , in seconds
 value
 confidence
 The value and confidence of the new observations .
 Types and values should conform to the namespace of the
 Annotation object .
 Examples
 > > > ann = jams . Annotation ( namespace = ' chord ' )
 > > > ann . append ( time = 3 , duration = 2 , value = ' E # ' )'''

implementation: self . data . add ( Observation ( time = float ( time ) , duration = float ( duration ) , value = value , confidence = confidence ) )


signature:
def append_records ( self , records ) :
    '''Add observations from row - major storage .
 This is primarily useful for deserializing sparsely packed data .
 Parameters
 records : iterable of dicts or Observations
 Each element of ` records ` corresponds to one observation .'''

implementation:
for obs in records :
    if isinstance ( obs , Observation ) :
        self . append ( ** obs . _asdict ( ) )
    else :
        self . append ( ** obs )


signature:
def append_columns ( self , columns ) :
    '''Add observations from column - major storage .
 This is primarily used for deserializing densely packed data .
 Parameters
 columns : dict of lists
 Keys must be ` time , duration , value , confidence ` ,
 and each much be a list of equal length .'''

implementation: self . append_records ( [ dict ( time = t , duration = d , value = v , confidence = c ) for ( t , d , v , c ) in six . moves . zip ( columns [ 'time' ] , columns [ 'duration' ] , columns [ 'value' ] , columns [ 'confidence' ] ) ] )


signature:
def validate ( self , strict = True ) :
    '''Validate this annotation object against the JAMS schema ,
 and its data against the namespace schema .
 Parameters
 strict : bool
 If ` True ` , then schema violations will cause an Exception .
 If ` False ` , then schema violations will issue a warning .
 Returns
 valid : bool
 ` True ` if the object conforms to schema .
 ` False ` if the object fails to conform to schema ,
 but ` strict = = False ` .
 Raises
 SchemaError
 If ` strict = = True ` and the object fails validation
 See Also
 JObject . validate'''

implementation:
# Get the schema for this annotation
ann_schema = schema . namespace_array ( self . namespace )
valid = True
try :
    jsonschema . validate ( self . __json_light__ ( data = False ) , schema . JAMS_SCHEMA )
    # validate each record in the frame
    data_ser = [ serialize_obj ( obs ) for obs in self . data ]
    jsonschema . validate ( data_ser , ann_schema )
except jsonschema . ValidationError as invalid :
    if strict :
        raise SchemaError ( str ( invalid ) )
    else :
        warnings . warn ( str ( invalid ) )
    valid = False
return valid


signature:
def trim ( self , start_time , end_time , strict = False ) :
    '''Trim the annotation and return as a new ` Annotation ` object .
 Trimming will result in the new annotation only containing observations
 that occur in the intersection of the time range spanned by the
 annotation and the time range specified by the user . The new annotation
 will span the time range ` ` [ trim _ start , trim _ end ] ` ` where
 ` ` trim _ start = max ( self . time , start _ time ) ` ` and ` ` trim _ end =
 min ( self . time + self . duration , end _ time ) ` ` .
 If ` ` strict = False ` ` ( default ) observations that start before
 ` ` trim _ start ` ` and end after it will be trimmed such that they start at
 ` ` trim _ start ` ` , and similarly observations that start before
 ` ` trim _ end ` ` and end after it will be trimmed to end at ` ` trim _ end ` ` .
 If ` ` strict = True ` ` such borderline observations will be discarded .
 The new duration of the annotation will be ` ` trim _ end - trim _ start ` ` .
 Note that if the range defined by ` ` [ start _ time , end _ time ] ` `
 doesn ' t intersect with the original time range spanned by the
 annotation the resulting annotation will contain no observations , will
 have the same start time as the original annotation and have duration
 This function also copies over all the annotation metadata from the
 original annotation and documents the trim operation by adding a list
 of tuples to the annotation ' s sandbox keyed by
 ` ` Annotation . sandbox . trim ` ` which documents each trim operation with a
 tuple ` ` ( start _ time , end _ time , trim _ start , trim _ end ) ` ` .
 Parameters
 start _ time : float
 The desired start time for the trimmed annotation in seconds .
 end _ time
 The desired end time for the trimmed annotation in seconds . Must be
 greater than ` ` start _ time ` ` .
 strict : bool
 When ` ` False ` ` ( default ) observations that lie at the boundaries of
 the trimming range ( given by ` ` [ trim _ start , trim _ end ] ` ` as
 described above ) , i . e . observations that start before and end after
 either the trim start or end time , will have their time and / or
 duration adjusted such that only the part of the observation that
 lies within the trim range is kept . When ` ` True ` ` such observations
 are discarded and not included in the trimmed annotation .
 Returns
 ann _ trimmed : Annotation
 The trimmed annotation , returned as a new jams . Annotation object .
 If the trim range specified by ` ` [ start _ time , end _ time ] ` ` does not
 intersect at all with the original time range of the annotation a
 warning will be issued and the returned annotation will be empty .
 Raises
 ParameterError
 If ` ` end _ time ` ` is not greater than ` ` start _ time ` ` .
 Examples
 > > > ann = jams . Annotation ( namespace = ' tag _ open ' , time = 2 , duration = 8)
 > > > ann . append ( time = 2 , duration = 2 , value = ' one ' )
 > > > ann . append ( time = 4 , duration = 2 , value = ' two ' )
 > > > ann . append ( time = 6 , duration = 2 , value = ' three ' )
 > > > ann . append ( time = 7 , duration = 2 , value = ' four ' )
 > > > ann . append ( time = 8 , duration = 2 , value = ' five ' )
 > > > ann _ trim = ann . trim ( 5 , 8 , strict = False )
 > > > print ( ann _ trim . time , ann _ trim . duration )
 (5 , 3)
 > > > ann _ trim . to _ dataframe ( )
 time duration value confidence
 0 5 1 two None
 1 6 2 three None
 2 7 1 four None
 > > > ann _ trim _ strict = ann . trim ( 5 , 8 , strict = True )
 > > > print ( ann _ trim _ strict . time , ann _ trim _ strict . duration )
 (5 , 3)
 > > > ann _ trim _ strict . to _ dataframe ( )
 time duration value confidence
 0 6 2 three None'''

implementation:
# Check for basic start _ time and end _ time validity
if end_time <= start_time :
    raise ParameterError ( 'end_time must be greater than start_time.' )
# If the annotation does not have a set duration value , we ' ll assume
# trimming is possible ( up to the user to ensure this is valid ) .
if self . duration is None :
    orig_time = start_time
    orig_duration = end_time - start_time
    warnings . warn ( "Annotation.duration is not defined, cannot check " "for temporal intersection, assuming the annotation " "is valid between start_time and end_time." )
else :
    orig_time = self . time
    orig_duration = self . duration
# Check whether there is intersection between the trim range and
# annotation : if not raise a warning and set trim _ start and trim _ end
# appropriately .
if start_time > ( orig_time + orig_duration ) or ( end_time < orig_time ) :
    warnings . warn ( 'Time range defined by [start_time,end_time] does not ' 'intersect with the time range spanned by this annotation, ' 'the trimmed annotation will be empty.' )
    trim_start = self . time
    trim_end = trim_start
else : # Determine new range
    trim_start = max ( orig_time , start_time )
    trim_end = min ( orig_time + orig_duration , end_time )
# Create new annotation with same namespace / metadata
ann_trimmed = Annotation ( self . namespace , data = None , annotation_metadata = self . annotation_metadata , sandbox = self . sandbox , time = trim_start , duration = trim_end - trim_start )
# Selectively add observations based on their start time / duration
# We do this rather than copying and directly manipulating the
# annotation ' data frame ( which might be faster ) since this way trim is
# independent of the internal data representation .
for obs in self . data :
    obs_start = obs . time
    obs_end = obs_start + obs . duration
    if obs_start < trim_end and obs_end > trim_start :
        new_start = max ( obs_start , trim_start )
        new_end = min ( obs_end , trim_end )
        new_duration = new_end - new_start
        if ( ( not strict ) or ( new_start == obs_start and new_end == obs_end ) ) :
            ann_trimmed . append ( time = new_start , duration = new_duration , value = obs . value , confidence = obs . confidence )
if 'trim' not in ann_trimmed . sandbox . keys ( ) :
    ann_trimmed . sandbox . update ( trim = [ { 'start_time' : start_time , 'end_time' : end_time , 'trim_start' : trim_start , 'trim_end' : trim_end } ] )
else :
    ann_trimmed . sandbox . trim . append ( { 'start_time' : start_time , 'end_time' : end_time , 'trim_start' : trim_start , 'trim_end' : trim_end } )
return ann_trimmed


signature:
def slice ( self , start_time , end_time , strict = False ) :
    '''Slice the annotation and return as a new ` Annotation ` object .
 Slicing has the same effect as trimming ( see ` Annotation . trim ` ) except
 that while trimming does not modify the start time of the annotation or
 the observations it contains , slicing will set the new annotation ' s
 start time to ` ` max ( 0 , trimmed _ annotation . time - start _ time ) ` ` and the
 start time of its observations will be set with respect to this new
 reference start time .
 This function documents the slice operation by adding a list of tuples
 to the annotation ' s sandbox keyed by ` ` Annotation . sandbox . slice ` ` which
 documents each slice operation with a tuple
 ` ` ( start _ time , end _ time , slice _ start , slice _ end ) ` ` , where
 ` ` slice _ start ` ` and ` ` slice _ end ` ` are given by ` ` trim _ start ` ` and
 ` ` trim _ end ` ` ( see ` Annotation . trim ` ) .
 Since slicing is implemented using trimming , the trimming operation
 will also be documented in ` ` Annotation . sandbox . trim ` ` as described in
 ` Annotation . trim ` .
 This function is useful for example when trimming an audio file ,
 allowing the user to trim the annotation while ensuring all time
 information matches the new trimmed audio file .
 Parameters
 start _ time : float
 The desired start time for slicing in seconds .
 end _ time
 The desired end time for slicing in seconds . Must be greater than
 ` ` start _ time ` ` .
 strict : bool
 When ` ` False ` ` ( default ) observations that lie at the boundaries of
 the slice ( see ` Annotation . trim ` for details ) will have their time
 and / or duration adjusted such that only the part of the observation
 that lies within the slice range is kept . When ` ` True ` ` such
 observations are discarded and not included in the sliced
 annotation .
 Returns
 sliced _ ann : Annotation
 The sliced annotation .
 See Also
 Annotation . trim
 Examples
 > > > ann = jams . Annotation ( namespace = ' tag _ open ' , time = 2 , duration = 8)
 > > > ann . append ( time = 2 , duration = 2 , value = ' one ' )
 > > > ann . append ( time = 4 , duration = 2 , value = ' two ' )
 > > > ann . append ( time = 6 , duration = 2 , value = ' three ' )
 > > > ann . append ( time = 7 , duration = 2 , value = ' four ' )
 > > > ann . append ( time = 8 , duration = 2 , value = ' five ' )
 > > > ann _ slice = ann . slice ( 5 , 8 , strict = False )
 > > > print ( ann _ slice . time , ann _ slice . duration )
 (0 , 3)
 > > > ann _ slice . to _ dataframe ( )
 time duration value confidence
 0 0.0 1.0 two None
 1 1.0 2.0 three None
 2 2.0 1.0 four None
 > > > ann _ slice _ strict = ann . slice ( 5 , 8 , strict = True )
 > > > print ( ann _ slice _ strict . time , ann _ slice _ strict . duration )
 (0 , 3)
 > > > ann _ slice _ strict . to _ dataframe ( )
 time duration value confidence
 0 1.0 2.0 three None'''

implementation:
# start by trimming the annotation
sliced_ann = self . trim ( start_time , end_time , strict = strict )
raw_data = sliced_ann . pop_data ( )
# now adjust the start time of the annotation and the observations it
# contains .
for obs in raw_data :
    new_time = max ( 0 , obs . time - start_time )
    # if obs . time > start _ time ,
    # duration doesn ' t change
    # if obs . time < start _ time ,
    # duration shrinks by start _ time - obs . time
    sliced_ann . append ( time = new_time , duration = obs . duration , value = obs . value , confidence = obs . confidence )
ref_time = sliced_ann . time
slice_start = ref_time
slice_end = ref_time + sliced_ann . duration
if 'slice' not in sliced_ann . sandbox . keys ( ) :
    sliced_ann . sandbox . update ( slice = [ { 'start_time' : start_time , 'end_time' : end_time , 'slice_start' : slice_start , 'slice_end' : slice_end } ] )
else :
    sliced_ann . sandbox . slice . append ( { 'start_time' : start_time , 'end_time' : end_time , 'slice_start' : slice_start , 'slice_end' : slice_end } )
# Update the timing for the sliced annotation
sliced_ann . time = max ( 0 , ref_time - start_time )
return sliced_ann


signature:
def pop_data ( self ) :
    '''Replace this observation ' s data with a fresh container .
 Returns
 annotation _ data : SortedKeyList
 The original annotation data container'''

implementation:
data = self . data
self . data = SortedKeyList ( key = self . _key )
return data


signature:
def to_interval_values ( self ) :
    '''Extract observation data in a ` mir _ eval ` - friendly format .
 Returns
 intervals : np . ndarray [ shape = ( n , 2 ) , dtype = float ]
 Start - and end - times of all valued intervals
 ` intervals [ i , : ] = [ time [ i ] , time [ i ] + duration [ i ] ] `
 labels : list
 List view of value field .'''

implementation:
ints , vals = [ ] , [ ]
for obs in self . data :
    ints . append ( [ obs . time , obs . time + obs . duration ] )
    vals . append ( obs . value )
if not ints :
    return np . empty ( shape = ( 0 , 2 ) , dtype = float ) , [ ]
return np . array ( ints ) , vals


signature:
def to_event_values ( self ) :
    '''Extract observation data in a ` mir _ eval ` - friendly format .
 Returns
 times : np . ndarray [ shape = ( n , ) , dtype = float ]
 Start - time of all observations
 labels : list
 List view of value field .'''

implementation:
ints , vals = [ ] , [ ]
for obs in self . data :
    ints . append ( obs . time )
    vals . append ( obs . value )
return np . array ( ints ) , vals


signature:
def to_samples ( self , times , confidence = False ) :
    '''Sample the annotation at specified times .
 Parameters
 times : np . ndarray , non - negative , ndim = 1
 The times ( in seconds ) to sample the annotation
 confidence : bool
 If ` True ` , return both values and confidences .
 If ` False ` ( default ) only return values .
 Returns
 values : list
 ` values [ i ] ` is a list of observation values for intervals
 that cover ` times [ i ] ` .
 confidence : list ( optional )
 ` confidence ` values corresponding to ` values `'''

implementation:
times = np . asarray ( times )
if times . ndim != 1 or np . any ( times < 0 ) :
    raise ParameterError ( 'times must be 1-dimensional and non-negative' )
idx = np . argsort ( times )
samples = times [ idx ]
values = [ list ( ) for _ in samples ]
confidences = [ list ( ) for _ in samples ]
for obs in self . data :
    start = np . searchsorted ( samples , obs . time )
    end = np . searchsorted ( samples , obs . time + obs . duration , side = 'right' )
    for i in range ( start , end ) :
        values [ idx [ i ] ] . append ( obs . value )
        confidences [ idx [ i ] ] . append ( obs . confidence )
if confidence :
    return values , confidences
else :
    return values


signature:
def to_html ( self , max_rows = None ) :
    '''Render this annotation list in HTML
 Returns
 rendered : str
 An HTML table containing this annotation ' s data .'''

implementation:
n = len ( self . data )
div_id = _get_divid ( self )
out = r'''

                            <button
                                type="button"
                                data-toggle="collapse"
                                data-parent="#accordion"
                                href="#{0}"
                                aria-expanded="false"
                                class="collapsed btn btn-info btn-block"
                                aria-controls="{0}">
                                {1:s}
                                {2:d}
                            </button>

''' . format ( div_id , self . namespace , n )
out += r'''

''' . format ( div_id )
out += r'''

                        {}

''' . format ( self . annotation_metadata . _repr_html_ ( ) )
out += r'''

                        {}

''' . format ( self . sandbox . _repr_html_ ( ) )
# - - Annotation content starts here
out += r'''

                            time
                            duration
                            value
                            confidence

''' . format ( self . namespace , n )
out += r'''
'''
if max_rows is None or n <= max_rows :
    out += self . _fmt_rows ( 0 , n )
else :
    out += self . _fmt_rows ( 0 , max_rows // 2 )
    out += r'''

                            ...
                            ...
                            ...
                            ...
                            ...

'''
    out += self . _fmt_rows ( n - max_rows // 2 , n )
out += r'''
'''
out += r'''

'''
out += r'''

'''
return out


signature:
def _key ( cls , obs ) :
    '''Provides sorting index for Observation objects'''

implementation:
if not isinstance ( obs , Observation ) :
    raise JamsError ( '{} must be of type jams.Observation' . format ( obs ) )
return obs . time


signature:
def search ( self , ** kwargs ) :
    '''Filter the annotation array down to only those Annotation
 objects matching the query .
 Parameters
 kwargs : search parameters
 See JObject . search
 Returns
 results : AnnotationArray
 An annotation array of the objects matching the query
 See Also
 JObject . search'''

implementation:
results = AnnotationArray ( )
for annotation in self :
    if annotation . search ( ** kwargs ) :
        results . append ( annotation )
return results


signature:
def trim ( self , start_time , end_time , strict = False ) :
    '''Trim every annotation contained in the annotation array using
 ` Annotation . trim ` and return as a new ` AnnotationArray ` .
 See ` Annotation . trim ` for details about trimming . This function does
 not modify the annotations in the original annotation array .
 Parameters
 start _ time : float
 The desired start time for the trimmed annotations in seconds .
 end _ time
 The desired end time for trimmed annotations in seconds . Must be
 greater than ` ` start _ time ` ` .
 strict : bool
 When ` ` False ` ` ( default ) observations that lie at the boundaries of
 the trimming range ( see ` Annotation . trim ` for details ) will have
 their time and / or duration adjusted such that only the part of the
 observation that lies within the trim range is kept . When ` ` True ` `
 such observations are discarded and not included in the trimmed
 annotation .
 Returns
 trimmed _ array : AnnotationArray
 An annotation array where every annotation has been trimmed .'''

implementation:
trimmed_array = AnnotationArray ( )
for ann in self :
    trimmed_array . append ( ann . trim ( start_time , end_time , strict = strict ) )
return trimmed_array


signature:
def slice ( self , start_time , end_time , strict = False ) :
    '''Slice every annotation contained in the annotation array using
 ` Annotation . slice `
 and return as a new AnnotationArray
 See ` Annotation . slice ` for details about slicing . This function does
 not modify the annotations in the original annotation array .
 Parameters
 start _ time : float
 The desired start time for slicing in seconds .
 end _ time
 The desired end time for slicing in seconds . Must be greater than
 ` ` start _ time ` ` .
 strict : bool
 When ` ` False ` ` ( default ) observations that lie at the boundaries of
 the slicing range ( see ` Annotation . slice ` for details ) will have
 their time and / or duration adjusted such that only the part of the
 observation that lies within the trim range is kept . When ` ` True ` `
 such observations are discarded and not included in the sliced
 annotation .
 Returns
 sliced _ array : AnnotationArray
 An annotation array where every annotation has been sliced .'''

implementation:
sliced_array = AnnotationArray ( )
for ann in self :
    sliced_array . append ( ann . slice ( start_time , end_time , strict = strict ) )
return sliced_array


signature:
def add ( self , jam , on_conflict = 'fail' ) :
    """Add the contents of another jam to this object .
 Note that , by default , this method fails if file _ metadata is not
 identical and raises a ValueError ; either resolve this manually
 ( because conflicts should almost never happen ) , force an ' overwrite ' ,
 or tell the method to ' ignore ' the metadata of the object being added .
 Parameters
 jam : JAMS object
 Object to add to this jam
 on _ conflict : str , default = ' fail '
 Strategy for resolving metadata conflicts ; one of
 [ ' fail ' , ' overwrite ' , or ' ignore ' ] .
 Raises
 ParameterError
 if ` on _ conflict ` is an unknown value
 JamsError
 If a conflict is detected and ` on _ conflict = ' fail ' `"""

implementation:
if on_conflict not in [ 'overwrite' , 'fail' , 'ignore' ] :
    raise ParameterError ( "on_conflict='{}' is not in ['fail', " "'overwrite', 'ignore']." . format ( on_conflict ) )
if not self . file_metadata == jam . file_metadata :
    if on_conflict == 'overwrite' :
        self . file_metadata = jam . file_metadata
    elif on_conflict == 'fail' :
        raise JamsError ( "Metadata conflict! " "Resolve manually or force-overwrite it." )
self . annotations . extend ( jam . annotations )
self . sandbox . update ( ** jam . sandbox )


signature:
def save ( self , path_or_file , strict = True , fmt = 'auto' ) :
    """Serialize annotation as a JSON formatted stream to file .
 Parameters
 path _ or _ file : str or file - like
 Path to save the JAMS object on disk
 OR
 An open file descriptor to write into
 strict : bool
 Force strict schema validation
 fmt : str [ ' auto ' , ' jams ' , ' jamz ' ]
 The output encoding format .
 If ` auto ` , it is inferred from the file name .
 If the input is an open file handle , ` jams ` encoding
 is used .
 Raises
 SchemaError
 If ` strict = = True ` and the JAMS object fails schema
 or namespace validation .
 See also
 validate"""

implementation:
self . validate ( strict = strict )
with _open ( path_or_file , mode = 'w' , fmt = fmt ) as fdesc :
    json . dump ( self . __json__ , fdesc , indent = 2 )


signature:
def validate ( self , strict = True ) :
    '''Validate a JAMS object against the schema .
 Parameters
 strict : bool
 If ` True ` , an exception will be raised on validation failure .
 If ` False ` , a warning will be raised on validation failure .
 Returns
 valid : bool
 ` True ` if the object passes schema validation .
 ` False ` otherwise .
 Raises
 SchemaError
 If ` strict = = True ` and the JAMS object does not match the schema
 See Also
 jsonschema . validate'''

implementation:
valid = True
try :
    jsonschema . validate ( self . __json_light__ , schema . JAMS_SCHEMA )
    for ann in self . annotations :
        if isinstance ( ann , Annotation ) :
            valid &= ann . validate ( strict = strict )
        else :
            msg = '{} is not a well-formed JAMS Annotation' . format ( ann )
            valid = False
            if strict :
                raise SchemaError ( msg )
            else :
                warnings . warn ( str ( msg ) )
except jsonschema . ValidationError as invalid :
    if strict :
        raise SchemaError ( str ( invalid ) )
    else :
        warnings . warn ( str ( invalid ) )
    valid = False
return valid


signature:
def trim ( self , start_time , end_time , strict = False ) :
    '''Trim all the annotations inside the jam and return as a new ` JAMS `
 object .
 See ` Annotation . trim ` for details about how the annotations
 are trimmed .
 This operation is also documented in the jam - level sandbox
 with a list keyed by ` ` JAMS . sandbox . trim ` ` containing a tuple for each
 jam - level trim of the form ` ` ( start _ time , end _ time ) ` ` .
 This function also copies over all of the file metadata from the
 original jam .
 Note : trimming does not affect the duration of the jam , i . e . the value
 of ` ` JAMS . file _ metadata . duration ` ` will be the same for the original
 and trimmed jams .
 Parameters
 start _ time : float
 The desired start time for the trimmed annotations in seconds .
 end _ time
 The desired end time for trimmed annotations in seconds . Must be
 greater than ` ` start _ time ` ` .
 strict : bool
 When ` ` False ` ` ( default ) observations that lie at the boundaries of
 the trimming range ( see ` Annotation . trim ` for details ) , will have
 their time and / or duration adjusted such that only the part of the
 observation that lies within the trim range is kept . When ` ` True ` `
 such observations are discarded and not included in the trimmed
 annotation .
 Returns
 jam _ trimmed : JAMS
 The trimmed jam with trimmed annotations , returned as a new JAMS
 object .'''

implementation:
# Make sure duration is set in file metadata
if self . file_metadata . duration is None :
    raise JamsError ( 'Duration must be set (jam.file_metadata.duration) before ' 'trimming can be performed.' )
# Make sure start and end times are within the file start / end times
if not ( 0 <= start_time <= end_time <= float ( self . file_metadata . duration ) ) :
    raise ParameterError ( 'start_time and end_time must be within the original file ' 'duration ({:f}) and end_time cannot be smaller than ' 'start_time.' . format ( float ( self . file_metadata . duration ) ) )
# Create a new jams
jam_trimmed = JAMS ( annotations = None , file_metadata = self . file_metadata , sandbox = self . sandbox )
# trim annotations
jam_trimmed . annotations = self . annotations . trim ( start_time , end_time , strict = strict )
# Document jam - level trim in top level sandbox
if 'trim' not in jam_trimmed . sandbox . keys ( ) :
    jam_trimmed . sandbox . update ( trim = [ { 'start_time' : start_time , 'end_time' : end_time } ] )
else :
    jam_trimmed . sandbox . trim . append ( { 'start_time' : start_time , 'end_time' : end_time } )
return jam_trimmed


signature:
def slice ( self , start_time , end_time , strict = False ) :
    '''Slice all the annotations inside the jam and return as a new ` JAMS `
 object .
 See ` Annotation . slice ` for details about how the annotations
 are sliced .
 This operation is also documented in the jam - level sandbox
 with a list keyed by ` ` JAMS . sandbox . slice ` ` containing a tuple for each
 jam - level slice of the form ` ` ( start _ time , end _ time ) ` ` .
 Since slicing is implemented using trimming , the operation will also be
 documented in ` ` JAMS . sandbox . trim ` ` as described in ` JAMS . trim ` .
 This function also copies over all of the file metadata from the
 original jam .
 Note : slicing will affect the duration of the jam , i . e . the new value
 of ` ` JAMS . file _ metadata . duration ` ` will be ` ` end _ time - start _ time ` ` .
 Parameters
 start _ time : float
 The desired start time for slicing in seconds .
 end _ time
 The desired end time for slicing in seconds . Must be greater than
 ` ` start _ time ` ` .
 strict : bool
 When ` ` False ` ` ( default ) observations that lie at the boundaries of
 the slicing range ( see ` Annotation . slice ` for details ) , will have
 their time and / or duration adjusted such that only the part of the
 observation that lies within the slice range is kept . When ` ` True ` `
 such observations are discarded and not included in the sliced
 annotation .
 Returns
 jam _ sliced : JAMS
 The sliced jam with sliced annotations , returned as a new
 JAMS object .'''

implementation:
# Make sure duration is set in file metadata
if self . file_metadata . duration is None :
    raise JamsError ( 'Duration must be set (jam.file_metadata.duration) before ' 'slicing can be performed.' )
# Make sure start and end times are within the file start / end times
if ( start_time < 0 or start_time > float ( self . file_metadata . duration ) or end_time < start_time or end_time > float ( self . file_metadata . duration ) ) :
    raise ParameterError ( 'start_time and end_time must be within the original file ' 'duration ({:f}) and end_time cannot be smaller than ' 'start_time.' . format ( float ( self . file_metadata . duration ) ) )
# Create a new jams
jam_sliced = JAMS ( annotations = None , file_metadata = self . file_metadata , sandbox = self . sandbox )
# trim annotations
jam_sliced . annotations = self . annotations . slice ( start_time , end_time , strict = strict )
# adjust dutation
jam_sliced . file_metadata . duration = end_time - start_time
# Document jam - level trim in top level sandbox
if 'slice' not in jam_sliced . sandbox . keys ( ) :
    jam_sliced . sandbox . update ( slice = [ { 'start_time' : start_time , 'end_time' : end_time } ] )
else :
    jam_sliced . sandbox . slice . append ( { 'start_time' : start_time , 'end_time' : end_time } )
return jam_sliced


signature:
def pprint_jobject ( obj , ** kwargs ) :
    '''Pretty - print a jobject .
 Parameters
 obj : jams . JObject
 kwargs
 additional parameters to ` json . dumps `
 Returns
 string
 A simplified display of ` obj ` contents .'''

implementation:
obj_simple = { k : v for k , v in six . iteritems ( obj . __json__ ) if v }
string = json . dumps ( obj_simple , ** kwargs )
# Suppress braces and quotes
string = re . sub ( r'[{}"]' , '' , string )
# Kill trailing commas
string = re . sub ( r',\n' , '\n' , string )
# Kill blank lines
string = re . sub ( r'^\s*$' , '' , string )
return string


signature:
def intervals ( annotation , ** kwargs ) :
    '''Plotting wrapper for labeled intervals'''

implementation:
times , labels = annotation . to_interval_values ( )
return mir_eval . display . labeled_intervals ( times , labels , ** kwargs )


signature:
def hierarchy ( annotation , ** kwargs ) :
    '''Plotting wrapper for hierarchical segmentations'''

implementation:
htimes , hlabels = hierarchy_flatten ( annotation )
htimes = [ np . asarray ( _ ) for _ in htimes ]
return mir_eval . display . hierarchy ( htimes , hlabels , ** kwargs )


signature:
def pitch_contour ( annotation , ** kwargs ) :
    '''Plotting wrapper for pitch contours'''

implementation:
ax = kwargs . pop ( 'ax' , None )
# If the annotation is empty , we need to construct a new axes
ax = mir_eval . display . __get_axes ( ax = ax ) [ 0 ]
times , values = annotation . to_interval_values ( )
indices = np . unique ( [ v [ 'index' ] for v in values ] )
for idx in indices :
    rows = [ i for ( i , v ) in enumerate ( values ) if v [ 'index' ] == idx ]
    freqs = np . asarray ( [ values [ r ] [ 'frequency' ] for r in rows ] )
    unvoiced = ~ np . asarray ( [ values [ r ] [ 'voiced' ] for r in rows ] )
    freqs [ unvoiced ] *= - 1
    ax = mir_eval . display . pitch ( times [ rows , 0 ] , freqs , unvoiced = True , ax = ax , ** kwargs )
return ax


signature:
def event ( annotation , ** kwargs ) :
    '''Plotting wrapper for events'''

implementation:
times , values = annotation . to_interval_values ( )
if any ( values ) :
    labels = values
else :
    labels = None
return mir_eval . display . events ( times , labels = labels , ** kwargs )


signature:
def beat_position ( annotation , ** kwargs ) :
    '''Plotting wrapper for beat - position data'''

implementation:
times , values = annotation . to_interval_values ( )
labels = [ _ [ 'position' ] for _ in values ]
# TODO : plot time signature , measure number
return mir_eval . display . events ( times , labels = labels , ** kwargs )


signature:
def piano_roll ( annotation , ** kwargs ) :
    '''Plotting wrapper for piano rolls'''

implementation:
times , midi = annotation . to_interval_values ( )
return mir_eval . display . piano_roll ( times , midi = midi , ** kwargs )


signature:
def display ( annotation , meta = True , ** kwargs ) :
    '''Visualize a jams annotation through mir _ eval
 Parameters
 annotation : jams . Annotation
 The annotation to display
 meta : bool
 If ` True ` , include annotation metadata in the figure
 kwargs
 Additional keyword arguments to mir _ eval . display functions
 Returns
 ax
 Axis handles for the new display
 Raises
 NamespaceError
 If the annotation cannot be visualized'''

implementation:
for namespace , func in six . iteritems ( VIZ_MAPPING ) :
    try :
        ann = coerce_annotation ( annotation , namespace )
        axes = func ( ann , ** kwargs )
        # Title should correspond to original namespace , not the coerced version
        axes . set_title ( annotation . namespace )
        if meta :
            description = pprint_jobject ( annotation . annotation_metadata , indent = 2 )
            anchored_box = AnchoredText ( description . strip ( '\n' ) , loc = 2 , frameon = True , bbox_to_anchor = ( 1.02 , 1.0 ) , bbox_transform = axes . transAxes , borderpad = 0.0 )
            axes . add_artist ( anchored_box )
            axes . figure . subplots_adjust ( right = 0.8 )
        return axes
    except NamespaceError :
        pass
raise NamespaceError ( 'Unable to visualize annotation of namespace="{:s}"' . format ( annotation . namespace ) )


signature:
def display_multi ( annotations , fig_kw = None , meta = True , ** kwargs ) :
    '''Display multiple annotations with shared axes
 Parameters
 annotations : jams . AnnotationArray
 A collection of annotations to display
 fig _ kw : dict
 Keyword arguments to ` plt . figure `
 meta : bool
 If ` True ` , display annotation metadata for each annotation
 kwargs
 Additional keyword arguments to the ` mir _ eval . display ` routines
 Returns
 fig
 The created figure
 axs
 List of subplot axes corresponding to each displayed annotation'''

implementation:
if fig_kw is None :
    fig_kw = dict ( )
fig_kw . setdefault ( 'sharex' , True )
fig_kw . setdefault ( 'squeeze' , True )
# Filter down to coercable annotations first
display_annotations = [ ]
for ann in annotations :
    for namespace in VIZ_MAPPING :
        if can_convert ( ann , namespace ) :
            display_annotations . append ( ann )
            break
# If there are no displayable annotations , fail here
if not len ( display_annotations ) :
    raise ParameterError ( 'No displayable annotations found' )
fig , axs = plt . subplots ( nrows = len ( display_annotations ) , ncols = 1 , ** fig_kw )
# MPL is stupid when making singleton subplots .
# We catch this and make it always iterable .
if len ( display_annotations ) == 1 :
    axs = [ axs ]
for ann , ax in zip ( display_annotations , axs ) :
    kwargs [ 'ax' ] = ax
    display ( ann , meta = meta , ** kwargs )
return fig , axs


signature:
def mkclick ( freq , sr = 22050 , duration = 0.1 ) :
    '''Generate a click sample .
 This replicates functionality from mir _ eval . sonify . clicks ,
 but exposes the target frequency and duration .'''

implementation:
times = np . arange ( int ( sr * duration ) )
click = np . sin ( 2 * np . pi * times * freq / float ( sr ) )
click *= np . exp ( - times / ( 1e-2 * sr ) )
return click


signature:
def clicks ( annotation , sr = 22050 , length = None , ** kwargs ) :
    '''Sonify events with clicks .
 This uses mir _ eval . sonify . clicks , and is appropriate for instantaneous
 events such as beats or segment boundaries .'''

implementation:
interval , _ = annotation . to_interval_values ( )
return filter_kwargs ( mir_eval . sonify . clicks , interval [ : , 0 ] , fs = sr , length = length , ** kwargs )


signature:
def downbeat ( annotation , sr = 22050 , length = None , ** kwargs ) :
    '''Sonify beats and downbeats together .'''

implementation:
beat_click = mkclick ( 440 * 2 , sr = sr )
downbeat_click = mkclick ( 440 * 3 , sr = sr )
intervals , values = annotation . to_interval_values ( )
beats , downbeats = [ ] , [ ]
for time , value in zip ( intervals [ : , 0 ] , values ) :
    if value [ 'position' ] == 1 :
        downbeats . append ( time )
    else :
        beats . append ( time )
if length is None :
    length = int ( sr * np . max ( intervals ) ) + len ( beat_click ) + 1
y = filter_kwargs ( mir_eval . sonify . clicks , np . asarray ( beats ) , fs = sr , length = length , click = beat_click )
y += filter_kwargs ( mir_eval . sonify . clicks , np . asarray ( downbeats ) , fs = sr , length = length , click = downbeat_click )
return y


signature:
def multi_segment ( annotation , sr = 22050 , length = None , ** kwargs ) :
    '''Sonify multi - level segmentations'''

implementation:
# Pentatonic scale , because why not
PENT = [ 1 , 32. / 27 , 4. / 3 , 3. / 2 , 16. / 9 ]
DURATION = 0.1
h_int , _ = hierarchy_flatten ( annotation )
if length is None :
    length = int ( sr * ( max ( np . max ( _ ) for _ in h_int ) + 1. / DURATION ) + 1 )
y = 0.0
for ints , ( oc , scale ) in zip ( h_int , product ( range ( 3 , 3 + len ( h_int ) ) , PENT ) ) :
    click = mkclick ( 440.0 * scale * oc , sr = sr , duration = DURATION )
    y = y + filter_kwargs ( mir_eval . sonify . clicks , np . unique ( ints ) , fs = sr , length = length , click = click )
return y


signature:
def chord ( annotation , sr = 22050 , length = None , ** kwargs ) :
    '''Sonify chords
 This uses mir _ eval . sonify . chords .'''

implementation:
intervals , chords = annotation . to_interval_values ( )
return filter_kwargs ( mir_eval . sonify . chords , chords , intervals , fs = sr , length = length , ** kwargs )


signature:
def pitch_contour ( annotation , sr = 22050 , length = None , ** kwargs ) :
    '''Sonify pitch contours .
 This uses mir _ eval . sonify . pitch _ contour , and should only be applied
 to pitch annotations using the pitch _ contour namespace .
 Each contour is sonified independently , and the resulting waveforms
 are summed together .'''

implementation:
# Map contours to lists of observations
times = defaultdict ( list )
freqs = defaultdict ( list )
for obs in annotation :
    times [ obs . value [ 'index' ] ] . append ( obs . time )
    freqs [ obs . value [ 'index' ] ] . append ( obs . value [ 'frequency' ] * ( - 1 ) ** ( ~ obs . value [ 'voiced' ] ) )
y_out = 0.0
for ix in times :
    y_out = y_out + filter_kwargs ( mir_eval . sonify . pitch_contour , np . asarray ( times [ ix ] ) , np . asarray ( freqs [ ix ] ) , fs = sr , length = length , ** kwargs )
    if length is None :
        length = len ( y_out )
return y_out


signature:
def piano_roll ( annotation , sr = 22050 , length = None , ** kwargs ) :
    '''Sonify a piano - roll
 This uses mir _ eval . sonify . time _ frequency , and is appropriate
 for sparse transcription data , e . g . , annotations in the ` note _ midi `
 namespace .'''

implementation:
intervals , pitches = annotation . to_interval_values ( )
# Construct the pitchogram
pitch_map = { f : idx for idx , f in enumerate ( np . unique ( pitches ) ) }
gram = np . zeros ( ( len ( pitch_map ) , len ( intervals ) ) )
for col , f in enumerate ( pitches ) :
    gram [ pitch_map [ f ] , col ] = 1
return filter_kwargs ( mir_eval . sonify . time_frequency , gram , pitches , intervals , sr , length = length , ** kwargs )


signature:
def sonify ( annotation , sr = 22050 , duration = None , ** kwargs ) :
    '''Sonify a jams annotation through mir _ eval
 Parameters
 annotation : jams . Annotation
 The annotation to sonify
 sr = : positive number
 The sampling rate of the output waveform
 duration : float ( optional )
 Optional length ( in seconds ) of the output waveform
 kwargs
 Additional keyword arguments to mir _ eval . sonify functions
 Returns
 y _ sonified : np . ndarray
 The waveform of the sonified annotation
 Raises
 NamespaceError
 If the annotation has an un - sonifiable namespace'''

implementation:
length = None
if duration is None :
    duration = annotation . duration
if duration is not None :
    length = int ( duration * sr )
# If the annotation can be directly sonified , try that first
if annotation . namespace in SONIFY_MAPPING :
    ann = coerce_annotation ( annotation , annotation . namespace )
    return SONIFY_MAPPING [ annotation . namespace ] ( ann , sr = sr , length = length , ** kwargs )
for namespace , func in six . iteritems ( SONIFY_MAPPING ) :
    try :
        ann = coerce_annotation ( annotation , namespace )
        return func ( ann , sr = sr , length = length , ** kwargs )
    except NamespaceError :
        pass
raise NamespaceError ( 'Unable to sonify annotation of namespace="{:s}"' . format ( annotation . namespace ) )


signature:
def validate ( schema_file = None , jams_files = None ) :
    '''Validate a jams file against a schema'''

implementation:
schema = load_json ( schema_file )
for jams_file in jams_files :
    try :
        jams = load_json ( jams_file )
        jsonschema . validate ( jams , schema )
        print '{:s} was successfully validated' . format ( jams_file )
    except jsonschema . ValidationError as exc :
        print '{:s} was NOT successfully validated' . format ( jams_file )
        print exc


signature:
def make_stream_features ( self , stream , features ) :
    """Add SASL features to the < features / > element of the stream .
 [ receving entity only ]
 : returns : update < features / > element ."""

implementation:
mechs = self . settings [ 'sasl_mechanisms' ]
if mechs and not stream . authenticated :
    sub = ElementTree . SubElement ( features , MECHANISMS_TAG )
    for mech in mechs :
        if mech in sasl . SERVER_MECHANISMS :
            ElementTree . SubElement ( sub , MECHANISM_TAG ) . text = mech
return features


signature:
def handle_stream_features ( self , stream , features ) :
    """Process incoming < stream : features / > element .
 [ initiating entity only ]"""

implementation:
element = features . find ( MECHANISMS_TAG )
self . peer_sasl_mechanisms = [ ]
if element is None :
    return None
for sub in element :
    if sub . tag != MECHANISM_TAG :
        continue
    self . peer_sasl_mechanisms . append ( sub . text )
if stream . authenticated or not self . peer_sasl_mechanisms :
    return StreamFeatureNotHandled ( "SASL" , mandatory = True )
username = self . settings . get ( "username" )
if not username : # TODO : other rules for s2s
    if stream . me . local :
        username = stream . me . local
    else :
        username = None
self . _sasl_authenticate ( stream , username , self . settings . get ( "authzid" ) )
return StreamFeatureHandled ( "SASL" , mandatory = True )


signature:
def process_sasl_auth ( self , stream , element ) :
    """Process incoming < sasl : auth / > element .
 [ receiving entity only ]"""

implementation:
if self . authenticator :
    logger . debug ( "Authentication already started" )
    return False
password_db = self . settings [ "password_database" ]
mechanism = element . get ( "mechanism" )
if not mechanism :
    stream . send_stream_error ( "bad-format" )
    raise FatalStreamError ( "<sasl:auth/> with no mechanism" )
stream . auth_method_used = mechanism
self . authenticator = sasl . server_authenticator_factory ( mechanism , password_db )
content = element . text . encode ( "us-ascii" )
ret = self . authenticator . start ( stream . auth_properties , a2b_base64 ( content ) )
if isinstance ( ret , sasl . Success ) :
    element = ElementTree . Element ( SUCCESS_TAG )
    element . text = ret . encode ( )
elif isinstance ( ret , sasl . Challenge ) :
    element = ElementTree . Element ( CHALLENGE_TAG )
    element . text = ret . encode ( )
else :
    element = ElementTree . Element ( FAILURE_TAG )
    ElementTree . SubElement ( element , SASL_QNP + ret . reason )
stream . write_element ( element )
if isinstance ( ret , sasl . Success ) :
    self . _handle_auth_success ( stream , ret )
elif isinstance ( ret , sasl . Failure ) :
    raise SASLAuthenticationFailed ( "SASL authentication failed: {0}" . format ( ret . reason ) )
return True


signature:
def _handle_auth_success ( self , stream , success ) :
    """Handle successful authentication .
 Send < success / > and mark the stream peer authenticated .
 [ receiver only ]"""

implementation:
if not self . _check_authorization ( success . properties , stream ) :
    element = ElementTree . Element ( FAILURE_TAG )
    ElementTree . SubElement ( element , SASL_QNP + "invalid-authzid" )
    return True
authzid = success . properties . get ( "authzid" )
if authzid :
    peer = JID ( success . authzid )
elif "username" in success . properties :
    peer = JID ( success . properties [ "username" ] , stream . me . domain )
else : # anonymous
    peer = None
stream . set_peer_authenticated ( peer , True )


signature:
def _process_sasl_challenge ( self , stream , element ) :
    """Process incoming < sasl : challenge / > element .
 [ initiating entity only ]"""

implementation:
if not self . authenticator :
    logger . debug ( "Unexpected SASL challenge" )
    return False
content = element . text . encode ( "us-ascii" )
ret = self . authenticator . challenge ( a2b_base64 ( content ) )
if isinstance ( ret , sasl . Response ) :
    element = ElementTree . Element ( RESPONSE_TAG )
    element . text = ret . encode ( )
else :
    element = ElementTree . Element ( ABORT_TAG )
stream . write_element ( element )
if isinstance ( ret , sasl . Failure ) :
    stream . disconnect ( )
    raise SASLAuthenticationFailed ( "SASL authentication failed" )
return True


signature:
def _process_sasl_response ( self , stream , element ) :
    """Process incoming < sasl : response / > element .
 [ receiving entity only ]"""

implementation:
if not self . authenticator :
    logger . debug ( "Unexpected SASL response" )
    return False
content = element . text . encode ( "us-ascii" )
ret = self . authenticator . response ( a2b_base64 ( content ) )
if isinstance ( ret , sasl . Success ) :
    element = ElementTree . Element ( SUCCESS_TAG )
    element . text = ret . encode ( )
elif isinstance ( ret , sasl . Challenge ) :
    element = ElementTree . Element ( CHALLENGE_TAG )
    element . text = ret . encode ( )
else :
    element = ElementTree . Element ( FAILURE_TAG )
    ElementTree . SubElement ( element , SASL_QNP + ret . reason )
stream . write_element ( element )
if isinstance ( ret , sasl . Success ) :
    self . _handle_auth_success ( stream , ret )
elif isinstance ( ret , sasl . Failure ) :
    raise SASLAuthenticationFailed ( "SASL authentication failed: {0!r}" . format ( ret . reson ) )
return True


signature:
def _check_authorization ( self , properties , stream ) :
    """Check authorization id and other properties returned by the
 authentication mechanism .
 [ receiving entity only ]
 Allow only no authzid or authzid equal to current username @ domain
 FIXME : other rules in s2s
 : Parameters :
 - ` properties ` : data obtained during authentication
 : Types :
 - ` properties ` : mapping
 : return : ` True ` if user is authorized to use a provided authzid
 : returntype : ` bool `"""

implementation:
authzid = properties . get ( "authzid" )
if not authzid :
    return True
try :
    jid = JID ( authzid )
except ValueError :
    return False
if "username" not in properties :
    result = False
elif jid . local != properties [ "username" ] :
    result = False
elif jid . domain != stream . me . domain :
    result = False
elif jid . resource :
    result = False
else :
    result = True
return result


signature:
def _process_sasl_success ( self , stream , element ) :
    """Process incoming < sasl : success / > element .
 [ initiating entity only ]"""

implementation:
if not self . authenticator :
    logger . debug ( "Unexpected SASL response" )
    return False
content = element . text
if content :
    data = a2b_base64 ( content . encode ( "us-ascii" ) )
else :
    data = None
ret = self . authenticator . finish ( data )
if isinstance ( ret , sasl . Success ) :
    logger . debug ( "SASL authentication succeeded" )
    authzid = ret . properties . get ( "authzid" )
    if authzid :
        me = JID ( authzid )
    elif "username" in ret . properties : # FIXME : other rules for server
        me = JID ( ret . properties [ "username" ] , stream . peer . domain )
    else :
        me = None
    stream . set_authenticated ( me , True )
else :
    logger . debug ( "SASL authentication failed" )
    raise SASLAuthenticationFailed ( "Additional success data" " procesing failed" )
return True


signature:
def _process_sasl_failure ( self , stream , element ) :
    """Process incoming < sasl : failure / > element .
 [ initiating entity only ]"""

implementation:
_unused = stream
if not self . authenticator :
    logger . debug ( "Unexpected SASL response" )
    return False
logger . debug ( "SASL authentication failed: {0!r}" . format ( element_to_unicode ( element ) ) )
raise SASLAuthenticationFailed ( "SASL authentication failed" )


signature:
def _process_sasl_abort ( self , stream , element ) :
    """Process incoming < sasl : abort / > element .
 [ receiving entity only ]"""

implementation:
_unused , _unused = stream , element
if not self . authenticator :
    logger . debug ( "Unexpected SASL response" )
    return False
self . authenticator = None
logger . debug ( "SASL authentication aborted" )
return True


signature:
def _sasl_authenticate ( self , stream , username , authzid ) :
    """Start SASL authentication process .
 [ initiating entity only ]
 : Parameters :
 - ` username ` : user name .
 - ` authzid ` : authorization ID .
 - ` mechanism ` : SASL mechanism to use ."""

implementation:
if not stream . initiator :
    raise SASLAuthenticationFailed ( "Only initiating entity start" " SASL authentication" )
if stream . features is None or not self . peer_sasl_mechanisms :
    raise SASLNotAvailable ( "Peer doesn't support SASL" )
props = dict ( stream . auth_properties )
if not props . get ( "service-domain" ) and ( stream . peer and stream . peer . domain ) :
    props [ "service-domain" ] = stream . peer . domain
if username is not None :
    props [ "username" ] = username
if authzid is not None :
    props [ "authzid" ] = authzid
if "password" in self . settings :
    props [ "password" ] = self . settings [ "password" ]
props [ "available_mechanisms" ] = self . peer_sasl_mechanisms
enabled = sasl . filter_mechanism_list ( self . settings [ 'sasl_mechanisms' ] , props , self . settings [ 'insecure_auth' ] )
if not enabled :
    raise SASLNotAvailable ( "None of SASL mechanism selected can be used" )
props [ "enabled_mechanisms" ] = enabled
mechanism = None
for mech in enabled :
    if mech in self . peer_sasl_mechanisms :
        mechanism = mech
        break
if not mechanism :
    raise SASLMechanismNotAvailable ( "Peer doesn't support any of" " our SASL mechanisms" )
logger . debug ( "Our mechanism: {0!r}" . format ( mechanism ) )
stream . auth_method_used = mechanism
self . authenticator = sasl . client_authenticator_factory ( mechanism )
initial_response = self . authenticator . start ( props )
if not isinstance ( initial_response , sasl . Response ) :
    raise SASLAuthenticationFailed ( "SASL initiation failed" )
element = ElementTree . Element ( AUTH_TAG )
element . set ( "mechanism" , mechanism )
if initial_response . data :
    if initial_response . encode :
        element . text = initial_response . encode ( )
    else :
        element . text = initial_response . data
stream . write_element ( element )


signature:
def from_xml ( cls , element ) :
    """Make a RosterItem from an XML element .
 : Parameters :
 - ` element ` : the XML element
 : Types :
 - ` element ` : : etree : ` ElementTree . Element `
 : return : a freshly created roster item
 : returntype : ` cls `"""

implementation:
if element . tag != ITEM_TAG :
    raise ValueError ( "{0!r} is not a roster item" . format ( element ) )
try :
    jid = JID ( element . get ( "jid" ) )
except ValueError :
    raise BadRequestProtocolError ( u"Bad item JID" )
subscription = element . get ( "subscription" )
ask = element . get ( "ask" )
name = element . get ( "name" )
duplicate_group = False
groups = set ( )
for child in element :
    if child . tag != GROUP_TAG :
        continue
    group = child . text
    if group is None :
        group = u""
    if group in groups :
        duplicate_group = True
    else :
        groups . add ( group )
approved = element . get ( "approved" )
if approved == "true" :
    approved = True
elif approved in ( "false" , None ) :
    approved = False
else :
    logger . debug ( "RosterItem.from_xml: got unknown 'approved':" " {0!r}, changing to False" . format ( approved ) )
    approved = False
result = cls ( jid , name , groups , subscription , ask , approved )
result . _duplicate_group = duplicate_group
return result


signature:
def as_xml ( self , parent = None ) :
    """Make an XML element from self .
 : Parameters :
 - ` parent ` : Parent element
 : Types :
 - ` parent ` : : etree : ` ElementTree . Element `"""

implementation:
if parent is not None :
    element = ElementTree . SubElement ( parent , ITEM_TAG )
else :
    element = ElementTree . Element ( ITEM_TAG )
element . set ( "jid" , unicode ( self . jid ) )
if self . name is not None :
    element . set ( "name" , self . name )
if self . subscription is not None :
    element . set ( "subscription" , self . subscription )
if self . ask :
    element . set ( "ask" , self . ask )
if self . approved :
    element . set ( "approved" , "true" )
for group in self . groups :
    ElementTree . SubElement ( element , GROUP_TAG ) . text = group
return element


signature:
def _verify ( self , valid_subscriptions , fix ) :
    """Check if ` self ` is valid roster item .
 Valid item must have proper ` subscription ` and valid value for ' ask ' .
 : Parameters :
 - ` valid _ subscriptions ` : sequence of valid subscription values
 - ` fix ` : if ` True ` than replace invalid ' subscription ' and ' ask '
 values with the defaults
 : Types :
 - ` fix ` : ` bool `
 : Raise : ` ValueError ` if the item is invalid ."""

implementation:
if self . subscription not in valid_subscriptions :
    if fix :
        logger . debug ( "RosterItem.from_xml: got unknown 'subscription':" " {0!r}, changing to None" . format ( self . subscription ) )
        self . subscription = None
    else :
        raise ValueError ( "Bad 'subscription'" )
if self . ask not in ( None , u"subscribe" ) :
    if fix :
        logger . debug ( "RosterItem.from_xml: got unknown 'ask':" " {0!r}, changing to None" . format ( self . ask ) )
        self . ask = None
    else :
        raise ValueError ( "Bad 'ask'" )


signature:
def verify_roster_result ( self , fix = False ) :
    """Check if ` self ` is valid roster item .
 Valid item must have proper ` subscription ` value other than ' remove '
 and valid value for ' ask ' .
 : Parameters :
 - ` fix ` : if ` True ` than replace invalid ' subscription ' and ' ask '
 values with the defaults
 : Types :
 - ` fix ` : ` bool `
 : Raise : ` ValueError ` if the item is invalid ."""

implementation: self . _verify ( ( None , u"from" , u"to" , u"both" ) , fix )


signature:
def verify_roster_push ( self , fix = False ) :
    """Check if ` self ` is valid roster push item .
 Valid item must have proper ` subscription ` value other and valid value
 for ' ask ' .
 : Parameters :
 - ` fix ` : if ` True ` than replace invalid ' subscription ' and ' ask '
 values with the defaults
 : Types :
 - ` fix ` : ` bool `
 : Raise : ` ValueError ` if the item is invalid ."""

implementation: self . _verify ( ( None , u"from" , u"to" , u"both" , u"remove" ) , fix )


signature:
def verify_roster_set ( self , fix = False , settings = None ) :
    """Check if ` self ` is valid roster set item .
 For use on server to validate incoming roster sets .
 Valid item must have proper ` subscription ` value other and valid value
 for ' ask ' . The lengths of name and group names must fit the configured
 limits .
 : Parameters :
 - ` fix ` : if ` True ` than replace invalid ' subscription ' and ' ask '
 values with right defaults
 - ` settings ` : settings object providing the name limits
 : Types :
 - ` fix ` : ` bool `
 - ` settings ` : ` XMPPSettings `
 : Raise : ` BadRequestProtocolError ` if the item is invalid ."""

implementation:
# pylint : disable = R0912
try :
    self . _verify ( ( None , u"remove" ) , fix )
except ValueError , err :
    raise BadRequestProtocolError ( unicode ( err ) )
if self . ask :
    if fix :
        self . ask = None
    else :
        raise BadRequestProtocolError ( "'ask' in roster set" )
if self . approved :
    if fix :
        self . approved = False
    else :
        raise BadRequestProtocolError ( "'approved' in roster set" )
if settings is None :
    settings = XMPPSettings ( )
name_length_limit = settings [ "roster_name_length_limit" ]
if self . name and len ( self . name ) > name_length_limit :
    raise NotAcceptableProtocolError ( u"Roster item name too long" )
group_length_limit = settings [ "roster_group_name_length_limit" ]
for group in self . groups :
    if not group :
        raise NotAcceptableProtocolError ( u"Roster group name empty" )
    if len ( group ) > group_length_limit :
        raise NotAcceptableProtocolError ( u"Roster group name too long" )
if self . _duplicate_group :
    raise BadRequestProtocolError ( u"Item group duplicated" )


signature:
def groups ( self ) :
    """Set of groups defined in the roster .
 : Return : the groups
 : ReturnType : ` set ` of ` unicode `"""

implementation:
groups = set ( )
for item in self . _items :
    groups |= item . groups
return groups


signature:
def get_items_by_name ( self , name , case_sensitive = True ) :
    """Return a list of items with given name .
 : Parameters :
 - ` name ` : name to look - up
 - ` case _ sensitive ` : if ` False ` the matching will be case
 insensitive .
 : Types :
 - ` name ` : ` unicode `
 - ` case _ sensitive ` : ` bool `
 : Returntype : ` list ` of ` RosterItem `"""

implementation:
if not case_sensitive and name :
    name = name . lower ( )
result = [ ]
for item in self . _items :
    if item . name == name :
        result . append ( item )
    elif item . name is None :
        continue
    elif not case_sensitive and item . name . lower ( ) == name :
        result . append ( item )
return result


signature:
def get_items_by_group ( self , group , case_sensitive = True ) :
    """Return a list of items within a given group .
 : Parameters :
 - ` name ` : name to look - up
 - ` case _ sensitive ` : if ` False ` the matching will be case
 insensitive .
 : Types :
 - ` name ` : ` unicode `
 - ` case _ sensitive ` : ` bool `
 : Returntype : ` list ` of ` RosterItem `"""

implementation:
result = [ ]
if not group :
    for item in self . _items :
        if not item . groups :
            result . append ( item )
    return result
if not case_sensitive :
    group = group . lower ( )
for item in self . _items :
    if group in item . groups :
        result . append ( item )
    elif not case_sensitive and group in [ g . lower ( ) for g in item . groups ] :
        result . append ( item )
return result


signature:
def add_item ( self , item , replace = False ) :
    """Add an item to the roster .
 This will not automatically update the roster on the server .
 : Parameters :
 - ` item ` : the item to add
 - ` replace ` : if ` True ` then existing item will be replaced ,
 otherwise a ` ValueError ` will be raised on conflict
 : Types :
 - ` item ` : ` RosterItem `
 - ` replace ` : ` bool `"""

implementation:
if item . jid in self . _jids :
    if replace :
        self . remove_item ( item . jid )
    else :
        raise ValueError ( "JID already in the roster" )
index = len ( self . _items )
self . _items . append ( item )
self . _jids [ item . jid ] = index


signature:
def remove_item ( self , jid ) :
    """Remove item from the roster .
 : Parameters :
 - ` jid ` : JID of the item to remove
 : Types :
 - ` jid ` : ` JID `"""

implementation:
if jid not in self . _jids :
    raise KeyError ( jid )
index = self . _jids [ jid ]
for i in range ( index , len ( self . _jids ) ) :
    self . _jids [ self . _items [ i ] . jid ] -= 1
del self . _jids [ jid ]
del self . _items [ index ]


signature:
def load_roster ( self , source ) :
    """Load roster from an XML file .
 Can be used before the connection is started to load saved
 roster copy , for efficient retrieval of versioned roster .
 : Parameters :
 - ` source ` : file name or a file object
 : Types :
 - ` source ` : ` str ` or file - like object"""

implementation:
try :
    tree = ElementTree . parse ( source )
except ElementTree . ParseError , err :
    raise ValueError ( "Invalid roster format: {0}" . format ( err ) )
roster = Roster . from_xml ( tree . getroot ( ) )
for item in roster :
    item . verify_roster_result ( True )
self . roster = roster


signature:
def save_roster ( self , dest , pretty = True ) :
    """Save the roster to an XML file .
 Can be used to save the last know roster copy for faster loading
 of a verisoned roster ( if server supports that ) .
 : Parameters :
 - ` dest ` : file name or a file object
 - ` pretty ` : pretty - format the roster XML
 : Types :
 - ` dest ` : ` str ` or file - like object
 - ` pretty ` : ` bool `"""

implementation:
if self . roster is None :
    raise ValueError ( "No roster" )
element = self . roster . as_xml ( )
if pretty :
    if len ( element ) :
        element . text = u'\n  '
    p_child = None
    for child in element :
        if p_child is not None :
            p_child . tail = u'\n  '
        if len ( child ) :
            child . text = u'\n    '
        p_grand = None
        for grand in child :
            if p_grand is not None :
                p_grand . tail = u'\n    '
            p_grand = grand
        if p_grand is not None :
            p_grand . tail = u'\n  '
        p_child = child
    if p_child is not None :
        p_child . tail = u"\n"
tree = ElementTree . ElementTree ( element )
tree . write ( dest , "utf-8" )


signature:
def handle_got_features_event ( self , event ) :
    """Check for roster related features in the stream features received
 and set ` server _ features ` accordingly ."""

implementation:
server_features = set ( )
logger . debug ( "Checking roster-related features" )
if event . features . find ( FEATURE_ROSTERVER ) is not None :
    logger . debug ( "  Roster versioning available" )
    server_features . add ( "versioning" )
if event . features . find ( FEATURE_APPROVALS ) is not None :
    logger . debug ( "  Subscription pre-approvals available" )
    server_features . add ( "pre-approvals" )
self . server_features = server_features


signature:
def handle_authorized_event ( self , event ) :
    """Request roster upon login ."""

implementation:
self . server = event . authorized_jid . bare ( )
if "versioning" in self . server_features :
    if self . roster is not None and self . roster . version is not None :
        version = self . roster . version
    else :
        version = u""
else :
    version = None
self . request_roster ( version )


signature:
def request_roster ( self , version = None ) :
    """Request roster from server .
 : Parameters :
 - ` version ` : if not ` None ` versioned roster will be requested
 for given local version . Use " " to request full roster .
 : Types :
 - ` version ` : ` unicode `"""

implementation:
processor = self . stanza_processor
request = Iq ( stanza_type = "get" )
request . set_payload ( RosterPayload ( version = version ) )
processor . set_response_handlers ( request , self . _get_success , self . _get_error )
processor . send ( request )


signature:
def _get_success ( self , stanza ) :
    """Handle successful response to the roster request ."""

implementation:
payload = stanza . get_payload ( RosterPayload )
if payload is None :
    if "versioning" in self . server_features and self . roster :
        logger . debug ( "Server will send roster delta in pushes" )
    else :
        logger . warning ( "Bad roster response (no payload)" )
        self . _event_queue . put ( RosterNotReceivedEvent ( self , stanza ) )
        return
else :
    items = list ( payload )
    for item in items :
        item . verify_roster_result ( True )
    self . roster = Roster ( items , payload . version )
self . _event_queue . put ( RosterReceivedEvent ( self , self . roster ) )


signature:
def _get_error ( self , stanza ) :
    """Handle failure of the roster request ."""

implementation:
if stanza :
    logger . debug ( u"Roster request failed: {0}" . format ( stanza . error . condition_name ) )
else :
    logger . debug ( u"Roster request failed: timeout" )
self . _event_queue . put ( RosterNotReceivedEvent ( self , stanza ) )


signature:
def handle_roster_push ( self , stanza ) :
    """Handle a roster push received from server ."""

implementation:
if self . server is None and stanza . from_jid :
    logger . debug ( u"Server address not known, cannot verify roster push" " from {0}" . format ( stanza . from_jid ) )
    return stanza . make_error_response ( u"service-unavailable" )
if self . server and stanza . from_jid and stanza . from_jid != self . server :
    logger . debug ( u"Roster push from invalid source: {0}" . format ( stanza . from_jid ) )
    return stanza . make_error_response ( u"service-unavailable" )
payload = stanza . get_payload ( RosterPayload )
if len ( payload ) != 1 :
    logger . warning ( "Bad roster push received ({0} items)" . format ( len ( payload ) ) )
    return stanza . make_error_response ( u"bad-request" )
if self . roster is None :
    logger . debug ( "Dropping roster push - no roster here" )
    return True
item = payload [ 0 ]
item . verify_roster_push ( True )
old_item = self . roster . get ( item . jid )
if item . subscription == "remove" :
    if old_item :
        self . roster . remove_item ( item . jid )
else :
    self . roster . add_item ( item , replace = True )
self . _event_queue . put ( RosterUpdatedEvent ( self , old_item , item ) )
return stanza . make_result_response ( )


signature:
def add_item ( self , jid , name = None , groups = None , callback = None , error_callback = None ) :
    """Add a contact to the roster .
 : Parameters :
 - ` jid ` : contact ' s jid
 - ` name ` : name for the contact
 - ` groups ` : sequence of group names the contact should belong to
 - ` callback ` : function to call when the request succeeds . It should
 accept a single argument - a ` RosterItem ` describing the
 requested change
 - ` error _ callback ` : function to call when the request fails . It
 should accept a single argument - an error stanza received
 ( ` None ` in case of timeout )
 : Types :
 - ` jid ` : ` JID `
 - ` name ` : ` unicode `
 - ` groups ` : sequence of ` unicode `"""

implementation:
# pylint : disable = R0913
if jid in self . roster :
    raise ValueError ( "{0!r} already in the roster" . format ( jid ) )
item = RosterItem ( jid , name , groups )
self . _roster_set ( item , callback , error_callback )


signature:
def update_item ( self , jid , name = NO_CHANGE , groups = NO_CHANGE , callback = None , error_callback = None ) :
    """Modify a contact in the roster .
 : Parameters :
 - ` jid ` : contact ' s jid
 - ` name ` : a new name for the contact
 - ` groups ` : a sequence of group names the contact should belong to
 - ` callback ` : function to call when the request succeeds . It should
 accept a single argument - a ` RosterItem ` describing the
 requested change
 - ` error _ callback ` : function to call when the request fails . It
 should accept a single argument - an error stanza received
 ( ` None ` in case of timeout )
 : Types :
 - ` jid ` : ` JID `
 - ` name ` : ` unicode `
 - ` groups ` : sequence of ` unicode `"""

implementation:
# pylint : disable = R0913
item = self . roster [ jid ]
if name is NO_CHANGE and groups is NO_CHANGE :
    return
if name is NO_CHANGE :
    name = item . name
if groups is NO_CHANGE :
    groups = item . groups
item = RosterItem ( jid , name , groups )
self . _roster_set ( item , callback , error_callback )


signature:
def remove_item ( self , jid , callback = None , error_callback = None ) :
    """Remove a contact from the roster .
 : Parameters :
 - ` jid ` : contact ' s jid
 - ` callback ` : function to call when the request succeeds . It should
 accept a single argument - a ` RosterItem ` describing the
 requested change
 - ` error _ callback ` : function to call when the request fails . It
 should accept a single argument - an error stanza received
 ( ` None ` in case of timeout )
 : Types :
 - ` jid ` : ` JID `"""

implementation:
item = self . roster [ jid ]
if jid not in self . roster :
    raise KeyError ( jid )
item = RosterItem ( jid , subscription = "remove" )
self . _roster_set ( item , callback , error_callback )


signature:
def free ( self ) :
    """Unlink and free the XML node owned by ` self ` ."""

implementation:
if not self . borrowed :
    self . xmlnode . unlinkNode ( )
    self . xmlnode . freeNode ( )
self . xmlnode = None


signature:
def xpath_eval ( self , expr ) :
    """Evaluate XPath expression in context of ` self . xmlnode ` .
 : Parameters :
 - ` expr ` : the XPath expression
 : Types :
 - ` expr ` : ` unicode `
 : return : the result of the expression evaluation .
 : returntype : list of ` libxml2 . xmlNode `"""

implementation:
ctxt = common_doc . xpathNewContext ( )
ctxt . setContextNode ( self . xmlnode )
ctxt . xpathRegisterNs ( "muc" , self . ns . getContent ( ) )
ret = ctxt . xpathEval ( to_utf8 ( expr ) )
ctxt . xpathFreeContext ( )
return ret


signature:
def set_history ( self , parameters ) :
    """Set history parameters .
 Types :
 - ` parameters ` : ` HistoryParameters `"""

implementation:
for child in xml_element_iter ( self . xmlnode . children ) :
    if get_node_ns_uri ( child ) == MUC_NS and child . name == "history" :
        child . unlinkNode ( )
        child . freeNode ( )
        break
if parameters . maxchars and parameters . maxchars < 0 :
    raise ValueError ( "History parameter maxchars must be positive" )
if parameters . maxstanzas and parameters . maxstanzas < 0 :
    raise ValueError ( "History parameter maxstanzas must be positive" )
if parameters . maxseconds and parameters . maxseconds < 0 :
    raise ValueError ( "History parameter maxseconds must be positive" )
hnode = self . xmlnode . newChild ( self . xmlnode . ns ( ) , "history" , None )
if parameters . maxchars is not None :
    hnode . setProp ( "maxchars" , str ( parameters . maxchars ) )
if parameters . maxstanzas is not None :
    hnode . setProp ( "maxstanzas" , str ( parameters . maxstanzas ) )
if parameters . maxseconds is not None :
    hnode . setProp ( "maxseconds" , str ( parameters . maxseconds ) )
if parameters . since is not None :
    hnode . setProp ( "since" , parameters . since . strftime ( "%Y-%m-%dT%H:%M:%SZ" ) )


signature:
def get_history ( self ) :
    """Return history parameters carried by the stanza .
 : returntype : ` HistoryParameters `"""

implementation:
for child in xml_element_iter ( self . xmlnode . children ) :
    if get_node_ns_uri ( child ) == MUC_NS and child . name == "history" :
        maxchars = from_utf8 ( child . prop ( "maxchars" ) )
        if maxchars is not None :
            maxchars = int ( maxchars )
        maxstanzas = from_utf8 ( child . prop ( "maxstanzas" ) )
        if maxstanzas is not None :
            maxstanzas = int ( maxstanzas )
        maxseconds = from_utf8 ( child . prop ( "maxseconds" ) )
        if maxseconds is not None :
            maxseconds = int ( maxseconds )
        # TODO : since - - requires parsing of Jabber dateTime profile
        since = None
        return HistoryParameters ( maxchars , maxstanzas , maxseconds , since )


signature:
def set_password ( self , password ) :
    """Set password for the MUC request .
 : Parameters :
 - ` password ` : password
 : Types :
 - ` password ` : ` unicode `"""

implementation:
for child in xml_element_iter ( self . xmlnode . children ) :
    if get_node_ns_uri ( child ) == MUC_NS and child . name == "password" :
        child . unlinkNode ( )
        child . freeNode ( )
        break
if password is not None :
    self . xmlnode . newTextChild ( self . xmlnode . ns ( ) , "password" , to_utf8 ( password ) )


signature:
def get_password ( self ) :
    """Get password from the MUC request .
 : returntype : ` unicode `"""

implementation:
for child in xml_element_iter ( self . xmlnode . children ) :
    if get_node_ns_uri ( child ) == MUC_NS and child . name == "password" :
        return from_utf8 ( child . getContent ( ) )
return None


signature:
def __init ( self , affiliation , role , jid = None , nick = None , actor = None , reason = None ) :
    """Initialize a ` MucItem ` object from a set of attributes .
 : Parameters :
 - ` affiliation ` : affiliation of the user .
 - ` role ` : role of the user .
 - ` jid ` : JID of the user .
 - ` nick ` : nickname of the user .
 - ` actor ` : actor modyfying the user data .
 - ` reason ` : reason of change of the user data .
 : Types :
 - ` affiliation ` : ` str `
 - ` role ` : ` str `
 - ` jid ` : ` JID `
 - ` nick ` : ` unicode `
 - ` actor ` : ` JID `
 - ` reason ` : ` unicode `"""

implementation:
if not affiliation :
    affiliation = None
elif affiliation not in affiliations :
    raise ValueError ( "Bad affiliation" )
self . affiliation = affiliation
if not role :
    role = None
elif role not in roles :
    raise ValueError ( "Bad role" )
self . role = role
if jid :
    self . jid = JID ( jid )
else :
    self . jid = None
if actor :
    self . actor = JID ( actor )
else :
    self . actor = None
self . nick = nick
self . reason = reason


signature:
def __from_xmlnode ( self , xmlnode ) :
    """Initialize a ` MucItem ` object from an XML node .
 : Parameters :
 - ` xmlnode ` : the XML node .
 : Types :
 - ` xmlnode ` : ` libxml2 . xmlNode `"""

implementation:
actor = None
reason = None
n = xmlnode . children
while n :
    ns = n . ns ( )
    if ns and ns . getContent ( ) != MUC_USER_NS :
        continue
    if n . name == "actor" :
        actor = n . getContent ( )
    if n . name == "reason" :
        reason = n . getContent ( )
    n = n . next
self . __init ( from_utf8 ( xmlnode . prop ( "affiliation" ) ) , from_utf8 ( xmlnode . prop ( "role" ) ) , from_utf8 ( xmlnode . prop ( "jid" ) ) , from_utf8 ( xmlnode . prop ( "nick" ) ) , from_utf8 ( actor ) , from_utf8 ( reason ) , ) ;


signature:
def as_xml ( self , parent ) :
    """Create XML representation of ` self ` .
 : Parameters :
 - ` parent ` : the element to which the created node should be linked to .
 : Types :
 - ` parent ` : ` libxml2 . xmlNode `
 : return : an XML node .
 : returntype : ` libxml2 . xmlNode `"""

implementation:
n = parent . newChild ( None , "item" , None )
if self . actor :
    n . newTextChild ( None , "actor" , to_utf8 ( self . actor ) )
if self . reason :
    n . newTextChild ( None , "reason" , to_utf8 ( self . reason ) )
n . setProp ( "affiliation" , to_utf8 ( self . affiliation ) )
if self . role :
    n . setProp ( "role" , to_utf8 ( self . role ) )
if self . jid :
    n . setProp ( "jid" , to_utf8 ( self . jid . as_unicode ( ) ) )
if self . nick :
    n . setProp ( "nick" , to_utf8 ( self . nick ) )
return n