test_util.py

''' Unit tests for utils
'''

import collections

import numpy as np
import nose.tools
import mir_eval
from mir_eval import util

A_TOL = 1e-12


def test_interpolate_intervals():
    """Check that an interval set is interpolated properly, with boundaries
    conditions and out-of-range values.
    """
    labels = list('abc')
    intervals = np.array([(n, n + 1.0) for n in range(len(labels))])
    time_points = [-1.0, 0.1, 0.9, 1.0, 2.3, 4.0]
    expected_ans = ['N', 'a', 'a', 'b', 'c', 'N']
    assert (util.interpolate_intervals(intervals, labels, time_points, 'N') ==
            expected_ans)


def test_interpolate_intervals_gap():
    """Check that an interval set is interpolated properly, with gaps."""
    labels = list('abc')
    intervals = np.array([[0.5, 1.0], [1.5, 2.0], [2.5, 3.0]])
    time_points = [0.0, 0.75, 1.25, 1.75, 2.25, 2.75, 3.5]
    expected_ans = ['N', 'a', 'N', 'b', 'N', 'c', 'N']
    assert (util.interpolate_intervals(intervals, labels, time_points, 'N') ==
            expected_ans)


@nose.tools.raises(ValueError)
def test_interpolate_intervals_badtime():
    """Check that interpolate_intervals throws an exception if
    input is unordered.
    """
    labels = list('abc')
    intervals = np.array([(n, n + 1.0) for n in range(len(labels))])
    time_points = [-1.0, 0.1, 0.9, 0.8, 2.3, 4.0]
    mir_eval.util.interpolate_intervals(intervals, labels, time_points)


def test_intervals_to_samples():
    """Check that an interval set is sampled properly, with boundaries
    conditions and out-of-range values.
    """
    labels = list('abc')
    intervals = np.array([(n, n + 1.0) for n in range(len(labels))])

    expected_times = [0.0, 0.5, 1.0, 1.5, 2.0, 2.5]
    expected_labels = ['a', 'a', 'b', 'b', 'c', 'c']
    result = util.intervals_to_samples(
        intervals, labels, offset=0, sample_size=0.5, fill_value='N')
    assert result[0] == expected_times
    assert result[1] == expected_labels

    expected_times = [0.25, 0.75, 1.25, 1.75, 2.25, 2.75]
    expected_labels = ['a', 'a', 'b', 'b', 'c', 'c']
    result = util.intervals_to_samples(
        intervals, labels, offset=0.25, sample_size=0.5, fill_value='N')
    assert result[0] == expected_times
    assert result[1] == expected_labels


def test_intersect_files():
    """Check that two non-identical yield correct results.
    """
    flist1 = ['/a/b/abc.lab', '/c/d/123.lab', '/e/f/xyz.lab']
    flist2 = ['/g/h/xyz.npy', '/i/j/123.txt', '/k/l/456.lab']
    sublist1, sublist2 = util.intersect_files(flist1, flist2)
    assert sublist1 == ['/e/f/xyz.lab', '/c/d/123.lab']
    assert sublist2 == ['/g/h/xyz.npy', '/i/j/123.txt']
    sublist1, sublist2 = util.intersect_files(flist1[:1], flist2[:1])
    assert sublist1 == []
    assert sublist2 == []


def test_merge_labeled_intervals():
    """Check that two labeled interval sequences merge correctly.
    """
    x_intvs = np.array([
        [0.0,    0.44],
        [0.44,  2.537],
        [2.537, 4.511],
        [4.511, 6.409]])
    x_labels = ['A', 'B', 'C', 'D']
    y_intvs = np.array([
        [0.0,   0.464],
        [0.464, 2.415],
        [2.415, 4.737],
        [4.737, 6.409]])
    y_labels = [0, 1, 2, 3]
    expected_intvs = [
        [0.0,    0.44],
        [0.44,  0.464],
        [0.464, 2.415],
        [2.415, 2.537],
        [2.537, 4.511],
        [4.511, 4.737],
        [4.737, 6.409]]
    expected_x_labels = ['A', 'B', 'B', 'B', 'C', 'D', 'D']
    expected_y_labels = [0,     0,   1,   2,   2,   2,   3]
    new_intvs, new_x_labels, new_y_labels = util.merge_labeled_intervals(
        x_intvs, x_labels, y_intvs, y_labels)

    assert new_x_labels == expected_x_labels
    assert new_y_labels == expected_y_labels
    assert new_intvs.tolist() == expected_intvs

    # Check that invalid inputs raise a ValueError
    y_intvs[-1, -1] = 10.0
    nose.tools.assert_raises(ValueError, util.merge_labeled_intervals, x_intvs,
                             x_labels, y_intvs, y_labels)


def test_boundaries_to_intervals():
    # Basic tests
    boundaries = np.arange(10)
    correct_intervals = np.array([np.arange(10 - 1), np.arange(1, 10)]).T
    intervals = mir_eval.util.boundaries_to_intervals(boundaries)
    assert np.all(intervals == correct_intervals)


def test_adjust_events():
    # Test appending at the end
    events = np.arange(1, 11)
    labels = [str(n) for n in range(10)]
    new_e, new_l = mir_eval.util.adjust_events(events, labels, 0.0, 11.)
    assert new_e[0] == 0.
    assert new_l[0] == '__T_MIN'
    assert new_e[-1] == 11.
    assert new_l[-1] == '__T_MAX'
    assert np.all(new_e[1:-1] == events)
    assert new_l[1:-1] == labels

    # Test trimming
    new_e, new_l = mir_eval.util.adjust_events(events, labels, 0.0, 9.)
    assert new_e[0] == 0.
    assert new_l[0] == '__T_MIN'
    assert new_e[-1] == 9.
    assert np.all(new_e[1:] == events[:-1])
    assert new_l[1:] == labels[:-1]


def test_bipartite_match():
    # This test constructs a graph as follows:
    #   v9 -- (u0)
    #   v8 -- (u0, u1)
    #   v7 -- (u0, u1, u2)
    #   ...
    #   v0 -- (u0, u1, ..., u9)
    #
    # This structure and ordering of this graph should force Hopcroft-Karp to
    # hit each algorithm/layering phase
    #
    G = collections.defaultdict(list)

    u_set = ['u{:d}'.format(_) for _ in range(10)]
    v_set = ['v{:d}'.format(_) for _ in range(len(u_set)+1)]
    for i, u in enumerate(u_set):
        for v in v_set[:-i-1]:
            G[v].append(u)

    matching = util._bipartite_match(G)

    # Make sure that each u vertex is matched
    nose.tools.eq_(len(matching), len(u_set))

    # Make sure that there are no duplicate keys
    lhs = set([k for k in matching])
    rhs = set([matching[k] for k in matching])

    nose.tools.eq_(len(matching), len(lhs))
    nose.tools.eq_(len(matching), len(rhs))

    # Finally, make sure that all detected edges are present in G
    for k in matching:
        v = matching[k]
        assert v in G[k] or k in G[v]


def test_outer_distance_mod_n():
    ref = [1., 2., 3.]
    est = [1.1, 6., 1.9, 5., 10.]
    expected = np.array([
        [0.1, 5., 0.9, 4., 3.],
        [0.9, 4., 0.1, 3., 4.],
        [1.9, 3., 1.1, 2., 5.]])
    actual = mir_eval.util._outer_distance_mod_n(ref, est)
    assert np.allclose(actual, expected)

    ref = [13., 14., 15.]
    est = [1.1, 6., 1.9, 5., 10.]
    expected = np.array([
        [0.1, 5., 0.9, 4., 3.],
        [0.9, 4., 0.1, 3., 4.],
        [1.9, 3., 1.1, 2., 5.]])
    actual = mir_eval.util._outer_distance_mod_n(ref, est)
    assert np.allclose(actual, expected)


def test_match_events():
    ref = [1., 2., 3.]
    est = [1.1, 6., 1.9, 5., 10.]
    expected = [(0, 0), (1, 2)]
    actual = mir_eval.util.match_events(ref, est, 0.5)
    assert actual == expected

    ref = [1., 2., 3., 11.9]
    est = [1.1, 6., 1.9, 5., 10., 0.]
    expected = [(0, 0), (1, 2), (3, 5)]
    actual = mir_eval.util.match_events(
        ref, est, 0.5, distance=mir_eval.util._outer_distance_mod_n)
    assert actual == expected


def test_fast_hit_windows():

    ref = [1., 2., 3.]
    est = [1.1, 6., 1.9, 5., 10.]

    ref_fast, est_fast = mir_eval.util._fast_hit_windows(ref, est, 0.5)
    ref_slow, est_slow = np.where(np.abs(np.subtract.outer(ref, est)) <= 0.5)

    assert np.all(ref_fast == ref_slow)
    assert np.all(est_fast == est_slow)


def test_validate_intervals():
    # Test for ValueError when interval shape is invalid
    nose.tools.assert_raises(
        ValueError, mir_eval.util.validate_intervals,
        np.array([[1.], [2.5], [5.]]))
    # Test for ValueError when times are negative
    nose.tools.assert_raises(
        ValueError, mir_eval.util.validate_intervals,
        np.array([[1., -2.], [2.5, 3.], [5., 6.]]))
    # Test for ValueError when duration is zero
    nose.tools.assert_raises(
        ValueError, mir_eval.util.validate_intervals,
        np.array([[1., 2.], [2.5, 2.5], [5., 6.]]))
    # Test for ValueError when duration is negative
    nose.tools.assert_raises(
        ValueError, mir_eval.util.validate_intervals,
        np.array([[1., 2.], [2.5, 1.5], [5., 6.]]))


def test_validate_events():
    # Test for ValueError when max_time is violated
    nose.tools.assert_raises(
        ValueError, mir_eval.util.validate_events, np.array([100., 100000.]))
    # Test for ValueError when events aren't 1-d arrays
    nose.tools.assert_raises(
        ValueError, mir_eval.util.validate_events,
        np.array([[1., 2.], [3., 4.]]))
    # Test for ValueError when event times are not increasing
    nose.tools.assert_raises(
        ValueError, mir_eval.util.validate_events,
        np.array([1., 2., 5., 3.]))


def test_validate_frequencies():
    # Test for ValueError when max_freq is violated
    nose.tools.assert_raises(
        ValueError, mir_eval.util.validate_frequencies,
        np.array([100., 100000.]), 5000., 20.)
    # Test for ValueError when min_freq is violated
    nose.tools.assert_raises(
        ValueError, mir_eval.util.validate_frequencies,
        np.array([2., 200.]), 5000., 20.)
    # Test for ValueError when events aren't 1-d arrays
    nose.tools.assert_raises(
        ValueError, mir_eval.util.validate_frequencies,
        np.array([[100., 200.], [300., 400.]]), 5000., 20.)
    # Test for ValueError when allow_negatives is false and negative values
    # are passed
    nose.tools.assert_raises(
        ValueError, mir_eval.util.validate_frequencies,
        np.array([[-100., 200.], [300., 400.]]), 5000., 20.,
        allow_negatives=False)
    # Test for ValueError when max_freq is violated and allow_negatives=True
    nose.tools.assert_raises(
        ValueError, mir_eval.util.validate_frequencies,
        np.array([100., -100000.]), 5000., 20., allow_negatives=True)
    # Test for ValueError when min_freq is violated and allow_negatives=True
    nose.tools.assert_raises(
        ValueError, mir_eval.util.validate_frequencies,
        np.array([-2., 200.]), 5000., 20., allow_negatives=True)


def test_has_kwargs():

    def __test(target, f):
        assert target == mir_eval.util.has_kwargs(f)

    def f1(_):
        return None

    def f2(_=5):
        return None

    def f3(*_):
        return None

    def f4(_, **kw):
        return None

    def f5(_=5, **kw):
        return None

    yield __test, False, f1
    yield __test, False, f2
    yield __test, False, f3
    yield __test, True, f4
    yield __test, True, f5


def test_sort_labeled_intervals():

    def __test_labeled(x, labels, x_true, lab_true):
        xs, ls = mir_eval.util.sort_labeled_intervals(x, labels)

        assert np.allclose(xs, x_true)
        nose.tools.eq_(ls, lab_true)

    def __test(x, x_true):
        xs = mir_eval.util.sort_labeled_intervals(x)
        assert np.allclose(xs, x_true)

    x1 = np.asarray([[10, 20], [0, 10]])
    x1_true = np.asarray([[0, 10], [10, 20]])
    labels = ['a', 'b']
    labels_true = ['b', 'a']

    yield __test_labeled, x1, labels, x1_true, labels_true
    yield __test, x1, x1_true
    yield __test_labeled, x1_true, labels_true, x1_true, labels_true
    yield __test, x1_true, x1_true


def test_estimate_hop_length():
    times1 = np.array([0.00, 0.01, 0.02, 0.03, 0.04])
    times2 = np.concatenate((times1, [0.10, 0.11, 0.12, 0.13, 0.14]))
    times3 = np.array([0.00, 0.01, 0.03, 0.06, 0.10])

    expected_hop = 0.01

    actual_hop1 = mir_eval.util.estimate_hop_length(times1)
    actual_hop2 = mir_eval.util.estimate_hop_length(times2)

    # NumPy diff() does not always return exact values
    assert abs(actual_hop1 - expected_hop) < A_TOL
    assert abs(actual_hop2 - expected_hop) < A_TOL

    nose.tools.assert_raises(ValueError, util.estimate_hop_length, times3)


def __times_equal(times_a, times_b):
    if len(times_a) != len(times_b):
        return False
    else:
        equal = True
        for time_a, time_b in zip(times_a, times_b):
            equal = equal and abs(time_a - time_b) < A_TOL
        return equal


def __frequencies_equal(freqs_a, freqs_b):
    if len(freqs_a) != len(freqs_b):
        return False
    else:
        equal = True
        for freq_a, freq_b in zip(freqs_a, freqs_b):
            if freq_a.size != freq_b.size:
                return False
            equal = equal and np.allclose(freq_a, freq_b, atol=A_TOL)
        return equal


def test_time_series_to_uniform():
    times1 = np.array([0.00, 0.01, 0.02, 0.03])
    times2 = np.array([0.00, 0.01, 0.03, 0.04])
    times3 = times2

    freqs = [np.array([100.]),
             np.array([100.]),
             np.array([200.]),
             np.array([200.])]

    hop_size = 0.01

    expected_times1 = times1
    expected_freqs1 = freqs

    expected_times2 = np.array([0.00, 0.01, 0.02, 0.03, 0.04])
    expected_freqs2 = [np.array([100.]),
                       np.array([100.]),
                       np.array([]),
                       np.array([200.]),
                       np.array([200.])]

    expected_times3 = hop_size * np.arange(20)
    expected_freqs3 = [np.array([100.]),
                       np.array([100.]),
                       np.array([]),
                       np.array([200.]),
                       np.array([200.])] + \
                      [np.array([])] * 15

    expected_times4 = np.array([0.00, 0.01, 0.02])
    expected_freqs4 = [np.array([])] * 3

    actual_times1, actual_values1 = mir_eval.util.time_series_to_uniform(times1, freqs, hop_size, None)
    actual_times2, actual_values2 = mir_eval.util.time_series_to_uniform(times2, freqs, hop_size, None)
    actual_times3, actual_values3 = mir_eval.util.time_series_to_uniform(times3, freqs, hop_size, 0.195)

    actual_times4, actual_values4 = mir_eval.util.time_series_to_uniform(np.array([]), [], hop_size, times1[-1])

    assert __times_equal(actual_times1, expected_times1)
    assert __frequencies_equal(actual_values1, expected_freqs1)

    assert __times_equal(actual_times2, expected_times2)
    assert __frequencies_equal(actual_values2, expected_freqs2)

    assert __times_equal(actual_times3, expected_times3)
    assert __frequencies_equal(actual_values3, expected_freqs3)

    assert __times_equal(actual_times4, expected_times4)
    assert __frequencies_equal(actual_values4, expected_freqs4)