Prompt48 commited on
Commit
0b7a138
·
verified ·
1 Parent(s): b71845e

Upload edit\Qwen3-TTS-test\.venv\Lib\site-packages\librosa\sequence.py with huggingface_hub

Browse files
edit//Qwen3-TTS-test//.venv//Lib//site-packages//librosa//sequence.py ADDED
@@ -0,0 +1,2051 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
+ #!/usr/bin/env python
2
+ # -*- encoding: utf-8 -*-
3
+ """
4
+ Sequential modeling
5
+ ===================
6
+
7
+ Sequence alignment
8
+ ------------------
9
+ .. autosummary::
10
+ :toctree: generated/
11
+
12
+ dtw
13
+ rqa
14
+
15
+ Viterbi decoding
16
+ ----------------
17
+ .. autosummary::
18
+ :toctree: generated/
19
+
20
+ viterbi
21
+ viterbi_discriminative
22
+ viterbi_binary
23
+
24
+ Transition matrices
25
+ -------------------
26
+ .. autosummary::
27
+ :toctree: generated/
28
+
29
+ transition_uniform
30
+ transition_loop
31
+ transition_cycle
32
+ transition_local
33
+ """
34
+ from __future__ import annotations
35
+
36
+ import numpy as np
37
+ from scipy.spatial.distance import cdist
38
+ from numba import jit
39
+ from .util import pad_center, fill_off_diagonal, is_positive_int, tiny, expand_to
40
+ from .util.exceptions import ParameterError
41
+ from .filters import get_window
42
+ from typing import Any, Iterable, List, Optional, Tuple, Union, overload
43
+ from typing_extensions import Literal
44
+ from ._typing import _WindowSpec
45
+
46
+ __all__ = [
47
+ "dtw",
48
+ "dtw_backtracking",
49
+ "rqa",
50
+ "viterbi",
51
+ "viterbi_discriminative",
52
+ "viterbi_binary",
53
+ "transition_uniform",
54
+ "transition_loop",
55
+ "transition_cycle",
56
+ "transition_local",
57
+ ]
58
+
59
+
60
+ @overload
61
+ def dtw(
62
+ X: np.ndarray,
63
+ Y: np.ndarray,
64
+ *,
65
+ metric: str = ...,
66
+ step_sizes_sigma: Optional[np.ndarray] = ...,
67
+ weights_add: Optional[np.ndarray] = ...,
68
+ weights_mul: Optional[np.ndarray] = ...,
69
+ subseq: bool = ...,
70
+ backtrack: Literal[False],
71
+ global_constraints: bool = ...,
72
+ band_rad: float = ...,
73
+ return_steps: Literal[False] = ...,
74
+ ) -> np.ndarray:
75
+ ...
76
+
77
+
78
+ @overload
79
+ def dtw(
80
+ *,
81
+ C: np.ndarray,
82
+ metric: str = ...,
83
+ step_sizes_sigma: Optional[np.ndarray] = ...,
84
+ weights_add: Optional[np.ndarray] = ...,
85
+ weights_mul: Optional[np.ndarray] = ...,
86
+ subseq: bool = ...,
87
+ backtrack: Literal[False],
88
+ global_constraints: bool = ...,
89
+ band_rad: float = ...,
90
+ return_steps: Literal[False] = ...,
91
+ ) -> np.ndarray:
92
+ ...
93
+
94
+
95
+ @overload
96
+ def dtw(
97
+ X: np.ndarray,
98
+ Y: np.ndarray,
99
+ *,
100
+ metric: str = ...,
101
+ step_sizes_sigma: Optional[np.ndarray] = ...,
102
+ weights_add: Optional[np.ndarray] = ...,
103
+ weights_mul: Optional[np.ndarray] = ...,
104
+ subseq: bool = ...,
105
+ backtrack: Literal[False],
106
+ global_constraints: bool = ...,
107
+ band_rad: float = ...,
108
+ return_steps: Literal[True],
109
+ ) -> Tuple[np.ndarray, np.ndarray]:
110
+ ...
111
+
112
+
113
+ @overload
114
+ def dtw(
115
+ *,
116
+ C: np.ndarray,
117
+ metric: str = ...,
118
+ step_sizes_sigma: Optional[np.ndarray] = ...,
119
+ weights_add: Optional[np.ndarray] = ...,
120
+ weights_mul: Optional[np.ndarray] = ...,
121
+ subseq: bool = ...,
122
+ backtrack: Literal[False],
123
+ global_constraints: bool = ...,
124
+ band_rad: float = ...,
125
+ return_steps: Literal[True],
126
+ ) -> Tuple[np.ndarray, np.ndarray]:
127
+ ...
128
+
129
+
130
+ @overload
131
+ def dtw(
132
+ X: np.ndarray,
133
+ Y: np.ndarray,
134
+ *,
135
+ metric: str = ...,
136
+ step_sizes_sigma: Optional[np.ndarray] = ...,
137
+ weights_add: Optional[np.ndarray] = ...,
138
+ weights_mul: Optional[np.ndarray] = ...,
139
+ subseq: bool = ...,
140
+ backtrack: Literal[True] = ...,
141
+ global_constraints: bool = ...,
142
+ band_rad: float = ...,
143
+ return_steps: Literal[False] = ...,
144
+ ) -> Tuple[np.ndarray, np.ndarray]:
145
+ ...
146
+
147
+
148
+ @overload
149
+ def dtw(
150
+ *,
151
+ C: np.ndarray,
152
+ metric: str = ...,
153
+ step_sizes_sigma: Optional[np.ndarray] = ...,
154
+ weights_add: Optional[np.ndarray] = ...,
155
+ weights_mul: Optional[np.ndarray] = ...,
156
+ subseq: bool = ...,
157
+ backtrack: Literal[True] = ...,
158
+ global_constraints: bool = ...,
159
+ band_rad: float = ...,
160
+ return_steps: Literal[False] = ...,
161
+ ) -> Tuple[np.ndarray, np.ndarray]:
162
+ ...
163
+
164
+
165
+ @overload
166
+ def dtw(
167
+ X: np.ndarray,
168
+ Y: np.ndarray,
169
+ *,
170
+ metric: str = ...,
171
+ step_sizes_sigma: Optional[np.ndarray] = ...,
172
+ weights_add: Optional[np.ndarray] = ...,
173
+ weights_mul: Optional[np.ndarray] = ...,
174
+ subseq: bool = ...,
175
+ backtrack: Literal[True] = ...,
176
+ global_constraints: bool = ...,
177
+ band_rad: float = ...,
178
+ return_steps: Literal[True],
179
+ ) -> Tuple[np.ndarray, np.ndarray, np.ndarray]:
180
+ ...
181
+
182
+
183
+ @overload
184
+ def dtw(
185
+ *,
186
+ C: np.ndarray,
187
+ metric: str = ...,
188
+ step_sizes_sigma: Optional[np.ndarray] = ...,
189
+ weights_add: Optional[np.ndarray] = ...,
190
+ weights_mul: Optional[np.ndarray] = ...,
191
+ subseq: bool = ...,
192
+ backtrack: Literal[True] = ...,
193
+ global_constraints: bool = ...,
194
+ band_rad: float = ...,
195
+ return_steps: Literal[True],
196
+ ) -> Tuple[np.ndarray, np.ndarray, np.ndarray]:
197
+ ...
198
+
199
+
200
+ def dtw(
201
+ X: Optional[np.ndarray] = None,
202
+ Y: Optional[np.ndarray] = None,
203
+ *,
204
+ C: Optional[np.ndarray] = None,
205
+ metric: str = "euclidean",
206
+ step_sizes_sigma: Optional[np.ndarray] = None,
207
+ weights_add: Optional[np.ndarray] = None,
208
+ weights_mul: Optional[np.ndarray] = None,
209
+ subseq: bool = False,
210
+ backtrack: bool = True,
211
+ global_constraints: bool = False,
212
+ band_rad: float = 0.25,
213
+ return_steps: bool = False,
214
+ ) -> Union[
215
+ np.ndarray, Tuple[np.ndarray, np.ndarray], Tuple[np.ndarray, np.ndarray, np.ndarray]
216
+ ]:
217
+ """Dynamic time warping (DTW).
218
+
219
+ This function performs a DTW and path backtracking on two sequences.
220
+ We follow the nomenclature and algorithmic approach as described in [#]_.
221
+
222
+ .. [#] Meinard Mueller
223
+ Fundamentals of Music Processing — Audio, Analysis, Algorithms, Applications
224
+ Springer Verlag, ISBN: 978-3-319-21944-8, 2015.
225
+
226
+ Parameters
227
+ ----------
228
+ X : np.ndarray [shape=(..., K, N)]
229
+ audio feature matrix (e.g., chroma features)
230
+
231
+ If ``X`` has more than two dimensions (e.g., for multi-channel inputs), all leading
232
+ dimensions are used when computing distance to ``Y``.
233
+
234
+ Y : np.ndarray [shape=(..., K, M)]
235
+ audio feature matrix (e.g., chroma features)
236
+
237
+ C : np.ndarray [shape=(N, M)]
238
+ Precomputed distance matrix. If supplied, X and Y must not be supplied and
239
+ ``metric`` will be ignored.
240
+
241
+ metric : str
242
+ Identifier for the cost-function as documented
243
+ in `scipy.spatial.distance.cdist()`
244
+
245
+ step_sizes_sigma : np.ndarray [shape=[n, 2]]
246
+ Specifies allowed step sizes as used by the DTW.
247
+
248
+ weights_add : np.ndarray [shape=[n, ]]
249
+ Additive weights to penalize certain step sizes.
250
+
251
+ weights_mul : np.ndarray [shape=[n, ]]
252
+ Multiplicative weights to penalize certain step sizes.
253
+
254
+ subseq : bool
255
+ Enable subsequence DTW, e.g., for retrieval tasks.
256
+
257
+ backtrack : bool
258
+ Enable backtracking in accumulated cost matrix.
259
+
260
+ global_constraints : bool
261
+ Applies global constraints to the cost matrix ``C`` (Sakoe-Chiba band).
262
+
263
+ band_rad : float
264
+ The Sakoe-Chiba band radius (1/2 of the width) will be
265
+ ``int(radius*min(C.shape))``.
266
+
267
+ return_steps : bool
268
+ If true, the function returns ``steps``, the step matrix, containing
269
+ the indices of the used steps from the cost accumulation step.
270
+
271
+ Returns
272
+ -------
273
+ D : np.ndarray [shape=(N, M)]
274
+ accumulated cost matrix.
275
+ The value at the final index position ``D[-1, -1]`` is the total alignment cost.
276
+ wp : np.ndarray [shape=(L, 2)]
277
+ Warping path with index pairs.
278
+ Each row of the array contains an index pair (n, m).
279
+ Only returned when ``backtrack`` is True.
280
+ Note that the length ``L`` of the warping path need not match the
281
+ lengths of the input data, depending on the ``step_sizes_sigma`` values
282
+ and ``subseq``.
283
+ steps : np.ndarray [shape=(N, M)]
284
+ Step matrix, containing the indices of the used steps from the cost
285
+ accumulation step.
286
+ Only returned when ``return_steps`` is True.
287
+
288
+ Raises
289
+ ------
290
+ ParameterError
291
+ If you are doing diagonal matching and Y is shorter than X or if an
292
+ incompatible combination of X, Y, and C are supplied.
293
+
294
+ If your input dimensions are incompatible.
295
+
296
+ If the cost matrix has NaN values.
297
+
298
+ Examples
299
+ --------
300
+ >>> import numpy as np
301
+ >>> import matplotlib.pyplot as plt
302
+ >>> y, sr = librosa.load(librosa.ex('brahms'), offset=10, duration=15)
303
+ >>> X = librosa.feature.chroma_cens(y=y, sr=sr)
304
+ >>> noise = np.random.rand(X.shape[0], 200)
305
+ >>> Y = np.concatenate((noise, noise, X, noise), axis=1)
306
+ >>> D, wp = librosa.sequence.dtw(X, Y, subseq=True)
307
+ >>> fig, ax = plt.subplots(nrows=2, sharex=True)
308
+ >>> img = librosa.display.specshow(D, x_axis='frames', y_axis='frames',
309
+ ... ax=ax[0])
310
+ >>> ax[0].set(title='DTW cost', xlabel='Noisy sequence', ylabel='Target')
311
+ >>> ax[0].plot(wp[:, 1], wp[:, 0], label='Optimal path', color='y')
312
+ >>> ax[0].legend()
313
+ >>> fig.colorbar(img, ax=ax[0])
314
+ >>> ax[1].plot(D[-1, :] / wp.shape[0])
315
+ >>> ax[1].set(xlim=[0, Y.shape[1]], ylim=[0, 2],
316
+ ... title='Matching cost function')
317
+ """
318
+ # Default Parameters
319
+ default_steps = np.array([[1, 1], [0, 1], [1, 0]], dtype=np.uint32)
320
+ default_weights_add = np.zeros(3, dtype=np.float64)
321
+ default_weights_mul = np.ones(3, dtype=np.float64)
322
+
323
+ if step_sizes_sigma is None:
324
+ # Use the default steps
325
+ step_sizes_sigma = default_steps
326
+
327
+ # Use default weights if none are provided
328
+ if weights_add is None:
329
+ weights_add = default_weights_add
330
+
331
+ if weights_mul is None:
332
+ weights_mul = default_weights_mul
333
+ else:
334
+ # If we have custom steps but no weights, construct them here
335
+ if weights_add is None:
336
+ weights_add = np.zeros(len(step_sizes_sigma), dtype=np.float64)
337
+
338
+ if weights_mul is None:
339
+ weights_mul = np.ones(len(step_sizes_sigma), dtype=np.float64)
340
+
341
+ # Make the default step weights infinite so that they are never
342
+ # preferred over custom steps
343
+ default_weights_add.fill(np.inf)
344
+ default_weights_mul.fill(np.inf)
345
+
346
+ # Append custom steps and weights to our defaults
347
+ step_sizes_sigma = np.concatenate((default_steps, step_sizes_sigma))
348
+ weights_add = np.concatenate((default_weights_add, weights_add))
349
+ weights_mul = np.concatenate((default_weights_mul, weights_mul))
350
+
351
+ # These asserts are bad, but mypy cannot trace the code paths properly
352
+ assert step_sizes_sigma is not None
353
+ assert weights_add is not None
354
+ assert weights_mul is not None
355
+
356
+ if np.any(step_sizes_sigma < 0):
357
+ raise ParameterError("step_sizes_sigma cannot contain negative values")
358
+
359
+ if len(step_sizes_sigma) != len(weights_add):
360
+ raise ParameterError("len(weights_add) must be equal to len(step_sizes_sigma)")
361
+ if len(step_sizes_sigma) != len(weights_mul):
362
+ raise ParameterError("len(weights_mul) must be equal to len(step_sizes_sigma)")
363
+
364
+ if C is None and (X is None or Y is None):
365
+ raise ParameterError("If C is not supplied, both X and Y must be supplied")
366
+ if C is not None and (X is not None or Y is not None):
367
+ raise ParameterError("If C is supplied, both X and Y must not be supplied")
368
+
369
+ c_is_transposed = False
370
+
371
+ # calculate pair-wise distances, unless already supplied.
372
+ # C_local will keep track of whether the distance matrix was supplied
373
+ # by the user (False) or constructed locally (True)
374
+ C_local = False
375
+ if C is None:
376
+ C_local = True
377
+ # mypy can't figure out that this case does not happen
378
+ assert X is not None and Y is not None
379
+ # take care of dimensions
380
+ X = np.atleast_2d(X)
381
+ Y = np.atleast_2d(Y)
382
+
383
+ # Perform some shape-squashing here
384
+ # Put the time axes around front
385
+ # Suppress types because mypy doesn't know these are ndarrays
386
+ X = np.swapaxes(X, -1, 0) # type: ignore
387
+ Y = np.swapaxes(Y, -1, 0) # type: ignore
388
+
389
+ # Flatten the remaining dimensions
390
+ # Use F-ordering to preserve columns
391
+ X = X.reshape((X.shape[0], -1), order="F")
392
+ Y = Y.reshape((Y.shape[0], -1), order="F")
393
+
394
+ try:
395
+ C = cdist(X, Y, metric=metric)
396
+ except ValueError as exc:
397
+ raise ParameterError(
398
+ "scipy.spatial.distance.cdist returned an error.\n"
399
+ "Please provide your input in the form X.shape=(K, N) "
400
+ "and Y.shape=(K, M).\n 1-dimensional sequences should "
401
+ "be reshaped to X.shape=(1, N) and Y.shape=(1, M)."
402
+ ) from exc
403
+
404
+ # for subsequence matching:
405
+ # if N > M, Y can be a subsequence of X
406
+ if subseq and (X.shape[0] > Y.shape[0]):
407
+ C = C.T
408
+ c_is_transposed = True
409
+
410
+ C = np.atleast_2d(C)
411
+
412
+ # if diagonal matching, Y has to be longer than X
413
+ # (X simply cannot be contained in Y)
414
+ if np.array_equal(step_sizes_sigma, np.array([[1, 1]])) and (
415
+ C.shape[0] > C.shape[1]
416
+ ):
417
+ raise ParameterError(
418
+ "For diagonal matching: Y.shape[-1] >= X.shape[-11] "
419
+ "(C.shape[1] >= C.shape[0])"
420
+ )
421
+
422
+ max_0 = step_sizes_sigma[:, 0].max()
423
+ max_1 = step_sizes_sigma[:, 1].max()
424
+
425
+ # check C here for nans before building global constraints
426
+ if np.any(np.isnan(C)):
427
+ raise ParameterError("DTW cost matrix C has NaN values. ")
428
+
429
+ if global_constraints:
430
+ # Apply global constraints to the cost matrix
431
+ if not C_local:
432
+ # If C was provided as input, make a copy here
433
+ C = np.copy(C)
434
+ fill_off_diagonal(C, radius=band_rad, value=np.inf)
435
+
436
+ # initialize whole matrix with infinity values
437
+ D = np.ones(C.shape + np.array([max_0, max_1])) * np.inf
438
+
439
+ # set starting point to C[0, 0]
440
+ D[max_0, max_1] = C[0, 0]
441
+
442
+ if subseq:
443
+ D[max_0, max_1:] = C[0, :]
444
+
445
+ # initialize step matrix with -1
446
+ # will be filled in calc_accu_cost() with indices from step_sizes_sigma
447
+ steps = np.zeros(D.shape, dtype=np.int32)
448
+
449
+ # these steps correspond to left- (first row) and up-(first column) moves
450
+ steps[0, :] = 1
451
+ steps[:, 0] = 2
452
+
453
+ # calculate accumulated cost matrix
454
+ D: np.ndarray
455
+ steps: np.ndarray
456
+ D, steps = __dtw_calc_accu_cost(
457
+ C, D, steps, step_sizes_sigma, weights_mul, weights_add, max_0, max_1
458
+ )
459
+
460
+ # delete infinity rows and columns
461
+ D = D[max_0:, max_1:]
462
+ steps = steps[max_0:, max_1:]
463
+
464
+ return_values: List[np.ndarray]
465
+ if backtrack:
466
+ wp: np.ndarray
467
+ if subseq:
468
+ if np.all(np.isinf(D[-1])):
469
+ raise ParameterError(
470
+ "No valid sub-sequence warping path could "
471
+ "be constructed with the given step sizes."
472
+ )
473
+ start = np.argmin(D[-1, :])
474
+ _wp = __dtw_backtracking(steps, step_sizes_sigma, subseq, start)
475
+ else:
476
+ # perform warping path backtracking
477
+ if np.isinf(D[-1, -1]):
478
+ raise ParameterError(
479
+ "No valid sub-sequence warping path could "
480
+ "be constructed with the given step sizes."
481
+ )
482
+
483
+ _wp = __dtw_backtracking(steps, step_sizes_sigma, subseq)
484
+ if _wp[-1] != (0, 0):
485
+ raise ParameterError(
486
+ "Unable to compute a full DTW warping path. "
487
+ "You may want to try again with subseq=True."
488
+ )
489
+
490
+ wp = np.asarray(_wp, dtype=int)
491
+
492
+ # since we transposed in the beginning, we have to adjust the index pairs back
493
+ if subseq and (
494
+ (X is not None and Y is not None and X.shape[0] > Y.shape[0])
495
+ or c_is_transposed
496
+ or C.shape[0] > C.shape[1]
497
+ ):
498
+ wp = np.fliplr(wp)
499
+ return_values = [D, wp]
500
+ else:
501
+ return_values = [D]
502
+
503
+ if return_steps:
504
+ return_values.append(steps)
505
+
506
+ if len(return_values) > 1:
507
+ # Suppressing type check here because mypy can't
508
+ # infer the exact length of the tuple
509
+ return tuple(return_values) # type: ignore
510
+ else:
511
+ return return_values[0]
512
+
513
+
514
+ @jit(nopython=True, cache=True) # type: ignore
515
+ def __dtw_calc_accu_cost(
516
+ C: np.ndarray,
517
+ D: np.ndarray,
518
+ steps: np.ndarray,
519
+ step_sizes_sigma: np.ndarray,
520
+ weights_mul: np.ndarray,
521
+ weights_add: np.ndarray,
522
+ max_0: int,
523
+ max_1: int,
524
+ ) -> Tuple[np.ndarray, np.ndarray]: # pragma: no cover
525
+ """Calculate the accumulated cost matrix D.
526
+
527
+ Use dynamic programming to calculate the accumulated costs.
528
+
529
+ Parameters
530
+ ----------
531
+ C : np.ndarray [shape=(N, M)]
532
+ pre-computed cost matrix
533
+ D : np.ndarray [shape=(N, M)]
534
+ accumulated cost matrix
535
+ steps : np.ndarray [shape=(N, M)]
536
+ Step matrix, containing the indices of the used steps from the cost
537
+ accumulation step.
538
+ step_sizes_sigma : np.ndarray [shape=[n, 2]]
539
+ Specifies allowed step sizes as used by the dtw.
540
+ weights_add : np.ndarray [shape=[n, ]]
541
+ Additive weights to penalize certain step sizes.
542
+ weights_mul : np.ndarray [shape=[n, ]]
543
+ Multiplicative weights to penalize certain step sizes.
544
+ max_0 : int
545
+ maximum number of steps in step_sizes_sigma in dim 0.
546
+ max_1 : int
547
+ maximum number of steps in step_sizes_sigma in dim 1.
548
+
549
+ Returns
550
+ -------
551
+ D : np.ndarray [shape=(N, M)]
552
+ accumulated cost matrix.
553
+ D[N, M] is the total alignment cost.
554
+ When doing subsequence DTW, D[N,:] indicates a matching function.
555
+ steps : np.ndarray [shape=(N, M)]
556
+ Step matrix, containing the indices of the used steps from the cost
557
+ accumulation step.
558
+
559
+ See Also
560
+ --------
561
+ dtw
562
+ """
563
+ for cur_n in range(max_0, D.shape[0]):
564
+ for cur_m in range(max_1, D.shape[1]):
565
+ # accumulate costs
566
+ for cur_step_idx, cur_w_add, cur_w_mul in zip(
567
+ range(step_sizes_sigma.shape[0]), weights_add, weights_mul
568
+ ):
569
+ cur_D = D[
570
+ cur_n - step_sizes_sigma[cur_step_idx, 0],
571
+ cur_m - step_sizes_sigma[cur_step_idx, 1],
572
+ ]
573
+ cur_C = cur_w_mul * C[cur_n - max_0, cur_m - max_1]
574
+ cur_C += cur_w_add
575
+ cur_cost = cur_D + cur_C
576
+
577
+ # check if cur_cost is smaller than the one stored in D
578
+ if cur_cost < D[cur_n, cur_m]:
579
+ D[cur_n, cur_m] = cur_cost
580
+
581
+ # save step-index
582
+ steps[cur_n, cur_m] = cur_step_idx
583
+
584
+ return D, steps
585
+
586
+
587
+ @jit(nopython=True, cache=True) # type: ignore
588
+ def __dtw_backtracking(
589
+ steps: np.ndarray,
590
+ step_sizes_sigma: np.ndarray,
591
+ subseq: bool,
592
+ start: Optional[int] = None,
593
+ ) -> List[Tuple[int, int]]: # pragma: no cover
594
+ """Backtrack optimal warping path.
595
+
596
+ Uses the saved step sizes from the cost accumulation
597
+ step to backtrack the index pairs for an optimal
598
+ warping path.
599
+
600
+ Parameters
601
+ ----------
602
+ steps : np.ndarray [shape=(N, M)]
603
+ Step matrix, containing the indices of the used steps from the cost
604
+ accumulation step.
605
+ step_sizes_sigma : np.ndarray [shape=[n, 2]]
606
+ Specifies allowed step sizes as used by the dtw.
607
+ subseq : bool
608
+ Enable subsequence DTW, e.g., for retrieval tasks.
609
+ start : int
610
+ Start column index for backtraing (only allowed for ``subseq=True``)
611
+
612
+ Returns
613
+ -------
614
+ wp : list [shape=(N,)]
615
+ Warping path with index pairs.
616
+ Each list entry contains an index pair
617
+ (n, m) as a tuple
618
+
619
+ See Also
620
+ --------
621
+ dtw
622
+ """
623
+ if start is None:
624
+ cur_idx = (steps.shape[0] - 1, steps.shape[1] - 1)
625
+ else:
626
+ cur_idx = (steps.shape[0] - 1, start)
627
+
628
+ wp = []
629
+ # Set starting point D(N, M) and append it to the path
630
+ wp.append((cur_idx[0], cur_idx[1]))
631
+
632
+ # Loop backwards.
633
+ # Stop criteria:
634
+ # Setting it to (0, 0) does not work for the subsequence dtw,
635
+ # so we only ask to reach the first row of the matrix.
636
+
637
+ while (subseq and cur_idx[0] > 0) or (not subseq and cur_idx != (0, 0)):
638
+ cur_step_idx = steps[(cur_idx[0], cur_idx[1])]
639
+
640
+ # save tuple with minimal acc. cost in path
641
+ cur_idx = (
642
+ cur_idx[0] - step_sizes_sigma[cur_step_idx][0],
643
+ cur_idx[1] - step_sizes_sigma[cur_step_idx][1],
644
+ )
645
+
646
+ # If we run off the side of the cost matrix, break here
647
+ if min(cur_idx) < 0:
648
+ break
649
+
650
+ # append to warping path
651
+ wp.append((cur_idx[0], cur_idx[1]))
652
+
653
+ return wp
654
+
655
+
656
+ def dtw_backtracking(
657
+ steps: np.ndarray,
658
+ *,
659
+ step_sizes_sigma: Optional[np.ndarray] = None,
660
+ subseq: bool = False,
661
+ start: Optional[Union[int, np.integer[Any]]] = None,
662
+ ) -> np.ndarray:
663
+ """Backtrack a warping path.
664
+
665
+ Uses the saved step sizes from the cost accumulation
666
+ step to backtrack the index pairs for a warping path.
667
+
668
+ Parameters
669
+ ----------
670
+ steps : np.ndarray [shape=(N, M)]
671
+ Step matrix, containing the indices of the used steps from the cost
672
+ accumulation step.
673
+ step_sizes_sigma : np.ndarray [shape=[n, 2]]
674
+ Specifies allowed step sizes as used by the dtw.
675
+ subseq : bool
676
+ Enable subsequence DTW, e.g., for retrieval tasks.
677
+ start : int
678
+ Start column index for backtraing (only allowed for ``subseq=True``)
679
+
680
+ Returns
681
+ -------
682
+ wp : list [shape=(N,)]
683
+ Warping path with index pairs.
684
+ Each list entry contains an index pair
685
+ (n, m) as a tuple
686
+
687
+ See Also
688
+ --------
689
+ dtw
690
+ """
691
+ if subseq is False and start is not None:
692
+ raise ParameterError(
693
+ f"start is only allowed to be set if subseq is True (start={start}, subseq={subseq})"
694
+ )
695
+
696
+ # Default Parameters
697
+ default_steps = np.array([[1, 1], [0, 1], [1, 0]], dtype=np.uint32)
698
+
699
+ if step_sizes_sigma is None:
700
+ # Use the default steps
701
+ step_sizes_sigma = default_steps
702
+ else:
703
+ # Append custom steps and weights to our defaults
704
+ step_sizes_sigma = np.concatenate((default_steps, step_sizes_sigma))
705
+
706
+ wp = __dtw_backtracking(steps, step_sizes_sigma, subseq, start)
707
+ return np.asarray(wp, dtype=int)
708
+
709
+
710
+ @overload
711
+ def rqa(
712
+ sim: np.ndarray,
713
+ *,
714
+ gap_onset: float = ...,
715
+ gap_extend: float = ...,
716
+ knight_moves: bool = ...,
717
+ backtrack: Literal[False],
718
+ ) -> np.ndarray:
719
+ ...
720
+
721
+
722
+ @overload
723
+ def rqa(
724
+ sim: np.ndarray,
725
+ *,
726
+ gap_onset: float = ...,
727
+ gap_extend: float = ...,
728
+ knight_moves: bool = ...,
729
+ backtrack: Literal[True] = ...,
730
+ ) -> Tuple[np.ndarray, np.ndarray]:
731
+ ...
732
+
733
+
734
+ @overload
735
+ def rqa(
736
+ sim: np.ndarray,
737
+ *,
738
+ gap_onset: float = ...,
739
+ gap_extend: float = ...,
740
+ knight_moves: bool = ...,
741
+ backtrack: bool = ...,
742
+ ) -> Union[np.ndarray, Tuple[np.ndarray, np.ndarray]]:
743
+ ...
744
+
745
+
746
+ def rqa(
747
+ sim: np.ndarray,
748
+ *,
749
+ gap_onset: float = 1,
750
+ gap_extend: float = 1,
751
+ knight_moves: bool = True,
752
+ backtrack: bool = True,
753
+ ) -> Union[np.ndarray, Tuple[np.ndarray, np.ndarray]]:
754
+ """Recurrence quantification analysis (RQA)
755
+
756
+ This function implements different forms of RQA as described by
757
+ Serra, Serra, and Andrzejak (SSA). [#]_ These methods take as input
758
+ a self- or cross-similarity matrix ``sim``, and calculate the value
759
+ of path alignments by dynamic programming.
760
+
761
+ Note that unlike dynamic time warping (`dtw`), alignment paths here are
762
+ maximized, not minimized, so the input should measure similarity rather
763
+ than distance.
764
+
765
+ The simplest RQA method, denoted as `L` (SSA equation 3) and equivalent
766
+ to the method described by Eckman, Kamphorst, and Ruelle [#]_, accumulates
767
+ the length of diagonal paths with positive values in the input:
768
+
769
+ - ``score[i, j] = score[i-1, j-1] + 1`` if ``sim[i, j] > 0``
770
+ - ``score[i, j] = 0`` otherwise.
771
+
772
+ The second method, denoted as `S` (SSA equation 4), is similar to the first,
773
+ but allows for "knight moves" (as in the chess piece) in addition to strict
774
+ diagonal moves:
775
+
776
+ - ``score[i, j] = max(score[i-1, j-1], score[i-2, j-1], score[i-1, j-2]) + 1`` if ``sim[i, j] >
777
+ 0``
778
+ - ``score[i, j] = 0`` otherwise.
779
+
780
+ The third method, denoted as `Q` (SSA equations 5 and 6) extends this by
781
+ allowing gaps in the alignment that incur some cost, rather than a hard
782
+ reset to 0 whenever ``sim[i, j] == 0``.
783
+ Gaps are penalized by two additional parameters, ``gap_onset`` and ``gap_extend``,
784
+ which are subtracted from the value of the alignment path every time a gap
785
+ is introduced or extended (respectively).
786
+
787
+ Note that setting ``gap_onset`` and ``gap_extend`` to `np.inf` recovers the second
788
+ method, and disabling knight moves recovers the first.
789
+
790
+ .. [#] Serrà, Joan, Xavier Serra, and Ralph G. Andrzejak.
791
+ "Cross recurrence quantification for cover song identification."
792
+ New Journal of Physics 11, no. 9 (2009): 093017.
793
+
794
+ .. [#] Eckmann, J. P., S. Oliffson Kamphorst, and D. Ruelle.
795
+ "Recurrence plots of dynamical systems."
796
+ World Scientific Series on Nonlinear Science Series A 16 (1995): 441-446.
797
+
798
+ Parameters
799
+ ----------
800
+ sim : np.ndarray [shape=(N, M), non-negative]
801
+ The similarity matrix to use as input.
802
+
803
+ This can either be a recurrence matrix (self-similarity)
804
+ or a cross-similarity matrix between two sequences.
805
+
806
+ gap_onset : float > 0
807
+ Penalty for introducing a gap to an alignment sequence
808
+
809
+ gap_extend : float > 0
810
+ Penalty for extending a gap in an alignment sequence
811
+
812
+ knight_moves : bool
813
+ If ``True`` (default), allow for "knight moves" in the alignment,
814
+ e.g., ``(n, m) => (n + 1, m + 2)`` or ``(n + 2, m + 1)``.
815
+
816
+ If ``False``, only allow for diagonal moves ``(n, m) => (n + 1, m + 1)``.
817
+
818
+ backtrack : bool
819
+ If ``True``, return the alignment path.
820
+
821
+ If ``False``, only return the score matrix.
822
+
823
+ Returns
824
+ -------
825
+ score : np.ndarray [shape=(N, M)]
826
+ The alignment score matrix. ``score[n, m]`` is the cumulative value of
827
+ the best alignment sequence ending in frames ``n`` and ``m``.
828
+ path : np.ndarray [shape=(k, 2)] (optional)
829
+ If ``backtrack=True``, ``path`` contains a list of pairs of aligned frames
830
+ in the best alignment sequence.
831
+
832
+ ``path[i] = [n, m]`` indicates that row ``n`` aligns to column ``m``.
833
+
834
+ See Also
835
+ --------
836
+ librosa.segment.recurrence_matrix
837
+ librosa.segment.cross_similarity
838
+ dtw
839
+
840
+ Examples
841
+ --------
842
+ Simple diagonal path enhancement (L-mode)
843
+
844
+ >>> import numpy as np
845
+ >>> import matplotlib.pyplot as plt
846
+ >>> y, sr = librosa.load(librosa.ex('nutcracker'), duration=30)
847
+ >>> chroma = librosa.feature.chroma_cqt(y=y, sr=sr)
848
+ >>> # Use time-delay embedding to reduce noise
849
+ >>> chroma_stack = librosa.feature.stack_memory(chroma, n_steps=10, delay=3)
850
+ >>> # Build recurrence, suppress self-loops within 1 second
851
+ >>> rec = librosa.segment.recurrence_matrix(chroma_stack, width=43,
852
+ ... mode='affinity',
853
+ ... metric='cosine')
854
+ >>> # using infinite cost for gaps enforces strict path continuation
855
+ >>> L_score, L_path = librosa.sequence.rqa(rec,
856
+ ... gap_onset=np.inf,
857
+ ... gap_extend=np.inf,
858
+ ... knight_moves=False)
859
+ >>> fig, ax = plt.subplots(ncols=2)
860
+ >>> librosa.display.specshow(rec, x_axis='frames', y_axis='frames', ax=ax[0])
861
+ >>> ax[0].set(title='Recurrence matrix')
862
+ >>> librosa.display.specshow(L_score, x_axis='frames', y_axis='frames', ax=ax[1])
863
+ >>> ax[1].set(title='Alignment score matrix')
864
+ >>> ax[1].plot(L_path[:, 1], L_path[:, 0], label='Optimal path', color='c')
865
+ >>> ax[1].legend()
866
+ >>> ax[1].label_outer()
867
+
868
+ Full alignment using gaps and knight moves
869
+
870
+ >>> # New gaps cost 5, extending old gaps cost 10 for each step
871
+ >>> score, path = librosa.sequence.rqa(rec, gap_onset=5, gap_extend=10)
872
+ >>> fig, ax = plt.subplots(ncols=2, sharex=True, sharey=True)
873
+ >>> librosa.display.specshow(rec, x_axis='frames', y_axis='frames', ax=ax[0])
874
+ >>> ax[0].set(title='Recurrence matrix')
875
+ >>> librosa.display.specshow(score, x_axis='frames', y_axis='frames', ax=ax[1])
876
+ >>> ax[1].set(title='Alignment score matrix')
877
+ >>> ax[1].plot(path[:, 1], path[:, 0], label='Optimal path', color='c')
878
+ >>> ax[1].legend()
879
+ >>> ax[1].label_outer()
880
+ """
881
+ if gap_onset < 0:
882
+ raise ParameterError("gap_onset={} must be strictly positive")
883
+ if gap_extend < 0:
884
+ raise ParameterError("gap_extend={} must be strictly positive")
885
+
886
+ score: np.ndarray
887
+ pointers: np.ndarray
888
+ score, pointers = __rqa_dp(sim, gap_onset, gap_extend, knight_moves)
889
+ if backtrack:
890
+ path = __rqa_backtrack(score, pointers)
891
+ return score, path
892
+
893
+ return score
894
+
895
+
896
+ @jit(nopython=True, cache=True) # type: ignore
897
+ def __rqa_dp(
898
+ sim: np.ndarray, gap_onset: float, gap_extend: float, knight: bool
899
+ ) -> Tuple[np.ndarray, np.ndarray]: # pragma: no cover
900
+ """RQA dynamic programming implementation"""
901
+ # The output array
902
+ score = np.zeros(sim.shape, dtype=sim.dtype)
903
+
904
+ # The backtracking array
905
+ backtrack = np.zeros(sim.shape, dtype=np.int8)
906
+
907
+ # These are place-holder arrays to limit the points being considered
908
+ # at each step of the DP
909
+ #
910
+ # If knight moves are enabled, values are indexed according to
911
+ # [(-1,-1), (-1, -2), (-2, -1)]
912
+ #
913
+ # If knight moves are disabled, then only the first entry is used.
914
+ #
915
+ # Using placeholder vectors here makes the code a bit cleaner down below.
916
+ sim_values = np.zeros(3)
917
+ score_values = np.zeros(3)
918
+ vec = np.zeros(3)
919
+
920
+ if knight:
921
+ # Initial limit is for the base case: diagonal + one knight
922
+ init_limit = 2
923
+
924
+ # Otherwise, we have 3 positions
925
+ limit = 3
926
+ else:
927
+ init_limit = 1
928
+ limit = 1
929
+
930
+ # backtracking rubric:
931
+ # 0 ==> diagonal move
932
+ # 1 ==> knight move up
933
+ # 2 ==> knight move left
934
+ # -1 ==> reset without inclusion
935
+ # -2 ==> reset with inclusion (ie positive value at init)
936
+
937
+ # Initialize the first row and column with the data
938
+ score[0, :] = sim[0, :]
939
+ score[:, 0] = sim[:, 0]
940
+
941
+ # backtracking initialization: the first row and column are all resets
942
+ # if there's a positive link here, it's an inclusive reset
943
+ for i in range(sim.shape[0]):
944
+ if sim[i, 0]:
945
+ backtrack[i, 0] = -2
946
+ else:
947
+ backtrack[i, 0] = -1
948
+
949
+ for j in range(sim.shape[1]):
950
+ if sim[0, j]:
951
+ backtrack[0, j] = -2
952
+ else:
953
+ backtrack[0, j] = -1
954
+
955
+ # Initialize the 1-1 case using only the diagonal
956
+ if sim[1, 1] > 0:
957
+ score[1, 1] = score[0, 0] + sim[1, 1]
958
+ backtrack[1, 1] = 0
959
+ else:
960
+ link = sim[0, 0] > 0
961
+ score[1, 1] = max(0, score[0, 0] - (link) * gap_onset - (~link) * gap_extend)
962
+ if score[1, 1] > 0:
963
+ backtrack[1, 1] = 0
964
+ else:
965
+ backtrack[1, 1] = -1
966
+
967
+ # Initialize the second row with diagonal and left-knight moves
968
+ i = 1
969
+ for j in range(2, sim.shape[1]):
970
+ score_values[:-1] = (score[i - 1, j - 1], score[i - 1, j - 2])
971
+ sim_values[:-1] = (sim[i - 1, j - 1], sim[i - 1, j - 2])
972
+ t_values = sim_values > 0
973
+ if sim[i, j] > 0:
974
+ backtrack[i, j] = np.argmax(score_values[:init_limit])
975
+ score[i, j] = score_values[backtrack[i, j]] + sim[i, j] # or + 1 for binary
976
+ else:
977
+ vec[:init_limit] = (
978
+ score_values[:init_limit]
979
+ - t_values[:init_limit] * gap_onset
980
+ - (~t_values[:init_limit]) * gap_extend
981
+ )
982
+
983
+ backtrack[i, j] = np.argmax(vec[:init_limit])
984
+ score[i, j] = max(0, vec[backtrack[i, j]])
985
+ # Is it a reset?
986
+ if score[i, j] == 0:
987
+ backtrack[i, j] = -1
988
+
989
+ # Initialize the second column with diagonal and up-knight moves
990
+ j = 1
991
+ for i in range(2, sim.shape[0]):
992
+ score_values[:-1] = (score[i - 1, j - 1], score[i - 2, j - 1])
993
+ sim_values[:-1] = (sim[i - 1, j - 1], sim[i - 2, j - 1])
994
+ t_values = sim_values > 0
995
+ if sim[i, j] > 0:
996
+ backtrack[i, j] = np.argmax(score_values[:init_limit])
997
+ score[i, j] = score_values[backtrack[i, j]] + sim[i, j] # or + 1 for binary
998
+
999
+ else:
1000
+ vec[:init_limit] = (
1001
+ score_values[:init_limit]
1002
+ - t_values[:init_limit] * gap_onset
1003
+ - (~t_values[:init_limit]) * gap_extend
1004
+ )
1005
+
1006
+ backtrack[i, j] = np.argmax(vec[:init_limit])
1007
+ score[i, j] = max(0, vec[backtrack[i, j]])
1008
+ # Is it a reset?
1009
+ if score[i, j] == 0:
1010
+ backtrack[i, j] = -1
1011
+
1012
+ # Now fill in the rest of the table
1013
+ for i in range(2, sim.shape[0]):
1014
+ for j in range(2, sim.shape[1]):
1015
+ score_values[:] = (
1016
+ score[i - 1, j - 1],
1017
+ score[i - 1, j - 2],
1018
+ score[i - 2, j - 1],
1019
+ )
1020
+ sim_values[:] = (sim[i - 1, j - 1], sim[i - 1, j - 2], sim[i - 2, j - 1])
1021
+ t_values = sim_values > 0
1022
+ if sim[i, j] > 0:
1023
+ # if knight is true, it's max of (-1,-1), (-1, -2), (-2, -1)
1024
+ # otherwise, it's just the diagonal move (-1, -1)
1025
+ # for backtracking purposes, if the max is 0 then it's the start of a new sequence
1026
+ # if the max is non-zero, then we extend the existing sequence
1027
+ backtrack[i, j] = np.argmax(score_values[:limit])
1028
+ score[i, j] = (
1029
+ score_values[backtrack[i, j]] + sim[i, j]
1030
+ ) # or + 1 for binary
1031
+
1032
+ else:
1033
+ # if the max of our options is negative, then it's a hard reset
1034
+ # otherwise, it's a skip move
1035
+ vec[:limit] = (
1036
+ score_values[:limit]
1037
+ - t_values[:limit] * gap_onset
1038
+ - (~t_values[:limit]) * gap_extend
1039
+ )
1040
+
1041
+ backtrack[i, j] = np.argmax(vec[:limit])
1042
+ score[i, j] = max(0, vec[backtrack[i, j]])
1043
+ # Is it a reset?
1044
+ if score[i, j] == 0:
1045
+ backtrack[i, j] = -1
1046
+
1047
+ return score, backtrack
1048
+
1049
+
1050
+ def __rqa_backtrack(score, pointers):
1051
+ """RQA path backtracking
1052
+
1053
+ Given the score matrix and backtracking index array,
1054
+ reconstruct the optimal path.
1055
+ """
1056
+ # backtracking rubric:
1057
+ # 0 ==> diagonal move
1058
+ # 1 ==> knight move up
1059
+ # 2 ==> knight move left
1060
+ # -1 ==> reset (sim = 0)
1061
+ # -2 ==> start of sequence (sim > 0)
1062
+
1063
+ # This array maps the backtracking values to the
1064
+ # relative index offsets
1065
+ offsets = [(-1, -1), (-1, -2), (-2, -1)]
1066
+
1067
+ # Find the maximum to end the path
1068
+ idx = list(np.unravel_index(np.argmax(score), score.shape))
1069
+
1070
+ # Construct the path
1071
+ path: List = []
1072
+ while True:
1073
+ bt_index = pointers[tuple(idx)]
1074
+
1075
+ # A -1 indicates a non-inclusive reset
1076
+ # this can only happen when sim[idx] == 0,
1077
+ # and a reset with zero score should not be included
1078
+ # in the path. In this case, we're done.
1079
+ if bt_index == -1:
1080
+ break
1081
+
1082
+ # Other bt_index values are okay for inclusion
1083
+ path.insert(0, idx)
1084
+
1085
+ # -2 indicates beginning of sequence,
1086
+ # so we can't backtrack any further
1087
+ if bt_index == -2:
1088
+ break
1089
+
1090
+ # Otherwise, prepend this index and continue
1091
+ idx = [idx[_] + offsets[bt_index][_] for _ in range(len(idx))]
1092
+
1093
+ # If there's no alignment path at all, eg an empty cross-similarity
1094
+ # matrix, return a properly shaped and typed array
1095
+ if not path:
1096
+ return np.empty((0, 2), dtype=np.uint)
1097
+
1098
+ return np.asarray(path, dtype=np.uint)
1099
+
1100
+
1101
+ @jit(nopython=True, cache=True) # type: ignore
1102
+ def _viterbi(
1103
+ log_prob: np.ndarray, log_trans: np.ndarray, log_p_init: np.ndarray
1104
+ ) -> Tuple[np.ndarray, np.ndarray]: # pragma: no cover
1105
+ """Core Viterbi algorithm.
1106
+
1107
+ This is intended for internal use only.
1108
+
1109
+ Parameters
1110
+ ----------
1111
+ log_prob : np.ndarray [shape=(T, m)]
1112
+ ``log_prob[t, s]`` is the conditional log-likelihood
1113
+ ``log P[X = X(t) | State(t) = s]``
1114
+ log_trans : np.ndarray [shape=(m, m)]
1115
+ The log transition matrix
1116
+ ``log_trans[i, j] = log P[State(t+1) = j | State(t) = i]``
1117
+ log_p_init : np.ndarray [shape=(m,)]
1118
+ log of the initial state distribution
1119
+
1120
+ Returns
1121
+ -------
1122
+ None
1123
+ All computations are performed in-place on ``state, value, ptr``.
1124
+ """
1125
+ n_steps, n_states = log_prob.shape
1126
+
1127
+ state = np.zeros(n_steps, dtype=np.uint16)
1128
+ value = np.zeros((n_steps, n_states), dtype=np.float64)
1129
+ ptr = np.zeros((n_steps, n_states), dtype=np.uint16)
1130
+
1131
+ # factor in initial state distribution
1132
+ value[0] = log_prob[0] + log_p_init
1133
+
1134
+ for t in range(1, n_steps):
1135
+ # Want V[t, j] <- p[t, j] * max_k V[t-1, k] * A[k, j]
1136
+ # assume at time t-1 we were in state k
1137
+ # transition k -> j
1138
+
1139
+ # Broadcast over rows:
1140
+ # Tout[k, j] = V[t-1, k] * A[k, j]
1141
+ # then take the max over columns
1142
+ # We'll do this in log-space for stability
1143
+
1144
+ trans_out = value[t - 1] + log_trans.T
1145
+
1146
+ # Unroll the max/argmax loop to enable numba support
1147
+ for j in range(n_states):
1148
+ ptr[t, j] = np.argmax(trans_out[j])
1149
+ # value[t, j] = log_prob[t, j] + np.max(trans_out[j])
1150
+ value[t, j] = log_prob[t, j] + trans_out[j, ptr[t][j]]
1151
+
1152
+ # Now roll backward
1153
+
1154
+ # Get the last state
1155
+ state[-1] = np.argmax(value[-1])
1156
+
1157
+ for t in range(n_steps - 2, -1, -1):
1158
+ state[t] = ptr[t + 1, state[t + 1]]
1159
+
1160
+ logp = value[-1:, state[-1]]
1161
+
1162
+ return state, logp
1163
+
1164
+
1165
+ @overload
1166
+ def viterbi(
1167
+ prob: np.ndarray,
1168
+ transition: np.ndarray,
1169
+ *,
1170
+ p_init: Optional[np.ndarray] = ...,
1171
+ return_logp: Literal[True],
1172
+ ) -> Tuple[np.ndarray, np.ndarray]:
1173
+ ...
1174
+
1175
+
1176
+ @overload
1177
+ def viterbi(
1178
+ prob: np.ndarray,
1179
+ transition: np.ndarray,
1180
+ *,
1181
+ p_init: Optional[np.ndarray] = ...,
1182
+ return_logp: Literal[False] = ...,
1183
+ ) -> np.ndarray:
1184
+ ...
1185
+
1186
+
1187
+ def viterbi(
1188
+ prob: np.ndarray,
1189
+ transition: np.ndarray,
1190
+ *,
1191
+ p_init: Optional[np.ndarray] = None,
1192
+ return_logp: bool = False,
1193
+ ) -> Union[np.ndarray, Tuple[np.ndarray, np.ndarray]]:
1194
+ """Viterbi decoding from observation likelihoods.
1195
+
1196
+ Given a sequence of observation likelihoods ``prob[s, t]``,
1197
+ indicating the conditional likelihood of seeing the observation
1198
+ at time ``t`` from state ``s``, and a transition matrix
1199
+ ``transition[i, j]`` which encodes the conditional probability of
1200
+ moving from state ``i`` to state ``j``, the Viterbi algorithm [#]_ computes
1201
+ the most likely sequence of states from the observations.
1202
+
1203
+ .. [#] Viterbi, Andrew. "Error bounds for convolutional codes and an
1204
+ asymptotically optimum decoding algorithm."
1205
+ IEEE transactions on Information Theory 13.2 (1967): 260-269.
1206
+
1207
+ Parameters
1208
+ ----------
1209
+ prob : np.ndarray [shape=(..., n_states, n_steps), non-negative]
1210
+ ``prob[..., s, t]`` is the probability of observation at time ``t``
1211
+ being generated by state ``s``.
1212
+ transition : np.ndarray [shape=(n_states, n_states), non-negative]
1213
+ ``transition[i, j]`` is the probability of a transition from i->j.
1214
+ Each row must sum to 1.
1215
+ p_init : np.ndarray [shape=(n_states,)]
1216
+ Optional: initial state distribution.
1217
+ If not provided, a uniform distribution is assumed.
1218
+ return_logp : bool
1219
+ If ``True``, return the log-likelihood of the state sequence.
1220
+
1221
+ Returns
1222
+ -------
1223
+ Either ``states`` or ``(states, logp)``:
1224
+ states : np.ndarray [shape=(..., n_steps,)]
1225
+ The most likely state sequence.
1226
+ If ``prob`` contains multiple channels of input, then each channel is
1227
+ decoded independently.
1228
+ logp : scalar [float] or np.ndarray
1229
+ If ``return_logp=True``, the log probability of ``states`` given
1230
+ the observations.
1231
+
1232
+ See Also
1233
+ --------
1234
+ viterbi_discriminative : Viterbi decoding from state likelihoods
1235
+
1236
+ Examples
1237
+ --------
1238
+ Example from https://en.wikipedia.org/wiki/Viterbi_algorithm#Example
1239
+
1240
+ In this example, we have two states ``healthy`` and ``fever``, with
1241
+ initial probabilities 60% and 40%.
1242
+
1243
+ We have three observation possibilities: ``normal``, ``cold``, and
1244
+ ``dizzy``, whose probabilities given each state are:
1245
+
1246
+ ``healthy => {normal: 50%, cold: 40%, dizzy: 10%}`` and
1247
+ ``fever => {normal: 10%, cold: 30%, dizzy: 60%}``
1248
+
1249
+ Finally, we have transition probabilities:
1250
+
1251
+ ``healthy => healthy (70%)`` and
1252
+ ``fever => fever (60%)``.
1253
+
1254
+ Over three days, we observe the sequence ``[normal, cold, dizzy]``,
1255
+ and wish to know the maximum likelihood assignment of states for the
1256
+ corresponding days, which we compute with the Viterbi algorithm below.
1257
+
1258
+ >>> p_init = np.array([0.6, 0.4])
1259
+ >>> p_emit = np.array([[0.5, 0.4, 0.1],
1260
+ ... [0.1, 0.3, 0.6]])
1261
+ >>> p_trans = np.array([[0.7, 0.3], [0.4, 0.6]])
1262
+ >>> path, logp = librosa.sequence.viterbi(p_emit, p_trans, p_init=p_init,
1263
+ ... return_logp=True)
1264
+ >>> print(logp, path)
1265
+ -4.19173690823075 [0 0 1]
1266
+ """
1267
+ n_states, n_steps = prob.shape[-2:]
1268
+
1269
+ if transition.shape != (n_states, n_states):
1270
+ raise ParameterError(
1271
+ f"transition.shape={transition.shape}, must be "
1272
+ f"(n_states, n_states)={n_states, n_states}"
1273
+ )
1274
+
1275
+ if np.any(transition < 0) or not np.allclose(transition.sum(axis=1), 1):
1276
+ raise ParameterError(
1277
+ "Invalid transition matrix: must be non-negative "
1278
+ "and sum to 1 on each row."
1279
+ )
1280
+
1281
+ if np.any(prob < 0) or np.any(prob > 1):
1282
+ raise ParameterError("Invalid probability values: must be between 0 and 1.")
1283
+
1284
+ # Compute log-likelihoods while avoiding log-underflow
1285
+ epsilon = tiny(prob)
1286
+
1287
+ if p_init is None:
1288
+ p_init = np.empty(n_states)
1289
+ p_init.fill(1.0 / n_states)
1290
+ elif (
1291
+ np.any(p_init < 0)
1292
+ or not np.allclose(p_init.sum(), 1)
1293
+ or p_init.shape != (n_states,)
1294
+ ):
1295
+ raise ParameterError(f"Invalid initial state distribution: p_init={p_init}")
1296
+
1297
+ log_trans = np.log(transition + epsilon)
1298
+ log_prob = np.log(prob + epsilon)
1299
+ log_p_init = np.log(p_init + epsilon)
1300
+
1301
+ def _helper(lp):
1302
+ # Transpose input
1303
+ _state, logp = _viterbi(lp.T, log_trans, log_p_init)
1304
+ # Transpose outputs for return
1305
+ return _state.T, logp
1306
+
1307
+ states: np.ndarray
1308
+ logp: np.ndarray
1309
+
1310
+ if log_prob.ndim == 2:
1311
+ states, logp = _helper(log_prob)
1312
+ else:
1313
+ # Vectorize the helper
1314
+ __viterbi = np.vectorize(
1315
+ _helper, otypes=[np.uint16, np.float64], signature="(s,t)->(t),(1)"
1316
+ )
1317
+
1318
+ states, logp = __viterbi(log_prob)
1319
+
1320
+ # Flatten out the trailing dimension introduced by vectorization
1321
+ logp = logp[..., 0]
1322
+
1323
+ if return_logp:
1324
+ return states, logp
1325
+
1326
+ return states
1327
+
1328
+
1329
+ @overload
1330
+ def viterbi_discriminative(
1331
+ prob: np.ndarray,
1332
+ transition: np.ndarray,
1333
+ *,
1334
+ p_state: Optional[np.ndarray] = ...,
1335
+ p_init: Optional[np.ndarray] = ...,
1336
+ return_logp: Literal[False] = ...,
1337
+ ) -> np.ndarray:
1338
+ ...
1339
+
1340
+
1341
+ @overload
1342
+ def viterbi_discriminative(
1343
+ prob: np.ndarray,
1344
+ transition: np.ndarray,
1345
+ *,
1346
+ p_state: Optional[np.ndarray] = ...,
1347
+ p_init: Optional[np.ndarray] = ...,
1348
+ return_logp: Literal[True],
1349
+ ) -> Tuple[np.ndarray, np.ndarray]:
1350
+ ...
1351
+
1352
+
1353
+ @overload
1354
+ def viterbi_discriminative(
1355
+ prob: np.ndarray,
1356
+ transition: np.ndarray,
1357
+ *,
1358
+ p_state: Optional[np.ndarray] = ...,
1359
+ p_init: Optional[np.ndarray] = ...,
1360
+ return_logp: bool,
1361
+ ) -> Union[np.ndarray, Tuple[np.ndarray, np.ndarray]]:
1362
+ ...
1363
+
1364
+
1365
+ def viterbi_discriminative(
1366
+ prob: np.ndarray,
1367
+ transition: np.ndarray,
1368
+ *,
1369
+ p_state: Optional[np.ndarray] = None,
1370
+ p_init: Optional[np.ndarray] = None,
1371
+ return_logp: bool = False,
1372
+ ) -> Union[np.ndarray, Tuple[np.ndarray, np.ndarray]]:
1373
+ """Viterbi decoding from discriminative state predictions.
1374
+
1375
+ Given a sequence of conditional state predictions ``prob[s, t]``,
1376
+ indicating the conditional likelihood of state ``s`` given the
1377
+ observation at time ``t``, and a transition matrix ``transition[i, j]``
1378
+ which encodes the conditional probability of moving from state ``i``
1379
+ to state ``j``, the Viterbi algorithm computes the most likely sequence
1380
+ of states from the observations.
1381
+
1382
+ This implementation uses the standard Viterbi decoding algorithm
1383
+ for observation likelihood sequences, under the assumption that
1384
+ ``P[Obs(t) | State(t) = s]`` is proportional to
1385
+ ``P[State(t) = s | Obs(t)] / P[State(t) = s]``, where the denominator
1386
+ is the marginal probability of state ``s`` occurring as given by ``p_state``.
1387
+
1388
+ Note that because the denominator ``P[State(t) = s]`` is not explicitly
1389
+ calculated, the resulting probabilities (or log-probabilities) are not
1390
+ normalized. If using the `return_logp=True` option (see below),
1391
+ be aware that the "probabilities" may not sum to (and may exceed) 1.
1392
+
1393
+ Parameters
1394
+ ----------
1395
+ prob : np.ndarray [shape=(..., n_states, n_steps), non-negative]
1396
+ ``prob[s, t]`` is the probability of state ``s`` conditional on
1397
+ the observation at time ``t``.
1398
+ Must be non-negative and sum to 1 along each column.
1399
+ transition : np.ndarray [shape=(n_states, n_states), non-negative]
1400
+ ``transition[i, j]`` is the probability of a transition from i->j.
1401
+ Each row must sum to 1.
1402
+ p_state : np.ndarray [shape=(n_states,)]
1403
+ Optional: marginal probability distribution over states,
1404
+ must be non-negative and sum to 1.
1405
+ If not provided, a uniform distribution is assumed.
1406
+ p_init : np.ndarray [shape=(n_states,)]
1407
+ Optional: initial state distribution.
1408
+ If not provided, it is assumed to be uniform.
1409
+ return_logp : bool
1410
+ If ``True``, return the log-likelihood of the state sequence.
1411
+
1412
+ Returns
1413
+ -------
1414
+ Either ``states`` or ``(states, logp)``:
1415
+ states : np.ndarray [shape=(..., n_steps,)]
1416
+ The most likely state sequence.
1417
+ If ``prob`` contains multiple input channels,
1418
+ then each channel is decoded independently.
1419
+ logp : scalar [float] or np.ndarray
1420
+ If ``return_logp=True``, the (unnormalized) log probability
1421
+ of ``states`` given the observations.
1422
+
1423
+ See Also
1424
+ --------
1425
+ viterbi :
1426
+ Viterbi decoding from observation likelihoods
1427
+ viterbi_binary :
1428
+ Viterbi decoding for multi-label, conditional state likelihoods
1429
+
1430
+ Examples
1431
+ --------
1432
+ This example constructs a simple, template-based discriminative chord estimator,
1433
+ using CENS chroma as input features.
1434
+
1435
+ .. note:: this chord model is not accurate enough to use in practice. It is only
1436
+ intended to demonstrate how to use discriminative Viterbi decoding.
1437
+
1438
+ >>> # Create templates for major, minor, and no-chord qualities
1439
+ >>> maj_template = np.array([1,0,0, 0,1,0, 0,1,0, 0,0,0])
1440
+ >>> min_template = np.array([1,0,0, 1,0,0, 0,1,0, 0,0,0])
1441
+ >>> N_template = np.array([1,1,1, 1,1,1, 1,1,1, 1,1,1.]) / 4.
1442
+ >>> # Generate the weighting matrix that maps chroma to labels
1443
+ >>> weights = np.zeros((25, 12), dtype=float)
1444
+ >>> labels = ['C:maj', 'C#:maj', 'D:maj', 'D#:maj', 'E:maj', 'F:maj',
1445
+ ... 'F#:maj', 'G:maj', 'G#:maj', 'A:maj', 'A#:maj', 'B:maj',
1446
+ ... 'C:min', 'C#:min', 'D:min', 'D#:min', 'E:min', 'F:min',
1447
+ ... 'F#:min', 'G:min', 'G#:min', 'A:min', 'A#:min', 'B:min',
1448
+ ... 'N']
1449
+ >>> for c in range(12):
1450
+ ... weights[c, :] = np.roll(maj_template, c) # c:maj
1451
+ ... weights[c + 12, :] = np.roll(min_template, c) # c:min
1452
+ >>> weights[-1] = N_template # the last row is the no-chord class
1453
+ >>> # Make a self-loop transition matrix over 25 states
1454
+ >>> trans = librosa.sequence.transition_loop(25, 0.9)
1455
+
1456
+ >>> # Load in audio and make features
1457
+ >>> y, sr = librosa.load(librosa.ex('nutcracker'), duration=15)
1458
+ >>> # Suppress percussive elements
1459
+ >>> y = librosa.effects.harmonic(y, margin=4)
1460
+ >>> chroma = librosa.feature.chroma_cqt(y=y, sr=sr)
1461
+ >>> # Map chroma (observations) to class (state) likelihoods
1462
+ >>> probs = np.exp(weights.dot(chroma)) # P[class | chroma] ~= exp(template' chroma)
1463
+ >>> probs /= probs.sum(axis=0, keepdims=True) # probabilities must sum to 1 in each column
1464
+ >>> # Compute independent frame-wise estimates
1465
+ >>> chords_ind = np.argmax(probs, axis=0)
1466
+ >>> # And viterbi estimates
1467
+ >>> chords_vit = librosa.sequence.viterbi_discriminative(probs, trans)
1468
+
1469
+ >>> # Plot the features and prediction map
1470
+ >>> import matplotlib.pyplot as plt
1471
+ >>> fig, ax = plt.subplots(nrows=2)
1472
+ >>> librosa.display.specshow(chroma, x_axis='time', y_axis='chroma', ax=ax[0])
1473
+ >>> librosa.display.specshow(weights, x_axis='chroma', ax=ax[1])
1474
+ >>> ax[1].set(yticks=np.arange(25) + 0.5, yticklabels=labels, ylabel='Chord')
1475
+
1476
+ >>> # And plot the results
1477
+ >>> fig, ax = plt.subplots()
1478
+ >>> librosa.display.specshow(probs, x_axis='time', cmap='gray', ax=ax)
1479
+ >>> times = librosa.times_like(chords_vit)
1480
+ >>> ax.scatter(times, chords_ind + 0.25, color='lime', alpha=0.5, marker='+',
1481
+ ... s=15, label='Independent')
1482
+ >>> ax.scatter(times, chords_vit - 0.25, color='deeppink', alpha=0.5, marker='o',
1483
+ ... s=15, label='Viterbi')
1484
+ >>> ax.set(yticks=np.unique(chords_vit),
1485
+ ... yticklabels=[labels[i] for i in np.unique(chords_vit)])
1486
+ >>> ax.legend()
1487
+ """
1488
+ n_states, n_steps = prob.shape[-2:]
1489
+
1490
+ if transition.shape != (n_states, n_states):
1491
+ raise ParameterError(
1492
+ f"transition.shape={transition.shape}, must be "
1493
+ f"(n_states, n_states)={n_states, n_states}"
1494
+ )
1495
+
1496
+ if np.any(transition < 0) or not np.allclose(transition.sum(axis=1), 1):
1497
+ raise ParameterError(
1498
+ "Invalid transition matrix: must be non-negative "
1499
+ "and sum to 1 on each row."
1500
+ )
1501
+
1502
+ if np.any(prob < 0) or not np.allclose(prob.sum(axis=-2), 1):
1503
+ raise ParameterError(
1504
+ "Invalid probability values: each column must "
1505
+ "sum to 1 and be non-negative"
1506
+ )
1507
+
1508
+ # Compute log-likelihoods while avoiding log-underflow
1509
+ epsilon = tiny(prob)
1510
+
1511
+ # Compute marginal log probabilities while avoiding underflow
1512
+ if p_state is None:
1513
+ p_state = np.empty(n_states)
1514
+ p_state.fill(1.0 / n_states)
1515
+ elif p_state.shape != (n_states,):
1516
+ raise ParameterError(
1517
+ "Marginal distribution p_state must have shape (n_states,). "
1518
+ f"Got p_state.shape={p_state.shape}"
1519
+ )
1520
+ elif np.any(p_state < 0) or not np.allclose(p_state.sum(axis=-1), 1):
1521
+ raise ParameterError(f"Invalid marginal state distribution: p_state={p_state}")
1522
+
1523
+ if p_init is None:
1524
+ p_init = np.empty(n_states)
1525
+ p_init.fill(1.0 / n_states)
1526
+ elif (
1527
+ np.any(p_init < 0)
1528
+ or not np.allclose(p_init.sum(), 1)
1529
+ or p_init.shape != (n_states,)
1530
+ ):
1531
+ raise ParameterError(f"Invalid initial state distribution: p_init={p_init}")
1532
+
1533
+ # By Bayes' rule, P[X | Y] * P[Y] = P[Y | X] * P[X]
1534
+ # P[X] is constant for the sake of maximum likelihood inference
1535
+ # and P[Y] is given by the marginal distribution p_state.
1536
+ #
1537
+ # So we have P[X | y] \propto P[Y | x] / P[Y]
1538
+ # if X = observation and Y = states, this can be done in log space as
1539
+ # log P[X | y] \propto \log P[Y | x] - \log P[Y]
1540
+ log_p_init = np.log(p_init + epsilon)
1541
+ log_trans = np.log(transition + epsilon)
1542
+ log_marginal = np.log(p_state + epsilon)
1543
+
1544
+ # reshape to broadcast against prob
1545
+ log_marginal = expand_to(log_marginal, ndim=prob.ndim, axes=-2)
1546
+
1547
+ log_prob = np.log(prob + epsilon) - log_marginal
1548
+
1549
+ def _helper(lp):
1550
+ # Transpose input
1551
+ _state, logp = _viterbi(lp.T, log_trans, log_p_init)
1552
+ # Transpose outputs for return
1553
+ return _state.T, logp
1554
+
1555
+ states: np.ndarray
1556
+ logp: np.ndarray
1557
+ if log_prob.ndim == 2:
1558
+ states, logp = _helper(log_prob)
1559
+ else:
1560
+ # Vectorize the helper
1561
+ __viterbi = np.vectorize(
1562
+ _helper, otypes=[np.uint16, np.float64], signature="(s,t)->(t),(1)"
1563
+ )
1564
+
1565
+ states, logp = __viterbi(log_prob)
1566
+
1567
+ # Flatten out the trailing dimension
1568
+ logp = logp[..., 0]
1569
+
1570
+ if return_logp:
1571
+ return states, logp
1572
+
1573
+ return states
1574
+
1575
+
1576
+ @overload
1577
+ def viterbi_binary(
1578
+ prob: np.ndarray,
1579
+ transition: np.ndarray,
1580
+ *,
1581
+ p_state: Optional[np.ndarray] = ...,
1582
+ p_init: Optional[np.ndarray] = ...,
1583
+ return_logp: Literal[False] = ...,
1584
+ ) -> np.ndarray:
1585
+ ...
1586
+
1587
+
1588
+ @overload
1589
+ def viterbi_binary(
1590
+ prob: np.ndarray,
1591
+ transition: np.ndarray,
1592
+ *,
1593
+ p_state: Optional[np.ndarray] = ...,
1594
+ p_init: Optional[np.ndarray] = ...,
1595
+ return_logp: Literal[True],
1596
+ ) -> Tuple[np.ndarray, np.ndarray]:
1597
+ ...
1598
+
1599
+
1600
+ @overload
1601
+ def viterbi_binary(
1602
+ prob: np.ndarray,
1603
+ transition: np.ndarray,
1604
+ *,
1605
+ p_state: Optional[np.ndarray] = ...,
1606
+ p_init: Optional[np.ndarray] = ...,
1607
+ return_logp: bool = ...,
1608
+ ) -> Union[np.ndarray, Tuple[np.ndarray, np.ndarray]]:
1609
+ ...
1610
+
1611
+
1612
+ def viterbi_binary(
1613
+ prob: np.ndarray,
1614
+ transition: np.ndarray,
1615
+ *,
1616
+ p_state: Optional[np.ndarray] = None,
1617
+ p_init: Optional[np.ndarray] = None,
1618
+ return_logp: bool = False,
1619
+ ) -> Union[np.ndarray, Tuple[np.ndarray, np.ndarray]]:
1620
+ """Viterbi decoding from binary (multi-label), discriminative state predictions.
1621
+
1622
+ Given a sequence of conditional state predictions ``prob[s, t]``,
1623
+ indicating the conditional likelihood of state ``s`` being active
1624
+ conditional on observation at time ``t``, and a 2*2 transition matrix
1625
+ ``transition`` which encodes the conditional probability of moving from
1626
+ state ``s`` to state ``~s`` (not-``s``), the Viterbi algorithm computes the
1627
+ most likely sequence of states from the observations.
1628
+
1629
+ This function differs from `viterbi_discriminative` in that it does not assume the
1630
+ states to be mutually exclusive. `viterbi_binary` is implemented by
1631
+ transforming the multi-label decoding problem to a collection
1632
+ of binary Viterbi problems (one for each *state* or label).
1633
+
1634
+ The output is a binary matrix ``states[s, t]`` indicating whether each
1635
+ state ``s`` is active at time ``t``.
1636
+
1637
+ Like `viterbi_discriminative`, the probabilities of the optimal state sequences
1638
+ are not normalized here. If using the `return_logp=True` option (see below),
1639
+ be aware that the "probabilities" may not sum to (and may exceed) 1.
1640
+
1641
+ Parameters
1642
+ ----------
1643
+ prob : np.ndarray [shape=(..., n_steps,) or (..., n_states, n_steps)], non-negative
1644
+ ``prob[s, t]`` is the probability of state ``s`` being active
1645
+ conditional on the observation at time ``t``.
1646
+ Must be non-negative and less than 1.
1647
+
1648
+ If ``prob`` is 1-dimensional, it is expanded to shape ``(1, n_steps)``.
1649
+
1650
+ If ``prob`` contains multiple input channels, then each channel is decoded independently.
1651
+
1652
+ transition : np.ndarray [shape=(2, 2) or (n_states, 2, 2)], non-negative
1653
+ If 2-dimensional, the same transition matrix is applied to each sub-problem.
1654
+ ``transition[0, i]`` is the probability of the state going from inactive to ``i``,
1655
+ ``transition[1, i]`` is the probability of the state going from active to ``i``.
1656
+ Each row must sum to 1.
1657
+
1658
+ If 3-dimensional, ``transition[s]`` is interpreted as the 2x2 transition matrix
1659
+ for state label ``s``.
1660
+
1661
+ p_state : np.ndarray [shape=(n_states,)]
1662
+ Optional: marginal probability for each state (between [0,1]).
1663
+ If not provided, a uniform distribution (0.5 for each state)
1664
+ is assumed.
1665
+
1666
+ p_init : np.ndarray [shape=(n_states,)]
1667
+ Optional: initial state distribution.
1668
+ If not provided, it is assumed to be uniform.
1669
+
1670
+ return_logp : bool
1671
+ If ``True``, return the (unnormalized) log-likelihood of the state sequences.
1672
+
1673
+ Returns
1674
+ -------
1675
+ Either ``states`` or ``(states, logp)``:
1676
+ states : np.ndarray [shape=(..., n_states, n_steps)]
1677
+ The most likely state sequence.
1678
+ logp : np.ndarray [shape=(..., n_states,)]
1679
+ If ``return_logp=True``, the (unnormalized) log probability of each
1680
+ state activation sequence ``states``
1681
+
1682
+ See Also
1683
+ --------
1684
+ viterbi :
1685
+ Viterbi decoding from observation likelihoods
1686
+ viterbi_discriminative :
1687
+ Viterbi decoding for discriminative (mutually exclusive) state predictions
1688
+
1689
+ Examples
1690
+ --------
1691
+ In this example, we have a sequence of binary state likelihoods that we want to de-noise
1692
+ under the assumption that state changes are relatively uncommon. Positive predictions
1693
+ should only be retained if they persist for multiple steps, and any transient predictions
1694
+ should be considered as errors. This use case arises frequently in problems such as
1695
+ instrument recognition, where state activations tend to be stable over time, but subject
1696
+ to abrupt changes (e.g., when an instrument joins the mix).
1697
+
1698
+ We assume that the 0 state has a self-transition probability of 90%, and the 1 state
1699
+ has a self-transition probability of 70%. We assume the marginal and initial
1700
+ probability of either state is 50%.
1701
+
1702
+ >>> trans = np.array([[0.9, 0.1], [0.3, 0.7]])
1703
+ >>> prob = np.array([0.1, 0.7, 0.4, 0.3, 0.8, 0.9, 0.8, 0.2, 0.6, 0.3])
1704
+ >>> librosa.sequence.viterbi_binary(prob, trans, p_state=0.5, p_init=0.5)
1705
+ array([[0, 0, 0, 0, 1, 1, 1, 0, 0, 0]])
1706
+ """
1707
+ prob = np.atleast_2d(prob)
1708
+
1709
+ n_states, n_steps = prob.shape[-2:]
1710
+
1711
+ if transition.shape == (2, 2):
1712
+ transition = np.tile(transition, (n_states, 1, 1))
1713
+ elif transition.shape != (n_states, 2, 2):
1714
+ raise ParameterError(
1715
+ f"transition.shape={transition.shape}, must be (2, 2) or "
1716
+ f"(n_states, 2, 2)={n_states}"
1717
+ )
1718
+
1719
+ if np.any(transition < 0) or not np.allclose(transition.sum(axis=-1), 1):
1720
+ raise ParameterError(
1721
+ "Invalid transition matrix: must be non-negative "
1722
+ "and sum to 1 on each row."
1723
+ )
1724
+
1725
+ if np.any(prob < 0) or np.any(prob > 1):
1726
+ raise ParameterError("Invalid probability values: prob must be between [0, 1]")
1727
+
1728
+ if p_state is None:
1729
+ p_state = np.empty(n_states)
1730
+ p_state.fill(0.5)
1731
+ else:
1732
+ p_state = np.atleast_1d(p_state)
1733
+
1734
+ assert p_state is not None
1735
+
1736
+ if p_state.shape != (n_states,) or np.any(p_state < 0) or np.any(p_state > 1):
1737
+ raise ParameterError(f"Invalid marginal state distributions: p_state={p_state}")
1738
+
1739
+ if p_init is None:
1740
+ p_init = np.empty(n_states)
1741
+ p_init.fill(0.5)
1742
+ else:
1743
+ p_init = np.atleast_1d(p_init)
1744
+
1745
+ assert p_init is not None
1746
+
1747
+ if p_init.shape != (n_states,) or np.any(p_init < 0) or np.any(p_init > 1):
1748
+ raise ParameterError(f"Invalid initial state distributions: p_init={p_init}")
1749
+
1750
+ shape_prefix = list(prob.shape[:-2])
1751
+ states = np.empty(shape_prefix + [n_states, n_steps], dtype=np.uint16)
1752
+ logp = np.empty(shape_prefix + [n_states])
1753
+
1754
+ prob_binary = np.empty(shape_prefix + [2, n_steps])
1755
+ p_state_binary = np.empty(2)
1756
+ p_init_binary = np.empty(2)
1757
+
1758
+ for state in range(n_states):
1759
+ prob_binary[..., 0, :] = 1 - prob[..., state, :]
1760
+ prob_binary[..., 1, :] = prob[..., state, :]
1761
+
1762
+ p_state_binary[0] = 1 - p_state[state]
1763
+ p_state_binary[1] = p_state[state]
1764
+
1765
+ p_init_binary[0] = 1 - p_init[state]
1766
+ p_init_binary[1] = p_init[state]
1767
+
1768
+ states[..., state, :], logp[..., state] = viterbi_discriminative(
1769
+ prob_binary,
1770
+ transition[state],
1771
+ p_state=p_state_binary,
1772
+ p_init=p_init_binary,
1773
+ return_logp=True,
1774
+ )
1775
+
1776
+ if return_logp:
1777
+ return states, logp
1778
+
1779
+ return states
1780
+
1781
+
1782
+ def transition_uniform(n_states: int) -> np.ndarray:
1783
+ """Construct a uniform transition matrix over ``n_states``.
1784
+
1785
+ Parameters
1786
+ ----------
1787
+ n_states : int > 0
1788
+ The number of states
1789
+
1790
+ Returns
1791
+ -------
1792
+ transition : np.ndarray [shape=(n_states, n_states)]
1793
+ ``transition[i, j] = 1./n_states``
1794
+
1795
+ Examples
1796
+ --------
1797
+ >>> librosa.sequence.transition_uniform(3)
1798
+ array([[0.333, 0.333, 0.333],
1799
+ [0.333, 0.333, 0.333],
1800
+ [0.333, 0.333, 0.333]])
1801
+ """
1802
+ if not is_positive_int(n_states):
1803
+ raise ParameterError(f"n_states={n_states} must be a positive integer")
1804
+
1805
+ transition = np.empty((n_states, n_states), dtype=np.float64)
1806
+ transition.fill(1.0 / n_states)
1807
+ return transition
1808
+
1809
+
1810
+ def transition_loop(n_states: int, prob: Union[float, Iterable[float]]) -> np.ndarray:
1811
+ """Construct a self-loop transition matrix over ``n_states``.
1812
+
1813
+ The transition matrix will have the following properties:
1814
+
1815
+ - ``transition[i, i] = p`` for all ``i``
1816
+ - ``transition[i, j] = (1 - p) / (n_states - 1)`` for all ``j != i``
1817
+
1818
+ This type of transition matrix is appropriate when states tend to be
1819
+ locally stable, and there is no additional structure between different
1820
+ states. This is primarily useful for de-noising frame-wise predictions.
1821
+
1822
+ Parameters
1823
+ ----------
1824
+ n_states : int > 1
1825
+ The number of states
1826
+
1827
+ prob : float in [0, 1] or iterable, length=n_states
1828
+ If a scalar, this is the probability of a self-transition.
1829
+
1830
+ If a vector of length ``n_states``, ``p[i]`` is the probability of self-transition in state ``i``
1831
+
1832
+ Returns
1833
+ -------
1834
+ transition : np.ndarray [shape=(n_states, n_states)]
1835
+ The transition matrix
1836
+
1837
+ Examples
1838
+ --------
1839
+ >>> librosa.sequence.transition_loop(3, 0.5)
1840
+ array([[0.5 , 0.25, 0.25],
1841
+ [0.25, 0.5 , 0.25],
1842
+ [0.25, 0.25, 0.5 ]])
1843
+
1844
+ >>> librosa.sequence.transition_loop(3, [0.8, 0.5, 0.25])
1845
+ array([[0.8 , 0.1 , 0.1 ],
1846
+ [0.25 , 0.5 , 0.25 ],
1847
+ [0.375, 0.375, 0.25 ]])
1848
+ """
1849
+ if not (is_positive_int(n_states) and (n_states > 1)):
1850
+ raise ParameterError(f"n_states={n_states} must be a positive integer > 1")
1851
+
1852
+ transition = np.empty((n_states, n_states), dtype=np.float64)
1853
+
1854
+ # if it's a float, make it a vector
1855
+ prob = np.asarray(prob, dtype=np.float64)
1856
+
1857
+ if prob.ndim == 0:
1858
+ prob = np.tile(prob, n_states)
1859
+
1860
+ if prob.shape != (n_states,):
1861
+ raise ParameterError(
1862
+ f"prob={prob} must have length equal to n_states={n_states}"
1863
+ )
1864
+
1865
+ if np.any(prob < 0) or np.any(prob > 1):
1866
+ raise ParameterError(f"prob={prob} must have values in the range [0, 1]")
1867
+
1868
+ for i, prob_i in enumerate(prob):
1869
+ transition[i] = (1.0 - prob_i) / (n_states - 1)
1870
+ transition[i, i] = prob_i
1871
+
1872
+ return transition
1873
+
1874
+
1875
+ def transition_cycle(n_states: int, prob: Union[float, Iterable[float]]) -> np.ndarray:
1876
+ """Construct a cyclic transition matrix over ``n_states``.
1877
+
1878
+ The transition matrix will have the following properties:
1879
+
1880
+ - ``transition[i, i] = p``
1881
+ - ``transition[i, i + 1] = (1 - p)``
1882
+
1883
+ This type of transition matrix is appropriate for state spaces
1884
+ with cyclical structure, such as metrical position within a bar.
1885
+ For example, a song in 4/4 time has state transitions of the form
1886
+
1887
+ 1->{1, 2}, 2->{2, 3}, 3->{3, 4}, 4->{4, 1}.
1888
+
1889
+ Parameters
1890
+ ----------
1891
+ n_states : int > 1
1892
+ The number of states
1893
+
1894
+ prob : float in [0, 1] or iterable, length=n_states
1895
+ If a scalar, this is the probability of a self-transition.
1896
+
1897
+ If a vector of length ``n_states``, ``p[i]`` is the probability of
1898
+ self-transition in state ``i``
1899
+
1900
+ Returns
1901
+ -------
1902
+ transition : np.ndarray [shape=(n_states, n_states)]
1903
+ The transition matrix
1904
+
1905
+ Examples
1906
+ --------
1907
+ >>> librosa.sequence.transition_cycle(4, 0.9)
1908
+ array([[0.9, 0.1, 0. , 0. ],
1909
+ [0. , 0.9, 0.1, 0. ],
1910
+ [0. , 0. , 0.9, 0.1],
1911
+ [0.1, 0. , 0. , 0.9]])
1912
+ """
1913
+ if not (is_positive_int(n_states) and n_states > 1):
1914
+ raise ParameterError(f"n_states={n_states} must be a positive integer > 1")
1915
+
1916
+ transition = np.zeros((n_states, n_states), dtype=np.float64)
1917
+
1918
+ # if it's a float, make it a vector
1919
+ prob = np.asarray(prob, dtype=np.float64)
1920
+
1921
+ if prob.ndim == 0:
1922
+ prob = np.tile(prob, n_states)
1923
+
1924
+ if prob.shape != (n_states,):
1925
+ raise ParameterError(
1926
+ f"prob={prob} must have length equal to n_states={n_states}"
1927
+ )
1928
+
1929
+ if np.any(prob < 0) or np.any(prob > 1):
1930
+ raise ParameterError(f"prob={prob} must have values in the range [0, 1]")
1931
+
1932
+ for i, prob_i in enumerate(prob):
1933
+ transition[i, np.mod(i + 1, n_states)] = 1.0 - prob_i
1934
+ transition[i, i] = prob_i
1935
+
1936
+ return transition
1937
+
1938
+
1939
+ def transition_local(
1940
+ n_states: int,
1941
+ width: Union[int, Iterable[int]],
1942
+ *,
1943
+ window: _WindowSpec = "triangle",
1944
+ wrap: bool = False,
1945
+ ) -> np.ndarray:
1946
+ """Construct a localized transition matrix.
1947
+
1948
+ The transition matrix will have the following properties:
1949
+
1950
+ - ``transition[i, j] = 0`` if ``|i - j| > width``
1951
+ - ``transition[i, i]`` is maximal
1952
+ - ``transition[i, i - width//2 : i + width//2]`` has shape ``window``
1953
+
1954
+ This type of transition matrix is appropriate for state spaces
1955
+ that discretely approximate continuous variables, such as in fundamental
1956
+ frequency estimation.
1957
+
1958
+ Parameters
1959
+ ----------
1960
+ n_states : int > 1
1961
+ The number of states
1962
+
1963
+ width : int >= 1 or iterable
1964
+ The maximum number of states to treat as "local".
1965
+ If iterable, it should have length equal to ``n_states``,
1966
+ and specify the width independently for each state.
1967
+
1968
+ window : str, callable, or window specification
1969
+ The window function to determine the shape of the "local" distribution.
1970
+
1971
+ Any window specification supported by `filters.get_window` will work here.
1972
+
1973
+ .. note:: Certain windows (e.g., 'hann') are identically 0 at the boundaries,
1974
+ so and effectively have ``width-2`` non-zero values. You may have to expand
1975
+ ``width`` to get the desired behavior.
1976
+
1977
+ wrap : bool
1978
+ If ``True``, then state locality ``|i - j|`` is computed modulo ``n_states``.
1979
+ If ``False`` (default), then locality is absolute.
1980
+
1981
+ See Also
1982
+ --------
1983
+ librosa.filters.get_window
1984
+
1985
+ Returns
1986
+ -------
1987
+ transition : np.ndarray [shape=(n_states, n_states)]
1988
+ The transition matrix
1989
+
1990
+ Examples
1991
+ --------
1992
+ Triangular distributions with and without wrapping
1993
+
1994
+ >>> librosa.sequence.transition_local(5, 3, window='triangle', wrap=False)
1995
+ array([[0.667, 0.333, 0. , 0. , 0. ],
1996
+ [0.25 , 0.5 , 0.25 , 0. , 0. ],
1997
+ [0. , 0.25 , 0.5 , 0.25 , 0. ],
1998
+ [0. , 0. , 0.25 , 0.5 , 0.25 ],
1999
+ [0. , 0. , 0. , 0.333, 0.667]])
2000
+
2001
+ >>> librosa.sequence.transition_local(5, 3, window='triangle', wrap=True)
2002
+ array([[0.5 , 0.25, 0. , 0. , 0.25],
2003
+ [0.25, 0.5 , 0.25, 0. , 0. ],
2004
+ [0. , 0.25, 0.5 , 0.25, 0. ],
2005
+ [0. , 0. , 0.25, 0.5 , 0.25],
2006
+ [0.25, 0. , 0. , 0.25, 0.5 ]])
2007
+
2008
+ Uniform local distributions with variable widths and no wrapping
2009
+
2010
+ >>> librosa.sequence.transition_local(5, [1, 2, 3, 3, 1], window='ones', wrap=False)
2011
+ array([[1. , 0. , 0. , 0. , 0. ],
2012
+ [0.5 , 0.5 , 0. , 0. , 0. ],
2013
+ [0. , 0.333, 0.333, 0.333, 0. ],
2014
+ [0. , 0. , 0.333, 0.333, 0.333],
2015
+ [0. , 0. , 0. , 0. , 1. ]])
2016
+ """
2017
+ if not (is_positive_int(n_states) and n_states > 1):
2018
+ raise ParameterError(f"n_states={n_states} must be a positive integer > 1")
2019
+
2020
+ width = np.asarray(width, dtype=int)
2021
+ if width.ndim == 0:
2022
+ width = np.tile(width, n_states)
2023
+
2024
+ if width.shape != (n_states,):
2025
+ raise ParameterError(
2026
+ f"width={width} must have length equal to n_states={n_states}"
2027
+ )
2028
+
2029
+ if np.any(width < 1):
2030
+ raise ParameterError(f"width={width} must be at least 1")
2031
+
2032
+ transition = np.zeros((n_states, n_states), dtype=np.float64)
2033
+
2034
+ # Fill in the widths. This is inefficient, but simple
2035
+ for i, width_i in enumerate(width):
2036
+ trans_row = pad_center(
2037
+ get_window(window, width_i, fftbins=False), size=n_states
2038
+ )
2039
+ trans_row = np.roll(trans_row, n_states // 2 + i + 1)
2040
+
2041
+ if not wrap:
2042
+ # Knock out the off-diagonal-band elements
2043
+ trans_row[min(n_states, i + width_i // 2 + 1) :] = 0
2044
+ trans_row[: max(0, i - width_i // 2)] = 0
2045
+
2046
+ transition[i] = trans_row
2047
+
2048
+ # Row-normalize
2049
+ transition /= transition.sum(axis=1, keepdims=True)
2050
+
2051
+ return transition