linmdtw package

Submodules

linmdtw.dtw module

linmdtw.dtw.check_euclidean_inputs(X, Y)

Check the input of two time series in Euclidean spaces, which are to be warped to each other. They must satisfy: 1. They are in the same dimension space 2. They are 32-bit 3. They are in C-contiguous order

If #2 or #3 are not satisfied, automatically fix them and warn the user. Furthermore, warn the user if X or Y has more columns than rows, since the convention is that points are along rows and dimensions are along columns

Parameters

• X (ndarray(M, d)) – The first time series

• Y (ndarray(N, d)) – The second time series

Returns

• X (ndarray(M, d)) – The first time series, possibly copied in memory to be 32-bit, C-contiguous

• Y (ndarray(N, d)) – The second time series, possibly copied in memory to be 32-bit, C-contiguous

linmdtw.dtw.dtw_brute(X, Y, debug=False)

Compute brute force dynamic time warping between two time-ordered point clouds in Euclidean space, using cython on the backend

Parameters

• X (ndarray(M, d)) – A d-dimensional Euclidean point cloud with M points

• Y (ndarray(N, d)) – A d-dimensional Euclidean point cloud with N points

• debug (boolean) – Whether to keep track of debugging information

Returns

• { –

  ‘cost’: float

  The optimal cost of the alignment (if computation didn’t stop prematurely),

  ’U’/’L’/’UL’: ndarray(M, N)

  The choice matrices (if debugging),

  ’S’: ndarray(M, N)

  The accumulated cost matrix (if debugging)

• }

linmdtw.dtw.dtw_brute_backtrace(X, Y, debug=False)

Compute dynamic time warping between two time-ordered point clouds in Euclidean space, using cython on the backend. Then, trace back through the matrix of backpointers to extract an alignment path

Parameters

• X (ndarray(M, d)) – A d-dimensional Euclidean point cloud with M points

• Y (ndarray(N, d)) – A d-dimensional Euclidean point cloud with N points

• debug (boolean) – Whether to keep track of debugging information

Returns

• If not debugging – (float: cost, ndarray(K, 2): The warping path)

• If debugging

• { –

  ‘cost’: float

  The optimal cost of the alignment (if computation didn’t stop prematurely),

  ’U’/’L’/’UL’: ndarray(M, N)

  The choice matrices (if debugging),

  ’S’: ndarray(M, N)

  The accumulated cost matrix (if debugging)

  ’path’: ndarray(K, 2)

  The warping path

• }

linmdtw.dtw.dtw_diag(X, Y, k_save=- 1, k_stop=- 1, box=None, reverse=False, debug=False, metadata=None)

A CPU version of linear memory diagonal DTW

Parameters

• X (ndarray(M, d)) – An M-dimensional Euclidean point cloud

• Y (ndarray(N, d)) – An N-dimensional Euclidean point cloud

• k_save (int) – Index of the diagonal d2 at which to save d0, d1, and d2

• k_stop (int) – Index of the diagonal d2 at which to stop computation

• box (list) – A list of [startx, endx, starty, endy]

• reverse (boolean) – Whether we’re going in reverse

• debug (boolean) – Whether to save the accumulated cost matrix

• metadata (dictionary) – A dictionary for storing information about the computation

Returns

• { –

  ‘cost’: float

  The optimal cost of the alignment (if computation didn’t stop prematurely),

  ’U’/’L’/’UL’: ndarray(M, N)

  The choice matrices (if debugging),

  ’d0’/’d1’/’d2’:ndarray(min(M, N))

  The saved rows if a save index was chosen,

  ’csm0’/’csm1’/’csm2’:ndarray(min(M, N))

  The saved cross-similarity distances if a save index was chosen

• }

linmdtw.dtw.linmdtw(X, Y, box=None, min_dim=500, do_gpu=True, metadata=None)

Linear memory exact, parallelizable DTW

Parameters

• X (ndarray(N1, d)) – An N1-dimensional Euclidean point cloud

• Y (ndarray(N2, d)) – An N2-dimensional Euclidean point cloud

• min_dim (int) – If one of the dimensions of the rectangular region to the left or to the right is less than this number, then switch to brute force CPU

• do_gpu (boolean) – If true, use the GPU diagonal DTW function as a subroutine. Otherwise, use the CPU version. Both are linear memory, but the GPU will go faster for larger synchronization problems

• metadata (dictionary) – A dictionary for storing information about the computation

Returns

(float (cost, ndarray(K, 2): The optimal warping path))

linmdtw.dtwapprox module

linmdtw.dtwapprox.cdtw(X, Y, radius, debug=False, do_plot=False)

Dynamic time warping with constraints, as per the Sakoe-Chiba band

Parameters

• X (ndarray(M, d)) – A d-dimensional Euclidean point cloud with M points

• Y (ndarray(N, d)) – A d-dimensional Euclidean point cloud with N points

• radius (int) – How far away a warping path is allowed to stray from the identity map in either direction

• debug (boolean) – Whether to keep track of debugging information

• do_plot (boolean) – Whether to plot the warping path at each level and save to image files

Returns

(float (cost, ndarray(K, 2): The warping path))

linmdtw.dtwapprox.fastdtw(X, Y, radius, debug=False, level=0, do_plot=False)

An implementation of [1] [1] FastDTW: Toward Accurate Dynamic Time Warping in Linear Time and Space. Stan Salvador and Philip Chan

Parameters

• X (ndarray(M, d)) – A d-dimensional Euclidean point cloud with M points

• Y (ndarray(N, d)) – A d-dimensional Euclidean point cloud with N points

• radius (int) – Radius of the l-infinity box that determines sparsity structure at each level

• debug (boolean) – Whether to keep track of debugging information

• level (int) – An int for keeping track of the level of recursion

• do_plot (boolean) – Whether to plot the warping path at each level and save to image files

Returns

(float (cost, ndarray(K, 2): The warping path))

linmdtw.dtwapprox.fill_block(A, p, radius, val)

Fill a square block with values

Parameters

• A (ndarray(M, N) or sparse(M, N)) – The array to fill

• p (list of [i, j]) – The coordinates of the center of the box

• radius (int) – Half the width of the box

• val (float) – Value to fill in

linmdtw.dtwapprox.get_box_area(a1, a2)

Get the area of a box specified by two anchors

Parameters

• a1 (list(2)) – Row/column of first anchor

• a2 (list(2)) – Row/column of second anchor

Returns

Area of box determined by these two anchors

linmdtw.dtwapprox.mrmsdtw(X, Y, tau, debug=False, refine=True)

An implementation of the approximate, memory-restricted multiscale DTW technique from [2] [2] “Memory-Restricted Multiscale Dynamic Time Warping” Thomas Praetzlich, Jonathan Driedger and Meinard Mueller

Parameters

• X (ndarray(M, d)) – A d-dimensional Euclidean point cloud with M points

• Y (ndarray(N, d)) – A d-dimensional Euclidean point cloud with N points

• tau (int) – The max amount of cells to be in memory at any given time

• debug (boolean) – Whether to keep track of debugging information

• refine (boolean) – Whether to do refinement with the “white anchors”

Returns

path (ndarray(K, 2)) – The warping path

linmdtw.alignmenttools module

linmdtw.alignmenttools.get_alignment_area_dist(P1, P2, do_plot=False)

Compute area-based alignment error between two warping paths.

Parameters

• ndarray(M (P1) – First warping path

• 2) (P2) – First warping path

• ndarray(N (P2) – Second warping path

• 2) – Second warping path

• do_plot (boolean) – Whether to draw a plot showing the enclosed area

Returns

score (float) – The area score

linmdtw.alignmenttools.get_alignment_row_col_dists(P1, P2)

A measurement of errors between two warping paths. For each point in the first path, record the distance of the closest point in the same row on the second path, and vice versa. Then repeat this along the columns

Parameters

• P1 (ndarray(M, 2)) – Ground truth warping path

• P2 (ndarray(N, 2)) – Test warping path

Returns

dists (ndarray(2M+2N)) – The errors

linmdtw.alignmenttools.get_alignment_row_dists(P1, P2)

A measurement of errors between two warping paths. For each point in the first path, record the distance of the closest point in the same row on the second path

Parameters

• P1 (ndarray(M, 2)) – Ground truth warping path

• P2 (ndarray(N, 2)) – Test warping path

Returns

dists (ndarray(M)) – The errors at each point on the first warping path

linmdtw.alignmenttools.get_csm(X, Y)

Return the Euclidean cross-similarity matrix between X and Y

Parameters

• X (ndarray(M, d)) – A d-dimensional Euclidean point cloud with M points

• Y (ndarray(N, d)) – A d-dimensional Euclidean point cloud with N points

Returns

D (ndarray(M, N)) – The cross-similarity matrix

linmdtw.alignmenttools.get_diag_indices(MTotal, NTotal, k, box=None, reverse=False)

Compute the indices on a diagonal into indices on an accumulated distance matrix

Parameters

• MTotal (int) – Total number of samples in X

• NTotal (int) – Total number of samples in Y

• k (int) – Index of the diagonal

• box (list [XStart, XEnd, YStart, YEnd]) – The coordinates of the box in which to search

Returns

• i (ndarray(dim)) – Row indices

• j (ndarray(dim)) – Column indices

linmdtw.alignmenttools.get_diag_len(box, k)

Return the number of elements in this particular diagonal

Parameters

• MTotal (int) – Total number of samples in X

• NTotal (int) – Total number of samples in Y

• k (int) – Index of the diagonal

Returns

Number of elements

linmdtw.alignmenttools.get_interpolated_euclidean_timeseries(X, t, kind='linear')

Resample a time series in Euclidean space using interp2d

Parameters

• X (ndarray(M, d)) – The Euclidean time series with n points

• t (ndarray(N)) – A re-parameterization function on the unit interval [0, 1]

• kind (string) – The kind of interpolation to do

Returns

Y (ndarray(N, d)) – The interpolated time series

linmdtw.alignmenttools.get_inverse_fn_equally_sampled(t, x)

Compute the inverse of a 1D function and equally sample it.

Parameters

• t (ndarray(N)) – The domain samples of the original function

• x (ndarray(N)) – The range samples of the original function

Returns

y (ndarray(N)) – The inverse function samples

linmdtw.alignmenttools.get_parameterization_dict(N, do_plot=False)

Construct a dictionary of different types of parameterizations on the unit interval

Parameters

• N (int) – Number of samples on the unit interval

• do_plot (boolean) – Whether to plot all of the parameterizations

Returns

D (ndarray(N, K)) – The dictionary of parameterizations, with each warping path down a different column

linmdtw.alignmenttools.get_path_cost(X, Y, path)

Return the cost of a warping path that matches two Euclidean point clouds

Parameters

• X (ndarray(M, d)) – A d-dimensional Euclidean point cloud with M points

• Y (ndarray(N, d)) – A d-dimensional Euclidean point cloud with N points

• P1 (ndarray(K, 2)) – Warping path

Returns

cost (float) – The sum of the Euclidean distances along the warping path between X and Y

linmdtw.alignmenttools.get_ssm(X)

Return the SSM between all rows of a time-ordered Euclidean point cloud X

Parameters

X (ndarray(M, d)) – A d-dimensional Euclidean point cloud with M points

Returns

D (ndarray(M, M)) – The self-similarity matrix

linmdtw.alignmenttools.make_path_strictly_increase(path)

Given a warping path, remove all rows that do not strictly increase from the row before

linmdtw.alignmenttools.param_to_warppath(t, M)

Convert a parameterization function into a valid warping path

Parameters

• t (ndarray(N)) – Samples along the parameterization function on the unit interval

• M (int) – The number of samples in the original time series

Returns

P (ndarray(K, 2)) – A warping path that best matches the parameterization

linmdtw.alignmenttools.refine_warping_path(path)

An implementation of the technique in section 4 of “Refinement Strategies for Music Synchronization” by Ewert and Müller

Parameters

path (ndarray(K, 2)) – A warping path

Returns

path_refined (ndarray(N >= K, 2)) – A refined set of correspondences

linmdtw.alignmenttools.sample_parameterization_dict(D, k, do_plot=False)

Return a warping path made up of k random elements drawn from dictionary D

Parameters

• D (ndarray(N, K)) – The dictionary of warping paths, with each warping path down a different column

• k (int) – The number of warping paths to take in a combination

linmdtw.alignmenttools.update_alignment_metadata(metadata=None, newcells=0)

Add new amount of cells to the total cells processed, and print out a percentage point if there’s been progress

Parameters

• newcells (int) – The number of new cells that are being processed

• metadata (dictionary) – Dictionary with ‘M’, ‘N’, ‘totalCells’, all ints and ‘timeStart’

linmdtw.audiotools module

linmdtw.audiotools.get_DLNC0(x, sr, hop_length, lag=10, do_plot=False)

Compute decaying locally adaptive normalize C0 (DLNC0) features

Parameters

• x (ndarray(N)) – Audio samples

• sr (int) – Sample rate

• hop_length (int) – Hop size between windows

• lag (int) – Number of lags to include

Returns

X (ndarray(n_win, 12)) – The DLNC0 features

linmdtw.audiotools.get_mfcc_mod(x, sr, hop_length, n_mfcc=120, drop=20, n_fft=2048)

Compute the mfcc_mod features, as described in Gadermaier 2019

Parameters

• x (ndarray(N)) – Audio samples

• sr (int) – Sample rate

• hop_length (int) – Hop size between windows

• n_mfcc (int) – Number of mfcc coefficients to compute

• drop (int) – Index under which to ignore coefficients

• n_fft (int) – Number of fft points to use in each window

Returns

X (ndarray(n_win, n_mfcc-drop)) – The mfcc-mod features

linmdtw.audiotools.get_mixed_DLNC0_CENS(x, sr, hop_length, lam=0.1)

Concatenate DLNC0 to CENS

Parameters

• x (ndarray(N)) – Audio samples

• sr (int) – Sample rate

• hop_length (int) – Hop size between windows

• lam (float) – The coefficient in front of the CENS features

Returns

X (ndarray(n_win, 24)) – DLNC0 features along the first 12 columns, weighted CENS along the next 12 columns

linmdtw.audiotools.load_audio(filename, sr=44100)

Load an audio waveform from a file. Try to use ffmpeg to convert it to a .wav file so scipy’s fast wavfile loader can work. Otherwise, fall back to the slower librosa

Parameters

• filename (string) – Path to audio file to load

• sr (int) – Sample rate to use

Returns

• y (ndarray(N)) – Audio samples

• sr (int) – The sample rate that was actually used

linmdtw.audiotools.save_audio(x, sr, outprefix)

Save audio to a file

Parameters

• x (ndarray(N, 2)) – Stereo audio to save

• sr (int) – Sample rate of audio to save

• outprefix (string) – Use this as the prefix of the file to which to save audio

linmdtw.audiotools.stretch_audio(x1, x2, sr, path, hop_length, refine=True)

Wrap around pyrubberband to warp one audio stream to another, according to some warping path

Parameters

• x1 (ndarray(M)) – First audio stream

• x2 (ndarray(N)) – Second audio stream

• sr (int) – Sample rate

• path (ndarray(P, 2)) – Warping path, in units of windows

• hop_length (int) – The hop length between windows

• refine (boolean) – Whether to refine the warping path before alignment

Returns

x3 (ndarray(N, 2)) – The synchronized audio. x2 is in the right channel, and x1 stretched to x2 is in the left channel

Module contents