Merge 4c197ed into f2ddb6e

librosa/beat.py

@@ -7,6 +7,13 @@
    :toctree: generated/
 
    beat_track
+   tempo
+
+Deprecated
+----------
+.. autosummary::
+   :toctree: generated/
+
    estimate_tempo
 """
 
@@ -17,9 +24,10 @@
 from . import core
 from . import onset
 from . import util
+from .feature import tempogram
 from .util.exceptions import ParameterError
 
-__all__ = ['beat_track', 'estimate_tempo']
+__all__ = ['beat_track', 'tempo', 'estimate_tempo']
 
 
 def beat_track(y=None, sr=22050, onset_envelope=None, hop_length=512,
@@ -174,10 +182,10 @@ def beat_track(y=None, sr=22050, onset_envelope=None, hop_length=512,
 
     # Estimate BPM if one was not provided
     if bpm is None:
-        bpm = estimate_tempo(onset_envelope,
-                             sr=sr,
-                             hop_length=hop_length,
-                             start_bpm=start_bpm)
+        bpm = tempo(onset_envelope=onset_envelope,
+                    sr=sr,
+                    hop_length=hop_length,
+                    start_bpm=start_bpm)[0]
 
     # Then, run the tracker
     beats = __beat_tracker(onset_envelope,
@@ -198,11 +206,15 @@ def beat_track(y=None, sr=22050, onset_envelope=None, hop_length=512,
     return (bpm, beats)
 
 
+@util.decorators.deprecated('0.5.0', '0.6')
 @cache(level=30)
 def estimate_tempo(onset_envelope, sr=22050, hop_length=512, start_bpm=120,
                    std_bpm=1.0, ac_size=4.0, duration=90.0, offset=0.0):
     """Estimate the tempo (beats per minute) from an onset envelope
 
+    .. warning:: Deprected in librosa 0.5
+                 Functionality is superseded by
+                 `librosa.beat.tempo`.
 
     Parameters
     ----------
@@ -316,6 +328,137 @@ def estimate_tempo(onset_envelope, sr=22050, hop_length=512, start_bpm=120,
     return start_bpm
 
 
+@cache(level=30)
+def tempo(y=None, sr=22050, onset_envelope=None, hop_length=512, start_bpm=120,
+          std_bpm=1.0, ac_size=4.0, max_tempo=320.0, aggregate=np.mean):
+    """Estimate the tempo (beats per minute)
+
+    Parameters
+    ----------
+    y : np.ndarray [shape=(n,)] or None
+        audio time series
+
+    sr : number > 0 [scalar]
+        sampling rate of the time series
+
+    onset_envelope    : np.ndarray [shape=(n,)]
+        pre-computed onset strength envelope
+
+    hop_length : int > 0 [scalar]
+        hop length of the time series
+
+    start_bpm : float [scalar]
+        initial guess of the BPM
+
+    std_bpm : float > 0 [scalar]
+        standard deviation of tempo distribution
+
+    ac_size : float > 0 [scalar]
+        length (in seconds) of the auto-correlation window
+
+    max_tempo : float > 0 [scalar, optional]
+        If provided, only estimate tempo below this threshold
+
+    aggregate : callable [optional]
+        Aggregation function for estimating global tempo.
+        If `None`, then tempo is estimated independently for each frame.
+
+    Returns
+    -------
+    tempo : np.ndarray [scalar]
+        estimated tempo (beats per minute)
+
+    See Also
+    --------
+    librosa.onset.onset_strength
+    librosa.feature.tempogram
+
+    Notes
+    -----
+    This function caches at level 30.
+
+    Examples
+    --------
+    >>> y, sr = librosa.load(librosa.util.example_audio_file())
+    >>> onset_env = librosa.onset.onset_strength(y, sr=sr)
+    >>> tempo = librosa.beat.tempo(onset_envelope=onset_env, sr=sr)
+    >>> tempo
+    array([103.359375])
+
+    Plot the estimated tempo against the onset autocorrelation
+
+    >>> import matplotlib.pyplot as plt
+    >>> # Compute 2-second windowed autocorrelation
+    >>> hop_length = 512
+    >>> ac = librosa.autocorrelate(onset_env, 2 * sr // hop_length)
+    >>> freqs = librosa.tempo_frequencies(len(ac), sr=sr,
+    ...                                   hop_length=hop_length)
+    >>> # Plot on a BPM axis.  We skip the first (0-lag) bin.
+    >>> plt.figure(figsize=(8,4))
+    >>> plt.semilogx(freqs[1:], librosa.util.normalize(ac)[1:],
+    ...              label='Onset autocorrelation', basex=2)
+    >>> plt.axvline(tempo, 0, 1, color='r', alpha=0.75, linestyle='--',
+    ...            label='Tempo: {:.2f} BPM'.format(tempo))
+    >>> plt.xlabel('Tempo (BPM)')
+    >>> plt.grid()
+    >>> plt.legend(frameon=True)
+    >>> plt.axis('tight')
+    """
+
+    if start_bpm <= 0:
+        raise ParameterError('start_bpm must be strictly positive')
+
+    win_length = np.asscalar(core.time_to_frames(ac_size, sr=sr,
+                                                 hop_length=hop_length))
+
+    tg = tempogram(y=y, sr=sr,
+                   onset_envelope=onset_envelope,
+                   hop_length=hop_length,
+                   win_length=win_length)
+
+    # Eventually, we want this to work for time-varying tempo
+    if aggregate is not None:
+        tg = aggregate(tg, axis=1, keepdims=True)
+
+    # Get the BPM values for each bin, skipping the 0-lag bin
+    bpms = core.tempo_frequencies(tg.shape[0], hop_length=hop_length, sr=sr)
+
+    # Weight the autocorrelation by a log-normal distribution
+    prior = np.exp(-0.5 * ((np.log2(bpms) - np.log2(start_bpm)) / std_bpm)**2)
+
+    # Kill everything above the max tempo
+    if max_tempo is not None:
+        max_idx = np.argmax(bpms < max_tempo)
+        prior[:max_idx] = 0
+
+    tg *= prior[:, np.newaxis]
+
+    # Really, instead of multiplying by the prior, we should set up a
+    # probabilistic model for tempo and add log-probabilities.
+    # This would give us a chance to recover from null signals and
+    # rely on the prior.
+    # it would also make time aggregation much more natural
+
+    # Get the local maximum of weighted correlation
+    x_peaks = util.localmax(tg, axis=0)
+
+    # For each peak, set its harmonics to true
+    peak_idy, peak_idx = np.nonzero(x_peaks)
+
+    for h in [1./3, 1./2, 2./3, 3./2, 2, 3]:
+        rows = (peak_idy * h).astype(int)
+        # Only take rows that stay within sample
+        v = (0 < rows) & (rows < x_peaks.shape[0])
+        x_peaks[rows[v], peak_idx[v]] = True
+
+    best_period = np.argmax(tg * x_peaks, axis=0)
+
+    tempi = bpms[best_period]
+    # Wherever the best tempo is index 0, return start_bpm
+    tempi[best_period == 0] = start_bpm
+    return tempi
+
+
 def __beat_tracker(onset_envelope, bpm, fft_res, tightness, trim):
     """Internal function that tracks beats in an onset strength envelope.
 

tests/test_beat.py

@@ -45,7 +45,7 @@ def __test(infile):
         yield (__test, infile)
 
 
-def test_tempo():
+def test_estimate_tempo():
 
     def __test(infile):
         DATA = load(infile)
@@ -63,6 +63,34 @@ def __test(infile):
         yield (__test, infile)
 
 
+def test_tempo():
+
+    def __test(tempo, sr, hop_length, ac_size, aggregate, y):
+
+        tempo_est = librosa.beat.tempo(y=y, sr=sr, hop_length=hop_length,
+                                       ac_size=ac_size,
+                                       aggregate=aggregate)
+
+        # Being within 10% for the stable frames is close enough
+        if aggregate is None:
+            win_size = int(ac_size * sr // hop_length)
+            assert np.all(np.abs(tempo_est[win_size:-win_size] - tempo) <= 0.10 * tempo), (tempo,
+                    tempo_est[win_size:-win_size])
+        else:
+            assert np.abs(tempo_est - tempo) <= 0.10 * tempo, (tempo, tempo_est)
+
+    for sr in [22050, 44100]:
+        for tempo in [40, 60, 80, 110, 150, 160]:
+            # Make a pulse train at the target tempo
+            y = np.zeros(20 * sr)
+            delay = np.asscalar(librosa.time_to_samples(60./tempo, sr=sr))
+            y[::delay] = 1
+            for hop_length in [512, 1024]:
+                for ac_size in [4, 8]:
+                    for aggregate in [None, np.mean]:
+                        yield __test, tempo, sr, hop_length, ac_size, aggregate, y
+
+
 @raises(librosa.ParameterError)
 def test_beat_no_input():
 
@@ -85,7 +113,7 @@ def test_beat_no_onsets():
     eq_(len(beats), 0)
 
 
-def test_tempo_no_onsets():
+def test_estimate_tempo_no_onsets():
 
     sr = 22050
     hop_length = 512
@@ -102,6 +130,25 @@ def __test(start_bpm):
         yield __test, start_bpm
 
 
+def test_tempo_no_onsets():
+
+    sr = 22050
+    hop_length = 512
+    duration = 30
+    onsets = np.zeros(duration * sr // hop_length)
+
+    def __test(start_bpm, aggregate):
+        tempo = librosa.beat.tempo(onset_envelope=onsets, sr=sr,
+                                   hop_length=hop_length,
+                                   start_bpm=start_bpm,
+                                   aggregate=aggregate)
+        assert np.allclose(tempo, start_bpm)
+
+    for start_bpm in [40, 60, 120, 240]:
+        for aggregate in [None, np.mean]:
+            yield __test, start_bpm, aggregate
+
+
 def test_beat():
 
     y, sr = librosa.load(__EXAMPLE_FILE)