msaf/input_output.py
def save_estimations(file_struct, times, labels, boundaries_id, labels_id,
                     **params):
    """Saves the segment estimations in a JAMS file.

    Parameters
    ----------
    file_struct : FileStruct
        Object with the different file paths of the current file.
    times : np.array or list
        Estimated boundary times.
        If `list`, estimated hierarchical boundaries.
    labels : np.array(N, 2)
        Estimated labels (None in case we are only storing boundary
        evaluations).
    boundaries_id : str
        Boundary algorithm identifier.
    labels_id : str
        Labels algorithm identifier.
    params : dict
        Dictionary with additional parameters for both algorithms.
    """
    # Remove features if they exist
    params.pop("features", None)

    # Get duration
    dur = get_duration(file_struct.features_file)

    # Convert to intervals and sanity check
    if 'numpy' in str(type(times)):
        # Flat check
        inters = utils.times_to_intervals(times)
        assert len(inters) == len(labels), "Number of boundary intervals " \
            "(%d) and labels (%d) do not match" % (len(inters), len(labels))
        # Put into lists to simplify the writing process later
        inters = [inters]
        labels = [labels]
    else:
        # Hierarchical check
        inters = []
        for level in range(len(times)):
            est_inters = utils.times_to_intervals(times[level])
            inters.append(est_inters)
            assert len(inters[level]) == len(labels[level]), \
                "Number of boundary intervals (%d) and labels (%d) do not " \
                "match in level %d" % (len(inters[level]), len(labels[level]),
                                       level)

    # Create new estimation
    namespace = "multi_segment" if params["hier"] else "segment_open"
    ann = jams.Annotation(namespace=namespace)

    # Find estimation in file
    if os.path.isfile(file_struct.est_file):
        jam = jams.load(file_struct.est_file, validate=False)
        curr_ann = find_estimation(jam, boundaries_id, labels_id, params)
        if curr_ann is not None:
            curr_ann.data = ann.data  # cleanup all data
            ann = curr_ann  # This will overwrite the existing estimation
        else:
            jam.annotations.append(ann)
    else:
        # Create new JAMS if it doesn't exist
        jam = jams.JAMS()
        jam.file_metadata.duration = dur
        jam.annotations.append(ann)

    # Save metadata and parameters
    ann.annotation_metadata.version = msaf.__version__
    ann.annotation_metadata.data_source = "MSAF"
    sandbox = {}
    sandbox["boundaries_id"] = boundaries_id
    sandbox["labels_id"] = labels_id
    sandbox["timestamp"] = \
        datetime.datetime.today().strftime("%Y/%m/%d %H:%M:%S")
    for key in params:
        sandbox[key] = params[key]
    ann.sandbox = sandbox

    # Save actual data
    for i, (level_inters, level_labels) in enumerate(zip(inters, labels)):
        for bound_inter, label in zip(level_inters, level_labels):
            dur = float(bound_inter[1]) - float(bound_inter[0])
            label = chr(int(label) + 65)
            if params["hier"]:
                value = {"label": label, "level": i}
            else:
                value = label
            ann.append(time=bound_inter[0], duration=dur,
                       value=value)

    # Write results
    jam.save(file_struct.est_file)
def save_estimations(file_struct, times, labels, boundaries_id, labels_id,
                     **params):
    # Remove features if they exist
    params.pop("features", None)

    # Get duration
    dur = get_duration(file_struct.features_file)

    # Convert to intervals and sanity check
    if 'numpy' in str(type(times)):
        # Flat check
        inters = utils.times_to_intervals(times)
        assert len(inters) == len(labels), "Number of boundary intervals " \
            "(%d) and labels (%d) do not match" % (len(inters), len(labels))
        # Put into lists to simplify the writing process later
        inters = [inters]
        labels = [labels]
    else:
        # Hierarchical check
        inters = []
        for level in range(len(times)):
            est_inters = utils.times_to_intervals(times[level])
            inters.append(est_inters)
            assert len(inters[level]) == len(labels[level]), \
                "Number of boundary intervals (%d) and labels (%d) do not " \
                "match in level %d" % (len(inters[level]), len(labels[level]),
                                       level)

    # Create new estimation
    namespace = "multi_segment" if params["hier"] else "segment_open"
    ann = jams.Annotation(namespace=namespace)

    # Find estimation in file
    if os.path.isfile(file_struct.est_file):
        jam = jams.load(file_struct.est_file, validate=False)
        curr_ann = find_estimation(jam, boundaries_id, labels_id, params)
        if curr_ann is not None:
            curr_ann.data = ann.data  # cleanup all data
            ann = curr_ann  # This will overwrite the existing estimation
        else:
            jam.annotations.append(ann)
    else:
        # Create new JAMS if it doesn't exist
        jam = jams.JAMS()
        jam.file_metadata.duration = dur
        jam.annotations.append(ann)

    # Save metadata and parameters
    ann.annotation_metadata.version = msaf.__version__
    ann.annotation_metadata.data_source = "MSAF"
    sandbox = {}
    sandbox["boundaries_id"] = boundaries_id
    sandbox["labels_id"] = labels_id
    sandbox["timestamp"] = \
        datetime.datetime.today().strftime("%Y/%m/%d %H:%M:%S")
    for key in params:
        sandbox[key] = params[key]
    ann.sandbox = sandbox

    # Save actual data
    for i, (level_inters, level_labels) in enumerate(zip(inters, labels)):
        for bound_inter, label in zip(level_inters, level_labels):
            dur = float(bound_inter[1]) - float(bound_inter[0])
            label = chr(int(label) + 65)
            if params["hier"]:
                value = {"label": label, "level": i}
            else:
                value = label
            ann.append(time=bound_inter[0], duration=dur,
                       value=value)

    # Write results
    jam.save(file_struct.est_file)
def save_estimations ( file_struct , times , labels , boundaries_id , labels_id , * * params ) : # Remove features if they exist params . pop ( "features" , None ) # Get duration dur = get_duration ( file_struct . features_file ) # Convert to intervals and sanity check if 'numpy' in str ( type ( times ) ) : # Flat check inters = utils . times_to_intervals ( times ) assert len ( inters ) == len ( labels ) , "Number of boundary intervals " "(%d) and labels (%d) do not match" % ( len ( inters ) , len ( labels ) ) # Put into lists to simplify the writing process later inters = [ inters ] labels = [ labels ] else : # Hierarchical check inters = [ ] for level in range ( len ( times ) ) : est_inters = utils . times_to_intervals ( times [ level ] ) inters . append ( est_inters ) assert len ( inters [ level ] ) == len ( labels [ level ] ) , "Number of boundary intervals (%d) and labels (%d) do not " "match in level %d" % ( len ( inters [ level ] ) , len ( labels [ level ] ) , level ) # Create new estimation namespace = "multi_segment" if params [ "hier" ] else "segment_open" ann = jams . Annotation ( namespace = namespace ) # Find estimation in file if os . path . isfile ( file_struct . est_file ) : jam = jams . load ( file_struct . est_file , validate = False ) curr_ann = find_estimation ( jam , boundaries_id , labels_id , params ) if curr_ann is not None : curr_ann . data = ann . data # cleanup all data ann = curr_ann # This will overwrite the existing estimation else : jam . annotations . append ( ann ) else : # Create new JAMS if it doesn't exist jam = jams . JAMS ( ) jam . file_metadata . duration = dur jam . annotations . append ( ann ) # Save metadata and parameters ann . annotation_metadata . version = msaf . __version__ ann . annotation_metadata . data_source = "MSAF" sandbox = { } sandbox [ "boundaries_id" ] = boundaries_id sandbox [ "labels_id" ] = labels_id sandbox [ "timestamp" ] = datetime . datetime . today ( ) . strftime ( "%Y/%m/%d %H:%M:%S" ) for key in params : sandbox [ key ] = params [ key ] ann . sandbox = sandbox # Save actual data for i , ( level_inters , level_labels ) in enumerate ( zip ( inters , labels ) ) : for bound_inter , label in zip ( level_inters , level_labels ) : dur = float ( bound_inter [ 1 ] ) - float ( bound_inter [ 0 ] ) label = chr ( int ( label ) + 65 ) if params [ "hier" ] : value = { "label" : label , "level" : i } else : value = label ann . append ( time = bound_inter [ 0 ] , duration = dur , value = value ) # Write results jam . save ( file_struct . est_file )
Saves the segment estimations in a JAMS file.

    Parameters
    ----------
    file_struct : FileStruct
        Object with the different file paths of the current file.
    times : np.array or list
        Estimated boundary times.
        If `list`, estimated hierarchical boundaries.
    labels : np.array(N, 2)
        Estimated labels (None in case we are only storing boundary
        evaluations).
    boundaries_id : str
        Boundary algorithm identifier.
    labels_id : str
        Labels algorithm identifier.
    params : dict
        Dictionary with additional parameters for both algorithms.
Saves the segment estimations in a JAMS file .