python/demos/demo_timestretch.py

   1 #! /usr/bin/env python
   2
   3 # Implementation of the timescale algorithm according to Dan Ellis, *A Phase
   4 # Vocoder in Matlab*.  http://www.ee.columbia.edu/~dpwe/resources/matlab/pvoc/
   5
   6 # This file follows the original implementation, with analysis in a first pass,
   7 # and synthesis in a second pass.
   8
   9 import sys
  10 from aubio import source, sink, pvoc, mfcc, cvec
  11 from aubio import unwrap2pi, float_type
  12 import numpy as np
  13
  14 win_s = 1024
  15 hop_s = win_s / 8 # 87.5 % overlap
  16
  17 warmup = win_s // hop_s - 1
  18
  19 if len(sys.argv) < 3:
  20     print("Usage: {:s} <source_filename> <output_filename> <rate> [samplerate]".format(sys.argv[0]))
  21     print("""Examples:
  22     # twice faster
  23     {0} track_01.mp3 track_01_faster.wav 2.0
  24     # twice slower
  25     {0} track_02.flac track_02_slower.wav 0.5
  26     # one and a half time faster, resampling first the input to 22050
  27     {0} track_02.flac track_02_slower.wav 1.5 22050""".format(sys.argv[0]))
  28     sys.exit(1)
  29
  30 source_filename = sys.argv[1]
  31 output_filename = sys.argv[2]
  32 rate = float(sys.argv[3])
  33
  34 samplerate = 0 if len(sys.argv) < 5 else int(sys.argv[4])
  35 source_in = source(source_filename, samplerate, hop_s)
  36 samplerate = source_in.samplerate
  37 p = pvoc(win_s, hop_s)
  38
  39 # allocate memory to store norms and phases
  40 n_blocks = source_in.duration // hop_s + 1
  41 # adding an empty frame at end of spectrogram
  42 norms  = np.zeros((n_blocks + 1, win_s // 2 + 1), dtype = float_type)
  43 phases = np.zeros((n_blocks + 1, win_s // 2 + 1), dtype = float_type)
  44
  45 block_read = 0
  46 while True:
  47     # read from source
  48     samples, read = source_in()
  49     # compute fftgrain
  50     spec = p(samples)
  51     # store current grain
  52     norms[block_read] = spec.norm
  53     phases[block_read] = spec.phas
  54     # until end of file
  55     if read < hop_s: break
  56     # increment block counter
  57     block_read += 1
  58
  59 # just to make sure
  60 #source_in.close()
  61
  62 sink_out = sink(output_filename, samplerate)
  63
  64 # interpolated time steps (j = alpha * i)
  65 steps = np.arange(0, n_blocks, rate, dtype = float_type)
  66 # initial phase
  67 phas_acc = phases[0]
  68 # excepted phase advance in each bin
  69 phi_advance = np.linspace(0, np.pi * hop_s, win_s / 2 + 1).astype (float_type)
  70
  71 new_grain = cvec(win_s)
  72
  73 for (t, step) in enumerate(steps):
  74
  75     frac = 1. - np.mod(step, 1.0)
  76     # get pair of frames
  77     t_norms = norms[int(step):int(step+2)]
  78     t_phases = phases[int(step):int(step+2)]
  79
  80     # compute interpolated frame
  81     new_grain.norm = frac * t_norms[0] + (1. - frac) * t_norms[1]
  82     new_grain.phas = phas_acc
  83     #print t, step, new_grain.norm
  84     #print t, step, phas_acc
  85
  86     # psola
  87     samples = p.rdo(new_grain)
  88     if t > warmup: # skip the first few frames to warm up phase vocoder
  89         # write to sink
  90         sink_out(samples, hop_s)
  91
  92     # calculate phase advance
  93     dphas = t_phases[1] - t_phases[0] - phi_advance
  94     # unwrap angle to [-pi; pi]
  95     dphas = unwrap2pi(dphas)
  96     # cumulate phase, to be used for next frame
  97     phas_acc += phi_advance + dphas
  98
  99 for t in range(warmup + 1): # purge the last frames from the phase vocoder
 100     new_grain.norm[:] = 0
 101     new_grain.phas[:] = 0
 102     samples = p.rdo(new_grain)
 103     sink_out(samples, read if t == warmup else hop_s)
 104
 105 # just to make sure
 106 #sink_out.close()
 107
 108 format_out = "read {:d} blocks from {:s} at {:d}Hz and rate {:f}, wrote {:d} blocks to {:s}"
 109 print (format_out.format(block_read, source_filename, samplerate, rate,
 110     len(steps), output_filename))