python/demos/demo_timestretch_online.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 performs both analysis and synthesis in a single pass. See also
   7 # `demo_timestretch.py` for a version following the original implementation.
   8
   9 import sys
  10 from aubio import source, sink, pvoc, cvec
  11 from aubio import unwrap2pi, float_type
  12 import numpy as np
  13
  14 win_s = 512
  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 sink_out = sink(output_filename, samplerate)
  40
  41 # excepted phase advance in each bin
  42 phi_advance = np.linspace(0, np.pi * hop_s, win_s / 2 + 1).astype (float_type)
  43
  44 old_grain = cvec(win_s)
  45 new_grain = cvec(win_s)
  46
  47 block_read = 0
  48 interp_read = 0
  49 interp_block = 0
  50 while True:
  51
  52     samples, read = source_in()
  53     cur_grain = p(samples)
  54
  55     if block_read == 1:
  56         phas_acc = old_grain.phas
  57
  58     #print "block_read", block_read
  59     while True and (block_read > 0):
  60         if interp_read >= block_read:
  61             break
  62         #print "`--- interp_block:", interp_block,
  63         #print 'at orig_block', interp_read, '<- from', block_read - 1, block_read,
  64         #print 'old_grain', old_grain, 'cur_grain', cur_grain
  65         # time to compute interp grain
  66         frac = 1. - np.mod(interp_read, 1.0)
  67
  68         # compute interpolated frame
  69         new_grain.norm = frac * old_grain.norm + (1. - frac) * cur_grain.norm
  70         new_grain.phas = phas_acc
  71
  72         # psola
  73         samples = p.rdo(new_grain)
  74         if interp_read > warmup: # skip the first frames to warm up phase vocoder
  75             # write to sink
  76             sink_out(samples, hop_s)
  77
  78         # calculate phase advance
  79         dphas = cur_grain.phas - old_grain.phas - phi_advance
  80         # unwrap angle to [-pi; pi]
  81         dphas = unwrap2pi(dphas)
  82         # cumulate phase, to be used for next frame
  83         phas_acc += phi_advance + dphas
  84
  85         # prepare for next interp block
  86         interp_block += 1
  87         interp_read = interp_block * rate
  88         if interp_read >= block_read:
  89             break
  90
  91     # copy cur_grain to old_grain
  92     old_grain.norm = np.copy(cur_grain.norm)
  93     old_grain.phas = np.copy(cur_grain.phas)
  94
  95     # until end of file
  96     if read < hop_s: break
  97     # increment block counter
  98     block_read += 1
  99
 100 for t in range(warmup + 2): # purge the last frames from the phase vocoder
 101     new_grain.norm[:] = 0
 102     new_grain.phas[:] = 0
 103     samples = p.rdo(new_grain)
 104     sink_out(samples, read if t == warmup + 1 else hop_s)
 105
 106 # just to make sure
 107 source_in.close()
 108 sink_out.close()
 109
 110 format_out = "read {:d} blocks from {:s} at {:d}Hz and rate {:f}, wrote {:d} blocks to {:s}"
 111 print (format_out.format(block_read, source_filename, samplerate, rate,
 112     interp_block, output_filename))