<commit_before>"""
This module contains multiple functions in order to run MSAF algorithms.
"""
import jams
import librosa
import logging
import numpy as np
import os

from joblib import Parallel, delayed

import msaf
from msaf import input_output as io
from msaf import utils
from msaf import plotting
from msaf.input_output import FileStruct
from msaf.features import Features
import msaf.algorithms as algorithms


def get_boundaries_module(boundaries_id):
    """Obtains the boundaries module given a boundary algorithm identificator.

    Parameters
    ----------
    boundaries_id: str
        Boundary algorithm identificator (e.g., foote, sf).

    Returns
    -------
    module: object
        Object containing the selected boundary module.
        None for "ground truth".
    """
    if boundaries_id == "gt":
        return None
    try:
        module = eval(algorithms.__name__ + "." + boundaries_id)
    except AttributeError:
        raise RuntimeError("Algorithm %s can not be found in msaf!" %
                           boundaries_id)
    if not module.is_boundary_type:
        raise RuntimeError("Algorithm %s can not identify boundaries!" %
                           boundaries_id)
    return module


def get_labels_module(labels_id):
    """Obtains the label module given a label algorithm identificator.

    Parameters
    ----------
    labels_id: str
        Label algorithm identificator (e.g., fmc2d, cnmf).

    Returns
    -------
    module: object
        Object containing the selected label module.
        None for not computing the labeling part of music segmentation.
    """
    if labels_id is None:
        return None
    try:
        module = eval(algorithms.__name__ + "." + labels_id)
    except AttributeError:
        raise RuntimeError("Algorithm %s can not be found in msaf!" %
                           labels_id)
    if not module.is_label_type:
        raise RuntimeError("Algorithm %s can not label segments!" %
                           labels_id)
    return module


def run_hierarchical(audio_file, bounds_module, labels_module, frame_times,
                     config, annotator_id=0):
    """Runs hierarchical algorithms with the specified identifiers on the
    audio_file. See run_algorithm for more information.
    """
    # Get features to make code nicer
    features = config["features"]

    if bounds_module is None:
        raise RuntimeError("A boundary algorithm is needed when using "
                           "hierarchical segmentation.")
    if labels_module is not None and \
            bounds_module.__name__ != labels_module.__name__:
        raise RuntimeError("The same algorithm for boundaries and labels is "
                           "needed when using hierarchical segmentation.")
    S = bounds_module.Segmenter(audio_file, **config)
    est_idxs, est_labels = S.processHierarchical()

    # Make sure the first and last boundaries are included for each
    # level in the hierarchy
    est_times = []
    cleaned_est_labels = []
    for level in range(len(est_idxs)):
        est_level_times, est_level_labels = \
            utils.process_segmentation_level(
                est_idxs[level], est_labels[level], features["pcp"].shape[0],
                frame_times, features["anal"]["dur"])
        est_times.append(est_level_times)
        cleaned_est_labels.append(est_level_labels)
    est_labels = cleaned_est_labels

    return est_times, est_labels


def run_flat(audio_file, bounds_module, labels_module, frame_times, config,
             annotator_id):
    """Runs the flat algorithms with the specified identifiers on the
    audio_file. See run_algorithm for more information.
    """
    # Get features to make code nicer
    features = config["features"].features

    # Segment using the specified boundaries and labels
    # Case when boundaries and labels algorithms are the same
    if bounds_module is not None and labels_module is not None and \
            bounds_module.__name__ == labels_module.__name__:
        S = bounds_module.Segmenter(audio_file, **config)
        est_idxs, est_labels = S.processFlat()
    # Different boundary and label algorithms
    else:
        # Identify segment boundaries
        if bounds_module is not None:
            S = bounds_module.Segmenter(audio_file, in_labels=[], **config)
            est_idxs, est_labels = S.processFlat()
        else:
            try:
                est_times, est_labels = io.read_references(
                    audio_file, annotator_id=annotator_id)
                est_idxs = io.align_times(est_times, frame_times[:-1])
                if est_idxs[0] != 0:
                    est_idxs = np.concatenate(([0], est_idxs))
                if est_idxs[-1] != features.shape[0] - 1:
                    est_idxs = np.concatenate((
                        est_idxs, [features.shape[0] - 1]))
            except:
                logging.warning("No references found for file: %s" %
                                audio_file)
                return [], []

        # Label segments
        if labels_module is not None:
            if len(est_idxs) == 2:
                est_labels = np.array([0])
            else:
                S = labels_module.Segmenter(audio_file,
                                            in_bound_idxs=est_idxs,
                                            **config)
                est_labels = S.processFlat()[1]

    # Make sure the first and last boundaries are included
    est_times, est_labels = utils.process_segmentation_level(
        est_idxs, est_labels, features.shape[0], frame_times,
        config["features"].dur)

    return est_times, est_labels


def run_algorithms(audio_file, boundaries_id, labels_id, config,
                   annotator_id=0):
    """Runs the algorithms with the specified identifiers on the audio_file.

    Parameters
    ----------
    audio_file: str
        Path to the audio file to segment.
    boundaries_id: str
        Identifier of the boundaries algorithm to use ("gt" for ground truth).
    labels_id: str
        Identifier of the labels algorithm to use (None for not labeling).
    config: dict
        Dictionary containing the custom parameters of the algorithms to use.
    annotator_id: int
        Annotator identificator in the ground truth.

    Returns
    -------
    est_times: np.array or list
        List of estimated times for the segment boundaries.
        If `list`, it will be a list of np.arrays, sorted by segmentation
        layer.
    est_labels: np.array or list
        List of all the labels associated segments.
        If `list`, it will be a list of np.arrays, sorted by segmentation
        layer.
    """
    # Check that there are enough audio frames
    if config["features"].features.shape[0] <= msaf.minimum__frames:
        logging.warning("Audio file too short, or too many few beats "
                        "estimated. Returning empty estimations.")
        return np.asarray([0, features["anal"]["dur"]]), \
            np.asarray([0], dtype=int)

    # Get the corresponding modules
    bounds_module = get_boundaries_module(boundaries_id)
    labels_module = get_labels_module(labels_id)

    # Get the correct frame times
    frame_times = config["features"].frame_times

    # Segment audio based on type of segmentation
    run_fun = run_hierarchical if config["hier"] else run_flat<commit_msg>Fixing little bugs from old behavior<commit_after>"""
This module contains multiple functions in order to run MSAF algorithms.
"""
import jams
import librosa
import logging
import numpy as np
import os

from joblib import Parallel, delayed

import msaf
from msaf import input_output as io
from msaf import utils
from msaf import plotting
from msaf.input_output import FileStruct
from msaf.features import Features
import msaf.algorithms as algorithms


def get_boundaries_module(boundaries_id):
    """Obtains the boundaries module given a boundary algorithm identificator.

    Parameters
    ----------
    boundaries_id: str
        Boundary algorithm identificator (e.g., foote, sf).

    Returns
    -------
    module: object
        Object containing the selected boundary module.
        None for "ground truth".
    """
    if boundaries_id == "gt":
        return None
    try:
        module = eval(algorithms.__name__ + "." + boundaries_id)
    except AttributeError:
        raise RuntimeError("Algorithm %s can not be found in msaf!" %
                           boundaries_id)
    if not module.is_boundary_type:
        raise RuntimeError("Algorithm %s can not identify boundaries!" %
                           boundaries_id)
    return module


def get_labels_module(labels_id):
    """Obtains the label module given a label algorithm identificator.

    Parameters
    ----------
    labels_id: str
        Label algorithm identificator (e.g., fmc2d, cnmf).

    Returns
    -------
    module: object
        Object containing the selected label module.
        None for not computing the labeling part of music segmentation.
    """
    if labels_id is None:
        return None
    try:
        module = eval(algorithms.__name__ + "." + labels_id)
    except AttributeError:
        raise RuntimeError("Algorithm %s can not be found in msaf!" %
                           labels_id)
    if not module.is_label_type:
        raise RuntimeError("Algorithm %s can not label segments!" %
                           labels_id)
    return module


def run_hierarchical(audio_file, bounds_module, labels_module, frame_times,
                     config, annotator_id=0):
    """Runs hierarchical algorithms with the specified identifiers on the
    audio_file. See run_algorithm for more information.
    """
    # Get features to make code nicer
    features = config["features"].features

    if bounds_module is None:
        raise RuntimeError("A boundary algorithm is needed when using "
                           "hierarchical segmentation.")
    if labels_module is not None and \
            bounds_module.__name__ != labels_module.__name__:
        raise RuntimeError("The same algorithm for boundaries and labels is "
                           "needed when using hierarchical segmentation.")
    S = bounds_module.Segmenter(audio_file, **config)
    est_idxs, est_labels = S.processHierarchical()

    # Make sure the first and last boundaries are included for each
    # level in the hierarchy
    est_times = []
    cleaned_est_labels = []
    for level in range(len(est_idxs)):
        est_level_times, est_level_labels = \
            utils.process_segmentation_level(
                est_idxs[level], est_labels[level], features.shape[0],
                frame_times, features.dur)
        est_times.append(est_level_times)
        cleaned_est_labels.append(est_level_labels)
    est_labels = cleaned_est_labels

    return est_times, est_labels


def run_flat(audio_file, bounds_module, labels_module, frame_times, config,
             annotator_id):
    """Runs the flat algorithms with the specified identifiers on the
    audio_file. See run_algorithm for more information.
    """
    # Get features to make code nicer
    features = config["features"].features

    # Segment using the specified boundaries and labels
    # Case when boundaries and labels algorithms are the same
    if bounds_module is not None and labels_module is not None and \
            bounds_module.__name__ == labels_module.__name__:
        S = bounds_module.Segmenter(audio_file, **config)
        est_idxs, est_labels = S.processFlat()
    # Different boundary and label algorithms
    else:
        # Identify segment boundaries
        if bounds_module is not None:
            S = bounds_module.Segmenter(audio_file, in_labels=[], **config)
            est_idxs, est_labels = S.processFlat()
        else:
            try:
                est_times, est_labels = io.read_references(
                    audio_file, annotator_id=annotator_id)
                est_idxs = io.align_times(est_times, frame_times[:-1])
                if est_idxs[0] != 0:
                    est_idxs = np.concatenate(([0], est_idxs))
                if est_idxs[-1] != features.shape[0] - 1:
                    est_idxs = np.concatenate((
                        est_idxs, [features.shape[0] - 1]))
            except:
                logging.warning("No references found for file: %s" %
                                audio_file)
                return [], []

        # Label segments
        if labels_module is not None:
            if len(est_idxs) == 2:
                est_labels = np.array([0])
            else:
                S = labels_module.Segmenter(audio_file,
                                            in_bound_idxs=est_idxs,
                                            **config)
                est_labels = S.processFlat()[1]

    # Make sure the first and last boundaries are included
    est_times, est_labels = utils.process_segmentation_level(
        est_idxs, est_labels, features.shape[0], frame_times,
        config["features"].dur)

    return est_times, est_labels


def run_algorithms(audio_file, boundaries_id, labels_id, config,
                   annotator_id=0):
    """Runs the algorithms with the specified identifiers on the audio_file.

    Parameters
    ----------
    audio_file: str
        Path to the audio file to segment.
    boundaries_id: str
        Identifier of the boundaries algorithm to use ("gt" for ground truth).
    labels_id: str
        Identifier of the labels algorithm to use (None for not labeling).
    config: dict
        Dictionary containing the custom parameters of the algorithms to use.
    annotator_id: int
        Annotator identificator in the ground truth.

    Returns
    -------
    est_times: np.array or list
        List of estimated times for the segment boundaries.
        If `list`, it will be a list of np.arrays, sorted by segmentation
        layer.
    est_labels: np.array or list
        List of all the labels associated segments.
        If `list`, it will be a list of np.arrays, sorted by segmentation
        layer.
    """
    # Check that there are enough audio frames
    if config["features"].features.shape[0] <= msaf.minimum__frames:
        logging.warning("Audio file too short, or too many few beats "
                        "estimated. Returning empty estimations.")
        return np.asarray([0, config["features"].dur]), \
            np.asarray([0], dtype=int)

    # Get the corresponding modules
    bounds_module = get_boundaries_module(boundaries_id)
    labels_module = get_labels_module(labels_id)

    # Get the correct frame times
    frame_times = config["features"].frame_times

    # Segment audio based on type of segmentation
    run_fun = run_hierarchical if config["hier"] else run_flat
{'fraction_non_alphanumeric': 0.07603049315253323, 'fraction_numerical': 0.0026985090737367606, 'mean_word_length': 2.2630420427030598, 'pattern_counts': {'":': 2, '<': 5, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 8, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3651428', 'n_tokens_mistral': 3976, 'n_tokens_neox': 3839, 'n_words': 1292}