def test_chirp():

    def __test(fmin, fmax, sr, length, duration, linear, phi):

        y = librosa.chirp(fmin=fmin,
                          fmax=fmax,
                          sr=sr,
                          length=length,
                          duration=duration,
                          linear=linear,
                          phi=phi)

        if length is not None:
            assert len(y) == length
        else:
            assert len(y) == np.ceil(duration * sr)

    # Bad cases
    yield raises(librosa.ParameterError)(__test), None, None, 22050, 22050, 1, False, None
    yield raises(librosa.ParameterError)(__test), 440, None, 22050, 22050, 1, False, None
    yield raises(librosa.ParameterError)(__test), None, 880, 22050, 22050, 1, False, None
    yield raises(librosa.ParameterError)(__test), 440, 880, 22050, None, None, False, None

    for sr in [11025, 22050]:
        for length in [None, 11025]:
            for duration in [None, 0.5]:
                for phi in [None, np.pi / 2]:
                    if length is not None or duration is not None:
                        yield __test, 440, 880, sr, length, duration, False, phi
                        yield __test, 880, 440, sr, length, duration, True, phi

 def test_init_wo_num(self):
     """
     SimObject.__init__ should raise ValueError if num_* are not specified.
     """
     raises(ValueError)(self.check_init_wo_num)()
     self.check_init_wo_num(num_i=0)
     self.check_init_wo_num(num_i=1)

def test_load_fail():
    # 1. test bad file path
    # 2. test non-json file
    # 3. test bad extensions
    # 4. test bad codecs

    def __test(filename, fmt):
        jams.load(filename, fmt=fmt)

    # Make a non-existent file
    tdir = tempfile.mkdtemp()
    yield raises(IOError)(__test), os.path.join(tdir, 'nonexistent.jams'), 'jams'
    os.rmdir(tdir)

    # Make a non-json file
    tdir = tempfile.mkdtemp()
    badfile = os.path.join(tdir, 'nonexistent.jams')
    with open(badfile, mode='w') as fp:
        fp.write('some garbage')
    yield raises(ValueError)(__test), os.path.join(tdir, 'nonexistent.jams'), 'jams'
    os.unlink(badfile)
    os.rmdir(tdir)

    tdir = tempfile.mkdtemp()
    for ext in ['txt', '']:
        badfile = os.path.join(tdir, 'nonexistent')
        yield raises(jams.ParameterError)(__test), '{:s}.{:s}'.format(badfile, ext), 'auto'
        yield raises(jams.ParameterError)(__test), '{:s}.{:s}'.format(badfile, ext), ext
        yield raises(jams.ParameterError)(__test), '{:s}.jams'.format(badfile), ext
    os.rmdir(tdir)

def test_ns_tag_msd_tagtraum_cd1():

    def __test(tag, confidence=None):
        ann = Annotation(namespace='tag_msd_tagtraum_cd1')

        ann.append(time=0, duration=1, value=tag, confidence=confidence)

        ann.validate()

    for tag in ['reggae',
                'pop/rock',
                'rnb',
                'jazz',
                'vocal',
                'new age',
                'latin',
                'rap',
                'country',
                'international',
                'blues',
                'electronic',
                'folk']:

        yield __test, tag
        yield __test, six.u(tag)
        yield raises(SchemaError)(__test), tag.upper()


    for tag in [23, None]:
        yield raises(SchemaError)(__test), tag

    yield raises(SchemaError)(__test), 'folk', 1.2
    yield raises(SchemaError)(__test), 'folk', -0.1
    yield __test, 'folk', 1.0
    yield __test, 'folk', 0.0

def test_load_value_dict():
    def new_network():
        return tn.SequentialNode(
            "seq",
            [tn.InputNode("i", shape=(10, 100)),
             tn.LinearMappingNode(
                 "lm",
                 output_dim=15,
                 inits=[treeano.inits.NormalWeightInit()])]
        ).network()

    n1 = new_network()
    n2 = new_network()

    fn1 = n1.function(["i"], ["lm"])
    fn2 = n2.function(["i"], ["lm"])

    x = np.random.randn(10, 100).astype(fX)

    def test():
        np.testing.assert_equal(fn1(x), fn2(x))

    # should fail
    nt.raises(AssertionError)(test)()
    # change weights
    canopy.network_utils.load_value_dict(
        n1, canopy.network_utils.to_value_dict(n2))
    # should not fail
    test()

def test_tmeasure_fail_span():

    # Does not start at 0
    ref = [[[1, 10]],
           [[1, 5],
            [5, 10]]]

    ref = [np.asarray(_) for _ in ref]

    yield raises(ValueError)(mir_eval.hierarchy.tmeasure), ref, ref

    # Does not end at the right time
    ref = [[[0, 5]],
           [[0, 5],
            [5, 6]]]
    ref = [np.asarray(_) for _ in ref]

    yield raises(ValueError)(mir_eval.hierarchy.tmeasure), ref, ref

    # Two annotaions of different shape
    ref = [[[0, 10]],
           [[0, 5],
            [5, 10]]]
    ref = [np.asarray(_) for _ in ref]

    est = [[[0, 15]],
           [[0, 5],
            [5, 15]]]
    est = [np.asarray(_) for _ in est]

    yield raises(ValueError)(mir_eval.hierarchy.tmeasure), ref, est

def test_match_events_onesided():

    events_from = np.asarray([5, 15, 25])
    events_to = np.asarray([0, 10, 20, 30])

    def __test(left, right, target):
        match = librosa.util.match_events(events_from, events_to,
                                          left=left, right=right)

        assert np.allclose(target, events_to[match])

    yield __test, False, True, [10, 20, 30]
    yield __test, True, False, [0, 10, 20]

    # Make a right-sided fail
    events_from[0] = 40
    yield raises(librosa.ParameterError)(__test), False, True, [10, 20, 30]

    # Make a left-sided fail
    events_from[0] = -1
    yield raises(librosa.ParameterError)(__test), True, False, [10, 20, 30]

    # Make a two-sided fail
    events_from[0] = -1
    yield raises(librosa.ParameterError)(__test), False, False, [10, 20, 30]

    # Make a two-sided success
    events_to[:-1] = events_from
    yield __test, False, False, events_from

def test_gcep_invalid_args():
    x = windowed_dummy_data(1024)

    def __test_gamma(gamma):
        pysptk.gcep(x, gamma=gamma)

    yield raises(ValueError)(__test_gamma), 0.1
    yield raises(ValueError)(__test_gamma), -2.1

    def __test_itype(itype=0):
        pysptk.gcep(x, itype=itype)

    yield raises(ValueError)(__test_itype), -1
    yield raises(ValueError)(__test_itype), 5

    def __test_eps(etype=0, eps=0.0):
        pysptk.gcep(x, etype=etype, eps=eps)

    yield raises(ValueError)(__test_eps), 0, -1.0
    yield raises(ValueError)(__test_eps), -1
    yield raises(ValueError)(__test_eps), -3
    yield raises(ValueError)(__test_eps), 1, -1.0
    yield raises(ValueError)(__test_eps), 2, -1.0

    def __test_min_det(min_det):
        pysptk.gcep(x, min_det=min_det)

    yield raises(ValueError)(__test_min_det), -1.0

def test_trans_cycle():
    def __trans(n, p):
        A = librosa.sequence.transition_cycle(n, p)

        # Right shape
        assert A.shape == (n, n)
        # diag is correct
        assert np.allclose(np.diag(A), p)

        for i in range(n):
            assert A[i, np.mod(i + 1, n)] == 1 - A[i, i]

        # we have well-formed distributions
        assert np.all(A >= 0)
        assert np.allclose(A.sum(axis=1), 1)

    # Test with constant self-loops
    for n in range(2, 4):
        yield __trans, n, 0.5

    # Test with variable self-loops
    yield __trans, 3, [0.8, 0.7, 0.5]

    # Failure if we don't have enough states
    yield raises(librosa.ParameterError)(__trans), 1, 0.5

    # Failure if n_states is wrong
    yield raises(librosa.ParameterError)(__trans), None, 0.5

    # Failure if p is not a probability
    yield raises(librosa.ParameterError)(__trans), 3, 1.5
    yield raises(librosa.ParameterError)(__trans), 3, -0.25

    # Failure if there's a shape mismatch
    yield raises(librosa.ParameterError)(__trans), 3, [0.5, 0.2]

    def test_raises(self):
        from nose.case import FunctionTestCase

        def raise_typeerror():
            raise TypeError("foo")

        def noraise():
            pass

        raise_good = raises(TypeError)(raise_typeerror)
        raise_other = raises(ValueError)(raise_typeerror)
        no_raise = raises(TypeError)(noraise)

        tc = FunctionTestCase(raise_good)
        self.assertEqual(str(tc), "%s.%s" % (__name__, 'raise_typeerror'))

        raise_good()
        try:
            raise_other()
        except TypeError as e:
            pass
        else:
            self.fail("raises did pass through unwanted exception")

        try:
            no_raise()
        except AssertionError as e:
            pass
        else:
            self.fail("raises did not raise assertion error on no exception")

def check_arrayaccess(clibname, list_num, list_cdt, cdt, dim,
                      _calloc_=None, carrtype=None):
    """Check C side array access"""
    if cdt in ['char', 'short', 'ushort', 'int', 'uint', 'long', 'ulong',
               'longlong', 'ulonglong', 'bool', 'size_t']:
        ass_eq = assert_equal
    elif cdt in ['float', 'double', 'longdouble']:
        ass_eq = assert_almost_equal

    ArrayAccess = gene_class_ArrayAccess(
        clibname, len(list_num), list_cdt, carrtype)
    num_dict = dict(zip(ArrayAccess.num_names, list_num))  # {num_i: 6, ...}
    if _calloc_ is not None:
        num_dict.update(_calloc_=_calloc_)
    aa = ArrayAccess(**num_dict)
    aa.fill()
    # arr_via_ret should return same array (garr)
    garr = aa.arr_via_ret(cdt, dim)
    arr = aa.arr(cdt, dim)
    ass_eq(garr, arr)
    # insert completely different value to 'arr'
    if cdt == 'char':
        arr.flat = alpharange(100, numpy.prod(arr.shape) + 100)
    elif cdt == 'bool':
        arr[:] = -arr
    else:
        arr += 100
    raises(AssertionError)(assert_equal)(garr, arr)
    # get array (garr2) via arr_via_ret again
    garr2 = aa.arr_via_ret(cdt, dim)
    assert_equal(garr2, arr)

def test_files():

    # Expected output
    output = [
        os.path.join(os.path.abspath(os.path.curdir), "data", s)
        for s in ["test1_22050.wav", "test1_44100.wav", "test2_8000.wav"]
    ]

    def __test(searchdir, ext, recurse, case_sensitive, limit, offset):
        files = librosa.util.find_files(
            searchdir, ext=ext, recurse=recurse, case_sensitive=case_sensitive, limit=limit, offset=offset
        )

        s1 = slice(offset, None)
        s2 = slice(limit)

        assert set(files) == set(output[s1][s2])

    for searchdir in [os.path.curdir, os.path.join(os.path.curdir, "data")]:
        for ext in [None, "wav", "WAV", ["wav"], ["WAV"]]:
            for recurse in [False, True]:
                for case_sensitive in [False, True]:
                    for limit in [None, 1, 2]:
                        for offset in [0, 1, -1]:
                            tf = __test

                            if searchdir == os.path.curdir and not recurse:
                                tf = raises(AssertionError)(__test)

                            if ext is not None and case_sensitive and (ext == "WAV" or set(ext) == set(["WAV"])):

                                tf = raises(AssertionError)(__test)

                            yield (tf, searchdir, ext, recurse, case_sensitive, limit, offset)

def test_melody_invalid():

    f1 = np.linspace(110.0, 440.0, 10)
    v1 = np.sign(np.random.randn(len(f1)))
    v2 = np.sign(np.random.randn(len(f1)))

    ref_ann = create_annotation(values=f1 * v1,
                                confidence=1.0,
                                duration=0.01,
                                namespace='pitch_hz')

    est_ann = create_annotation(values=f1 * v2,
                                confidence=1.0,
                                duration=0.01,
                                namespace='pitch_midi')


    yield raises(jams.NamespaceError)(jams.eval.melody), ref_ann, est_ann
    yield raises(jams.NamespaceError)(jams.eval.melody), est_ann, ref_ann

    est_ann = create_annotation(values=['a', 'b', 'c'],
                                confidence=1.0,
                                duration=0.01,
                                namespace='pitch_hz')

    yield raises(jams.SchemaError)(jams.eval.melody), ref_ann, est_ann
    yield raises(jams.SchemaError)(jams.eval.melody), est_ann, ref_ann

def test_delta():
    # Note: this test currently only checks first-order differences

    def __test(width, order, axis, x):
        delta   = librosa.feature.delta(x,
                                        width=width,
                                        order=order,
                                        axis=axis)

        # Check that trimming matches the expected shape
        eq_(x.shape, delta.shape)

        # Once we're sufficiently far into the signal (ie beyond half_len)
        # (x + delta)[t] should approximate x[t+1] if x is actually linear
        slice_orig = [slice(None)] * x.ndim
        slice_out = [slice(None)] * delta.ndim
        slice_orig[axis] = slice(width//2 + 1, -width//2 + 1)
        slice_out[axis] = slice(width//2, -width//2)
        assert np.allclose((x + delta)[slice_out], x[slice_orig])

    x = np.vstack([np.arange(100.0)] * 3)

    for width in range(-1, 8):
        for slope in np.linspace(-2, 2, num=6):
            for bias in [-10, 0, 10]:
                for order in [0, 1]:
                    for axis in range(x.ndim):
                        tf = __test
                        if width < 3 or np.mod(width, 2) != 1 or width > x.shape[axis]:
                            tf = raises(librosa.ParameterError)(__test)
                        if order != 1:
                            tf = raises(librosa.ParameterError)(__test)
                        yield tf, width, order, axis, slope * x + bias

def test_stack_memory():

    def __test(data, n_steps, delay):
        data_stack = librosa.feature.stack_memory(data,
                                                  n_steps=n_steps,
                                                  delay=delay)

        # If we're one-dimensional, reshape for testing
        if data.ndim == 1:
            data = data.reshape((1, -1))

        d, t = data.shape

        eq_(data_stack.shape[0], n_steps * d)
        eq_(data_stack.shape[1], t)

        for i in range(d):
            for step in range(1, n_steps):
                assert np.allclose(data[i, :- step * delay],
                                   data_stack[step * d + i, step * delay:])

    srand()

    for ndim in [1, 2]:
        data = np.random.randn(* ([5] * ndim))

        for n_steps in [-1, 0, 1, 2, 3, 4]:
            for delay in [-1, 0, 1, 2, 4]:
                tf = __test
                if n_steps < 1:
                    tf = raises(librosa.ParameterError)(__test)
                if delay < 1:
                    tf = raises(librosa.ParameterError)(__test)
                yield tf, data, n_steps, delay

def check_query(backend_factory, backend_kwargs={}):

    message = pickle.dumps('some data')
    backend_kwargs.setdefault('type', 380)

    with nested(create_test_client(), closing(backend_factory(
        **backend_kwargs))) as (client, backend):

        with timedcontext(4):
            client.connect(server.server_address, sync=True)

            raises(OperationFailed)(lambda: client.query(fields={'to':
                backend.instance_id, 'workflow': 'some workflow'},
                message=message, type=1136, timeout=1))()

            backend.connect(server.server_address, sync=True)

            th = TestThread(target=backend.handle_one)
            th.setDaemon(True)
            th.start()

            response = client.query(fields={'to': backend.instance_id,
                'workflow': 'some workflow'}, message=message, type=1136,
                timeout=1)

            eq_(response.message, message)
            eq_(response.from_, backend.instance_id)
            eq_(response.to, client.instance_id)
            eq_(response.workflow, 'some workflow')

            th.join()

def test_network_nanguard():
    class CustomNode(treeano.NodeImpl):
        input_keys = ()

        def compute_output(self, network):
            network.create_vw(
                "default",
                is_shared=True,
                shape=(),
                inits=[]
            )

    network = CustomNode("c").network()
    # build eagerly to share weights
    network.build()

    fn = canopy.handlers.handled_fn(
        network,
        [canopy.handlers.network_nanguard()],
        {},
        {})

    vw = network["c"].get_vw("default")
    for x in [3, 4, 1e9, 9e9, -9e9, 0]:
        vw.variable.set_value(treeano.utils.as_fX(x))
        fn({})

    for x in [np.inf, -np.inf, np.nan, 2e10]:
        vw.variable.set_value(treeano.utils.as_fX(x))
        nt.raises(Exception)(lambda x: fn(x))({})

def test_load_value_dict_not_strict_keys():
    n1 = tn.SequentialNode(
        "seq",
        [tn.InputNode("i", shape=(10, 100)),
         tn.LinearMappingNode(
             "lm",
             output_dim=15,
             inits=[treeano.inits.NormalWeightInit()])]
    ).network()
    n2 = tn.InputNode("i", shape=()).network()

    def test1(strict_keys):
        canopy.network_utils.load_value_dict(
            n1,
            canopy.network_utils.to_value_dict(n2),
            strict_keys=strict_keys)

    def test2(strict_keys):
        canopy.network_utils.load_value_dict(
            n2,
            canopy.network_utils.to_value_dict(n1),
            strict_keys=strict_keys)

    nt.raises(AssertionError)(test1)(strict_keys=True)
    nt.raises(AssertionError)(test2)(strict_keys=True)
    test1(strict_keys=False)
    test2(strict_keys=False)

def make_dummy_ungridded_data_single_point(lat=0.0, lon=0.0, value=1.0, time=None, altitude=None, pressure=None,
                                           mask=None):
    from cis.data_io.Coord import CoordList, Coord
    from cis.data_io.ungridded_data import UngriddedData, Metadata
    import datetime
    import numpy

    x = Coord(numpy.array(lat), Metadata('latitude'), 'x')
    y = Coord(numpy.array(lon), Metadata('longitude'), 'y')

    if (time is not None) + (altitude is not None) + (pressure is not None) > 1:
        raises(NotImplementedError)
    elif time is None and altitude is None and pressure is None:
        coords = CoordList([x, y])
    elif altitude is not None:
        z = Coord(numpy.array(altitude), Metadata('altitude'), 'z')
        coords = CoordList([x, y, z])
    elif time is not None:
        t = Coord(numpy.array(time), Metadata('time'), 't')
        coords = CoordList([x, y, t])
    elif pressure is not None:
        p = Coord(numpy.array(pressure), Metadata('air_pressure'), 'p')
        coords = CoordList([x, y, p])

    data = numpy.array(value)
    if mask:
        data = ma.masked_array(data, mask=mask)
    return UngriddedData(data, Metadata(name='Rain', standard_name='rainfall_rate', long_name="Total Rainfall",
                                        units="kg m-2 s-1", missing_value=-999), coords)

def test_deserialize():
    for case in encoded_messages:
        msg = parse_message(MultiplexerMessage, case['encoded'])
        assert dict((field.name, value) for field, value in msg.ListFields()) \
                == case['pythonized'] == dict_message(msg, all_fields=False)

    raises(DecodeError)(lambda: parse_message(VariousFields, ''))()