From a8730c5330d1748f4e60b213a5d2786f4892a057 Mon Sep 17 00:00:00 2001 From: "jiyun.park" Date: Sat, 3 Oct 2026 11:31:41 +0200 Subject: [PATCH 1/9] Audio features: take the modulus of the complex CQT and STFT CQTProcessor, CQTSpectralFluxProcessor (OPHMM) and RawSpectrumProcessor (SKF) passed dtype=np.float32 to librosa.cqt / librosa.stft. librosa 0.11 then returns a real array, dropping the imaginary part, so np.abs gave |Re X| instead of |X| (about 36% mean magnitude error). Compute the complex transform and cast its modulus to float32. OPHMM's score templates already used the complex modulus. --- matchmaker/features/audio.py | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/matchmaker/features/audio.py b/matchmaker/features/audio.py index 704efb7..00ddd1b 100644 --- a/matchmaker/features/audio.py +++ b/matchmaker/features/audio.py @@ -116,11 +116,11 @@ def __call__( sr=self.sample_rate, hop_length=self.hop_length, norm=self.norm, - dtype=np.float32, fmin=librosa.note_to_hz("A0"), n_bins=88, ) - return np.abs(cqt).T[1:-1], f_time + # the modulus of the complex CQT (a real dtype would make librosa drop the imaginary part) + return np.abs(cqt).T.astype(np.float32)[1:-1], f_time class CQTSpectralFluxProcessor(Processor): @@ -162,9 +162,9 @@ def __call__( n_bins=self.n_bins, bins_per_octave=self.bins_per_octave, norm=self.norm, - dtype=np.float32, ) - cqt_features = np.abs(cqt).T + # the modulus of the complex CQT (a real dtype would make librosa drop the imaginary part) + cqt_features = np.abs(cqt).T.astype(np.float32) if self.include_spectral_flux: if self.prev_magnitude is None: @@ -323,9 +323,9 @@ def __call__(self, data: InputAudioFrame): win_length=self.n_fft, hop_length=self.hop_length, center=False, - dtype=np.float32, ) - return np.abs(stft).T, f_time # (n_frames, n_bins) + # the modulus of the complex STFT (a real dtype would make librosa drop the imaginary part) + return np.abs(stft).T.astype(np.float32), f_time # (n_frames, n_bins) def compute_features_from_audio( From 011b476a2887db1a66da1136cd8428a9098b580d Mon Sep 17 00:00:00 2001 From: "jiyun.park" Date: Sat, 3 Oct 2026 11:31:41 +0200 Subject: [PATCH 2/9] outerhmm: neighbourhood transitions and s_j of Nakamura et al. (2016) The audio OPHMM used the symmetric +-3 transition table of the MIDI model (Nakamura et al. 2014, Table 3, with the +1 entry mistyped as 0.94531 for 0.84531), filled the whole matrix with 1e-6 outside the band that the forward step uses, boosted the exit probability by spectral flux, and defaulted to s_j = 1e-5. Follow the audio paper instead (Sec. III-B, IV-B2): nbh(i) = {j; 0 <= i - j <= 2} with a_{i,i+2} = 1e-50 and no backward moves, no mass outside the band, no flux boost, and s_j = 1e-100 (Sec. IV-C). --- matchmaker/prob/outer_product_hmm_audio.py | 37 ++++++++-------------- 1 file changed, 14 insertions(+), 23 deletions(-) diff --git a/matchmaker/prob/outer_product_hmm_audio.py b/matchmaker/prob/outer_product_hmm_audio.py index f325e7c..12fffab 100644 --- a/matchmaker/prob/outer_product_hmm_audio.py +++ b/matchmaker/prob/outer_product_hmm_audio.py @@ -14,25 +14,21 @@ NDArrayFloat = NDArray[np.float32] NDArrayInt = NDArray[np.int32] -# Score-position jump probs (d = i - j) from Nakamura 2014 Table 3, |d| <= 3. -# d=0 (staying on the same note) is omitted: it is zeroed by fill_diagonal and -# instead derived per-chord from note duration and tempo (a00, set in __init__). +# Neighbourhood transitions of Nakamura et al. (2016), Sec. III-B and IV-B2: +# nbh(i) = {j; 0 <= i - j <= 2}, with a_{i,i+2} = 1e-50 for deletion errors and +# a_{i,i} = 0 for insertion errors (staying on a note is the bottom-level a00). DEFAULT_TRANSITIONS = [ - (-3, 0.00509), - (-2, 0.00516), - (-1, 0.00886), - (1, 0.94531), # normal forward progression - (2, 0.00610), # deletion (skip one note) - (3, 0.00073), + (1, 1.0 - 1e-50), # normal forward progression + (2, 1e-50), # deletion (skip one note) ] -DEFAULT_D1 = 3 -DEFAULT_D2 = 3 +# neighbourhood reach in the forward step: D1 states back, D2 states ahead +DEFAULT_D1 = 0 +DEFAULT_D2 = 2 IOI_THRESHOLD = 0.035 # seconds SUSTAINED_DECAY: float = 0.3 # exponential decay rate for sustained notes -_FLUX_EXIT_BOOST: float = 1.0 _OTHER_PROB: float = 1e-6 _EMISSION_BETA: float = 5.0 _UNIFORM_FLOOR: float = 0.2 @@ -209,16 +205,16 @@ def compute_transition_matrix( if transitions is None: transitions = DEFAULT_TRANSITIONS - # Initialize transition matrix with epsilons - alpha = np.full((N, N), 1e-6, dtype=float) + # only the band is used in the forward step; moves outside it are the s_j r_i term + alpha = np.zeros((N, N), dtype=float) for delta, prob in transitions: for i in range(N): j = i + delta if 0 <= j < N: alpha[i, j] = prob - alpha += np.finfo(float).eps - alpha /= alpha.sum(axis=1, keepdims=True) + row_sums = alpha.sum(axis=1, keepdims=True) + alpha = np.divide(alpha, row_sums, out=np.zeros_like(alpha), where=row_sums > 0) return alpha, D1, D2 @@ -231,7 +227,7 @@ def __init__( tempo: float = 120.0, sample_rate: int = 16000, hop_length: int = 320, - s_j: float = 1e-5, + s_j: float = 1e-100, r_i: Optional[np.ndarray] = None, **kwargs, ) -> None: @@ -511,12 +507,7 @@ def forward_step( ) emit_pause = np.full(N, emit_pause_scalar, dtype=float) - # Spectral-flux-driven exit boost - flux = float(obs[88]) if obs.size > 88 else 0.0 - f = flux / (flux + 1.0) # [0,1) - boost = 1.0 + _FLUX_EXIT_BOOST * f - e0 = np.clip(self.e0 * boost, 1e-10, 1.0 - self.a01 - 1e-10) - a00 = np.clip(1.0 - self.a01 - e0, 1e-10, 1.0 - 1e-10) + e0, a00 = self.e0, self.a00 # Exit masses from each top state j (Eq.(6)) exit_mass = prev_sound * e0 + prev_pause * self.e1 # (N,) From 7688c45292db6612b5688857f01d921405760ba4 Mon Sep 17 00:00:00 2001 From: "jiyun.park" Date: Sat, 3 Oct 2026 11:31:41 +0200 Subject: [PATCH 3/9] SKF chord lengths in quarter notes; score tempo in quarter-note BPM build_chord_sequence measured note lengths in score beats divided by 4, which is wrong whenever the beat is not a quarter note; use onset_quarter. The tempo read from a score marking now honours the marking's unit (e.g. half = 100 is 200 quarter-note BPM). --- matchmaker/prob/skf.py | 12 ++++++------ matchmaker/utils/misc.py | 4 +++- 2 files changed, 9 insertions(+), 7 deletions(-) diff --git a/matchmaker/prob/skf.py b/matchmaker/prob/skf.py index 4e104f0..4d0fe76 100644 --- a/matchmaker/prob/skf.py +++ b/matchmaker/prob/skf.py @@ -62,9 +62,10 @@ def build_chord_sequence(note_array: np.ndarray): """ onsets = note_array["onset_beat"] pitches = note_array["pitch"] - durations = note_array["duration_beat"] + # score time in quarters is independent of the time signature's beat unit + quarters = note_array["onset_quarter"] if "onset_quarter" in note_array.dtype.names else onsets - unique_onsets = np.unique(onsets) + unique_onsets, first = np.unique(onsets, return_index=True) K = len(unique_onsets) chords: List[List[int]] = [] @@ -75,11 +76,10 @@ def build_chord_sequence(note_array: np.ndarray): chords.append(pitches[mask].tolist()) onset_beats[k] = ub - # Nominal length l_k = onset_{k+1} - onset_k (in beat units, then /4 - # to convert to whole-note units since 1 whole note = 4 beats) + # Nominal length l_k = onset_{k+1} - onset_k in whole notes (4 quarters) + onset_quarters = quarters[first] lengths = np.zeros(K) - for k in range(K - 1): - lengths[k] = (onset_beats[k + 1] - onset_beats[k]) / 4.0 + lengths[:-1] = np.diff(onset_quarters) / 4.0 # Last chord: use same length as previous, or 1 beat lengths[K - 1] = lengths[K - 2] if K > 1 else 0.25 diff --git a/matchmaker/utils/misc.py b/matchmaker/utils/misc.py index 884442d..73640a7 100644 --- a/matchmaker/utils/misc.py +++ b/matchmaker/utils/misc.py @@ -15,6 +15,7 @@ import numpy as np import partitura from partitura.score import ScoreLike +from partitura.utils.music import to_quarter_tempo # Tempo marking to BPM mapping # Reference: https://en.wikipedia.org/wiki/Tempo#Basic_tempo_markings @@ -262,7 +263,8 @@ def get_tempo_from_score( try: for tempo_obj in score_part.iter_all(partitura.score.Tempo): if hasattr(tempo_obj, "bpm") and tempo_obj.bpm is not None: - return float(tempo_obj.bpm) + # the marking's unit ("h" = 100 is 200 quarters per minute) + return float(to_quarter_tempo(tempo_obj.unit or "q", tempo_obj.bpm)) except Exception: pass From c7b09a489e6f9123de05fdd8f104fcba9af01df0 Mon Sep 17 00:00:00 2001 From: "jiyun.park" Date: Sat, 3 Oct 2026 11:31:41 +0200 Subject: [PATCH 4/9] Add arzt_tempo and arzt_multi_fold baselines arzt_tempo: frame-level Arzt OLTW with the simple tempo model of Arzt & Widmer (SMC 2010), computed from a backward path (dtw_loop records backpointers). arzt_multi_fold: for folded scores, one Arzt matcher per unfolded version of the score, outputting the matcher with the lowest recent cost (adapted from Arzt, Widmer & Dixon, ECAI 2008, and Widmer & Arzt, Vienna Talk 2010). --- README.md | 2 + matchmaker/dp/__init__.py | 2 + matchmaker/dp/dtw_loop.pyx | 17 +- matchmaker/dp/oltw_arzt.py | 304 +++++++++++++++++++++- matchmaker/dp/oltw_arzt_multi_fold.py | 355 ++++++++++++++++++++++++++ matchmaker/methods.yaml | 31 +++ tests/test_registry.py | 4 +- tests/test_tempo_models.py | 201 +++++++++++++++ 8 files changed, 911 insertions(+), 5 deletions(-) create mode 100644 matchmaker/dp/oltw_arzt_multi_fold.py create mode 100644 tests/test_tempo_models.py diff --git a/README.md b/README.md index a181751..0822c1d 100644 --- a/README.md +++ b/README.md @@ -349,6 +349,8 @@ Default method: `"arzt"` | `"dixon"` | On-line time warping by Dixon (2005) | | `"outerhmm"` | Outer-product HMM score follower by Nakamura (2014) | | `"skf"` | Switching Kalman Filter with hidden tempo by Jiang and Raphael (2020) | +| `"arzt_tempo"` | On-line time warping by Arzt with a tempo model (Arzt and Widmer, 2010) | +| `"arzt_multi_fold"` | Arzt OLTW tracking several unfolded score variants in parallel | ### MIDI (`input_type="midi"`) diff --git a/matchmaker/dp/__init__.py b/matchmaker/dp/__init__.py index 4a063d3..7d7e8ed 100644 --- a/matchmaker/dp/__init__.py +++ b/matchmaker/dp/__init__.py @@ -12,9 +12,11 @@ OnlineTimeWarpingArzt, OnlineTimeWarpingArztEvent, OnlineTimeWarpingArztFrame, + OnlineTimeWarpingArztTempoFrame, ) from .oltw_dixon import ( OnlineTimeWarpingDixon, OnlineTimeWarpingDixonEvent, OnlineTimeWarpingDixonFrame, ) +from .oltw_arzt_multi_fold import OnlineTimeWarpingArztMultiFold diff --git a/matchmaker/dp/dtw_loop.pyx b/matchmaker/dp/dtw_loop.pyx index 4ec4379..b1f8242 100644 --- a/matchmaker/dp/dtw_loop.pyx +++ b/matchmaker/dp/dtw_loop.pyx @@ -23,7 +23,8 @@ def oltw_arzt_loop( int window_end, int input_index, float min_costs, - int min_index + int min_index, + signed char[:] backpointers = None ): """ Cythonized main loop for online time warping. @@ -46,6 +47,8 @@ def oltw_arzt_loop( Minimum costs of current window. min_index: int Index of minimum costs of current window. + backpointers: numpy.ndarray, optional + 1D array of shape (window_end - window_start,) to record predecessor steps. Returns ------- @@ -56,7 +59,7 @@ def oltw_arzt_loop( min_costs: float Minimum costs of current window. """ - cdef float dist1, dist2, dist3, local_dist, norm_cost + cdef float dist1, dist2, dist3, min_dist, local_dist, norm_cost cdef int idx = 0, score_index = window_start cdef Py_ssize_t k, wk = window_cost.shape[0] cdef Py_ssize_t N = global_cost_matrix.shape[0] @@ -84,6 +87,14 @@ def oltw_arzt_loop( min_dist = min(dist1, dist2, dist3) global_cost_matrix[score_index + 1, 1] = min_dist + if backpointers is not None: + if dist1 <= dist2 and dist1 <= dist3: + backpointers[idx] = 1 + elif dist2 <= dist3: + backpointers[idx] = 2 + else: + backpointers[idx] = 3 + norm_cost = min_dist / (input_index + score_index + 1.0) # check if new cell has lower costs and might be current position @@ -96,7 +107,7 @@ def oltw_arzt_loop( update_cost_matrix(global_cost_matrix, N) - return global_cost_matrix, min_index, min_costs + return np.asarray(global_cost_matrix), min_index, min_costs @cython.boundscheck(False) @cython.wraparound(False) diff --git a/matchmaker/dp/oltw_arzt.py b/matchmaker/dp/oltw_arzt.py index c74c9bd..1dc789d 100644 --- a/matchmaker/dp/oltw_arzt.py +++ b/matchmaker/dp/oltw_arzt.py @@ -84,7 +84,6 @@ def __init__( ) self.N_ref: int = self.reference_features.shape[0] self.step_size: int = step_size - self.reset() def reset(self) -> None: self.current_index = 0 @@ -165,6 +164,7 @@ def __init__( "min_latency": float("inf"), } self._init_distance_func(distance_func) + self.reset() def __call__(self, observation: Any, perf_time: float) -> float: t0 = time.time() @@ -278,6 +278,307 @@ def step(self, input_features: NDArray[np.float32]) -> None: self.input_index += 1 +class OnlineTimeWarpingArztTempoFrame(OnlineTimeWarpingArztFrame): + """Frame-level OLTW with Arzt & Widmer (2010) Simple Tempo Model. + + Dynamically tracks relative tempo via rectified backward path + and stretches/compresses reference score features on the fly. + """ + + def __init__( + self, + reference_features: NDArray[np.float32], + score_positions: NDArray[np.float32], + window_size: int = WINDOW_SIZE, + step_size: int = STEP_SIZE, + distance_func: Union[ + str, Callable, Tuple[str, Dict[str, Any]] + ] = OnlineTimeWarpingArztFrame.DEFAULT_DISTANCE_FUNC, + start_window_size: Union[float, int] = START_WINDOW_SIZE, + frame_rate: int = FRAME_RATE, + ref_frame_to_beat: NDArray = None, + queue: Optional[RECVQueue] = None, + tempo_window_size: float = 3.0, + tempo_min_past: float = 1.0, + tempo_n_onsets: int = 20, + random_seed: Optional[int] = 1984, + use_tempo_model: bool = True, + **kwargs, + ) -> None: + if ref_frame_to_beat is None: + raise ValueError( + "Frame-level Arzt requires `ref_frame_to_beat` (per-frame beat mapping)." + ) + self.use_tempo_model = use_tempo_model + self.tempo_window_size = tempo_window_size + self.tempo_min_past = tempo_min_past + self.tempo_n_onsets = tempo_n_onsets + self.random_seed = random_seed + self._original_reference_features = reference_features.copy() + self._original_ref_frame_to_beat = ref_frame_to_beat.copy() + super().__init__( + reference_features=reference_features, + score_positions=score_positions, + window_size=window_size, + step_size=step_size, + distance_func=distance_func, + start_window_size=start_window_size, + frame_rate=frame_rate, + ref_frame_to_beat=ref_frame_to_beat, + queue=queue, + **kwargs, + ) + self.reset() + + def reset(self) -> None: + super().reset() + self.current_relative_tempo: float = 1.0 + self._consecutive_alterations: int = 0 + self._backpointers: list = [] + self._rng = np.random.RandomState(self.random_seed) + # The tempo model stretches the reference on the fly; restore the + # pristine copy so each run starts from the notated score. + self.reference_features = self._original_reference_features.copy() + self._ref_frame_to_beat = self._original_ref_frame_to_beat.copy() + self.N_ref = self.reference_features.shape[0] + self.global_cost_matrix = np.full( + (self.N_ref + 1, 2), np.inf, dtype=np.float32 + ) + self._init_onset_mask() + + def _init_onset_mask(self) -> None: + self.is_onset_frame = np.zeros(self.N_ref, dtype=bool) + onset_frames = np.searchsorted(self._ref_frame_to_beat, self.score_positions) + valid = onset_frames < self.N_ref + self.is_onset_frame[onset_frames[valid]] = True + + def _get_backward_path(self, max_history_frames: int = 500) -> list: + """Trace backward path from current position using recorded backpointers.""" + if not self._backpointers or self.input_index <= 0: + return [(self._current_frame, 0)] + + path = [] + s = self._current_frame + i = self.input_index - 1 + min_i = max(0, i - max_history_frames) + while i >= min_i and s >= 0: + path.append((s, i)) + if i >= len(self._backpointers): + i -= 1 + s = max(0, s - 1) + continue + w_start, w_end, bp = self._backpointers[i] + if bp is not None and w_start <= s < w_end: + code = bp[s - w_start] + if code == 1: + s -= 1 + elif code == 2: + i -= 1 + elif code == 3: + s -= 1 + i -= 1 + else: + break + else: + s = max(0, s - 1) + i -= 1 + path.reverse() + return path + + def _compute_relative_tempo(self) -> float: + """Estimate current relative tempo via rectified backward path (Arzt 2010 Sec. 4.1).""" + if self.input_index < int(self.frame_rate * 1.5): + return 1.0 + + cur_perf_time = self.input_index / self.frame_rate + max_history = int(self.frame_rate * 10) + path = self._get_backward_path(max_history_frames=max_history) + if len(path) < 2: + return 1.0 + + score_frames = np.array([p[0] for p in path]) + perf_frames = np.array([p[1] for p in path]) + + s_min, s_max = int(score_frames[0]), int(score_frames[-1]) + if s_max <= s_min: + return 1.0 + + local_onsets = np.where(self.is_onset_frame[s_min : s_max + 1])[0] + s_min + if len(local_onsets) < 2: + return 1.0 + + t_s_list = [] + t_p_list = [] + for sf in local_onsets: + idx = int(np.searchsorted(score_frames, sf)) + if idx < len(score_frames): + t_s_list.append(sf / self.frame_rate) + t_p_list.append(perf_frames[idx] / self.frame_rate) + + if len(t_s_list) < 2: + return 1.0 + + t_s_arr = np.array(t_s_list) + t_p_arr = np.array(t_p_list) + + # Onsets at least tempo_min_past seconds in the past + valid_mask = t_p_arr <= (cur_perf_time - self.tempo_min_past) + valid_indices = np.where(valid_mask)[0] + if len(valid_indices) == 0: + return 1.0 + + if len(valid_indices) > self.tempo_n_onsets: + valid_indices = valid_indices[-self.tempo_n_onsets:] + + half_win = self.tempo_window_size / 2.0 + tempi = [] + for idx in valid_indices: + ts = t_s_arr[idx] + w_start = ts - half_win + w_end = ts + half_win + if w_start >= t_s_arr[0] and w_end <= t_s_arr[-1]: + tp_start = float(np.interp(w_start, t_s_arr, t_p_arr)) + tp_end = float(np.interp(w_end, t_s_arr, t_p_arr)) + delta_p = tp_end - tp_start + delta_s = w_end - w_start + if delta_p > 1e-4: + tempi.append(delta_s / delta_p) + else: + tempi.append(1.0) + + if not tempi: + return 1.0 + + m = len(tempi) + weights = np.arange(1, m + 1) + t_est = float(np.sum(np.array(tempi) * weights) / np.sum(weights)) + return float(np.clip(t_est, 0.33, 3.0)) + + def _alter_score_representation(self, t: float) -> None: + """Alter score representation on the fly according to relative tempo (Arzt 2010 Sec. 4.2).""" + ps = self._current_frame + if ps + 2 >= self.N_ref: + return + + if self._consecutive_alterations >= 3: + self._consecutive_alterations = 0 + return + + r = self._rng.uniform(0.0, 1.0) + + if t > 1.0: + threshold = 1.0 / t + if r > threshold: + if self.is_onset_frame[ps + 1] or self.is_onset_frame[ps + 2]: + self._consecutive_alterations = 0 + return + + mean_feat = 0.5 * ( + self.reference_features[ps + 1] + self.reference_features[ps + 2] + ) + self.reference_features[ps + 1] = mean_feat + self.reference_features = np.delete( + self.reference_features, ps + 2, axis=0 + ) + + mean_beat = 0.5 * ( + self._ref_frame_to_beat[ps + 1] + self._ref_frame_to_beat[ps + 2] + ) + self._ref_frame_to_beat[ps + 1] = mean_beat + self._ref_frame_to_beat = np.delete(self._ref_frame_to_beat, ps + 2) + + self.is_onset_frame = np.delete(self.is_onset_frame, ps + 2) + self.global_cost_matrix = np.delete( + self.global_cost_matrix, ps + 3, axis=0 + ) + self.N_ref = self.reference_features.shape[0] + self._consecutive_alterations += 1 + else: + self._consecutive_alterations = 0 + + elif t < 1.0: + threshold = t + if r > threshold: + if self.is_onset_frame[ps] or self.is_onset_frame[ps + 1]: + self._consecutive_alterations = 0 + return + + mean_feat = 0.5 * ( + self.reference_features[ps] + self.reference_features[ps + 1] + ) + self.reference_features = np.insert( + self.reference_features, ps + 1, mean_feat, axis=0 + ) + + mean_beat = 0.5 * ( + self._ref_frame_to_beat[ps] + self._ref_frame_to_beat[ps + 1] + ) + self._ref_frame_to_beat = np.insert( + self._ref_frame_to_beat, ps + 1, mean_beat + ) + + self.is_onset_frame = np.insert(self.is_onset_frame, ps + 1, False) + row1 = np.asarray(self.global_cost_matrix[ps + 1]) + row2 = np.asarray(self.global_cost_matrix[ps + 2]) + new_cost_row = 0.5 * (row1 + row2) + new_cost_row[1] = np.inf + self.global_cost_matrix = np.insert( + self.global_cost_matrix, ps + 2, new_cost_row, axis=0 + ) + self.N_ref = self.reference_features.shape[0] + self._consecutive_alterations += 1 + else: + self._consecutive_alterations = 0 + else: + self._consecutive_alterations = 0 + + def step(self, input_features: NDArray[np.float32]) -> None: + if self.use_tempo_model and self.input_index > 0: + self.current_relative_tempo = self._compute_relative_tempo() + self._alter_score_representation(self.current_relative_tempo) + + min_costs = np.inf + min_index = max(self.window_index - self.step_size, 0) + + window_start, window_end = self.get_window() + window_cost = self.vdist( + self.reference_features[window_start:window_end], + input_features.squeeze(), + self.distance_func, + ) + + bp = None + if self.use_tempo_model: + bp = np.zeros(window_end - window_start, dtype=np.int8) + + self.global_cost_matrix, min_index, min_costs = oltw_arzt_loop( + global_cost_matrix=self.global_cost_matrix, + window_cost=window_cost, + window_start=window_start, + window_end=window_end, + input_index=self.input_index, + min_costs=min_costs, + min_index=min_index, + backpointers=bp, + ) + + if self.use_tempo_model: + self._backpointers.append((window_start, window_end, bp)) + + if self.input_index > 0: + self._current_frame = int( + np.clip( + min_index, + self._current_frame, + self._current_frame + self.step_size, + ) + ) + else: + self._current_frame = min_index + self.current_index = self._frame_to_score_idx(self._current_frame) + self.input_index += 1 + + # --------------------------------------------------------------------------- # Event-level subclass (MIDI) # --------------------------------------------------------------------------- @@ -313,6 +614,7 @@ def __init__( self._window_size = window_size self._start_window_size = start_window_size self._scipy_metric = distance_func + self.reset() def get_window(self) -> Tuple[int, int]: w = self._window_size diff --git a/matchmaker/dp/oltw_arzt_multi_fold.py b/matchmaker/dp/oltw_arzt_multi_fold.py new file mode 100644 index 0000000..9dfabf7 --- /dev/null +++ b/matchmaker/dp/oltw_arzt_multi_fold.py @@ -0,0 +1,355 @@ +#!/usr/bin/python +# -*- coding: utf-8 -*- +""" +Parallel Multi-Fold Online Time Warping (OLTWArztMultiFold) + +A baseline for folded scores with repeats. The score is unfolded into a few versions +(every repeat taken, the folded score itself, and the first four versions partitura's +``iter_unfolded_parts`` yields, duplicates removed); one frame-level Arzt OLTW matcher +(``OnlineTimeWarpingArztFrame``) runs on each version from the start, and the output, in +notated (folded) beats, is the matcher whose local matching cost averaged over the last +``selection_window`` frames is lowest, switching only when another is lower by +``switch_margin``. + +The idea of running several OLTW matchers in parallel and comparing their alignment costs +is adapted from + A. Arzt, G. Widmer and S. Dixon, "Automatic page turning for musicians via real-time + machine listening", Proc. ECAI, 2008 (Strategy 3) + G. Widmer and A. Arzt, "Robust real-time music tracking", Proc. Vienna Talk, 2010. +The unfolded versions, the cost window and the switching margin are this implementation's +own choices, not those papers'. +""" + +from __future__ import annotations + +import collections +import copy +import re +import time +from typing import Any, Callable, Dict, Generator, List, Optional, Tuple, Union + +import numpy as np +import partitura as pt +from numpy.typing import NDArray +from partitura.score import Part, merge_parts + +from matchmaker.base import OnlineAlignment +from matchmaker.dp.oltw_arzt import ( + OnlineTimeWarpingArztFrame, + START_WINDOW_SIZE, + STEP_SIZE, + WINDOW_SIZE, +) +from matchmaker.features.audio import FRAME_RATE +from matchmaker.features.processor import Processor +from matchmaker.io.audio import QUEUE_TIMEOUT +from matchmaker.io.queue import RECVQueue +from matchmaker.utils.misc import generate_score_audio, set_latency_stats +from partitura.io.exportmidi import get_ppq + + +class OnlineTimeWarpingArztMultiFold(OnlineTimeWarpingArztFrame): + """Multi-fold parallel OLTW tracker for folded scores with repeats. + + Parameters + ---------- + score_part : Part + The original (folded) partitura Part. + tempo : float + Nominal score tempo in BPM. + sample_rate : int + Audio sample rate. + processor : Processor + Audio feature processor (e.g. ChromaProcessor). + frame_rate : int + Audio frame rate (default: FRAME_RATE, 30). + window_size : int + Search window size in seconds (default: 10). + step_size : int + Maximum forward advance per input frame (default: 3). + start_window_size : float or int + Window size during warmup (default: 0.1). + selection_window : int + Number of trailing frames over which running matching cost is averaged + for hypothesis selection (default: 50 frames, about 1.7 s at 30 fps). + switch_margin : float + Hysteresis factor requiring an alternative hypothesis to have cost + strictly lower by this fraction before switching (default: 0.05). + queue : Optional[RECVQueue] + Input queue for streaming observations. + """ + + def __init__( + self, + reference_features: Optional[NDArray[np.float32]] = None, + score_positions: Optional[NDArray[np.float32]] = None, + score_part: Optional[Part] = None, + tempo: Optional[float] = None, + sample_rate: int = 44100, + processor: Optional[Processor] = None, + frame_rate: int = FRAME_RATE, + window_size: int = WINDOW_SIZE, + step_size: int = STEP_SIZE, + start_window_size: Union[float, int] = START_WINDOW_SIZE, + distance_func: str = "Manhattan", + selection_window: int = 50, + switch_margin: float = 0.05, + queue: Optional[RECVQueue] = None, + ref_frame_to_beat: Optional[NDArray] = None, + **kwargs, + ) -> None: + if score_part is None: + raise ValueError("OnlineTimeWarpingArztMultiFold requires `score_part`.") + if tempo is None: + tempo = 120.0 + + self.score_part = score_part + self.tempo = float(tempo) + self.sample_rate = sample_rate + self.processor = processor + self.frame_rate = frame_rate + self.window_size = window_size + self.step_size = step_size + self.start_window_size = start_window_size + self.distance_func_name = distance_func + self.selection_window = selection_window + self.switch_margin = switch_margin + self.queue_timeout = QUEUE_TIMEOUT + + # Folded note lookup: original note ID -> notated beat + folded_notes = self.score_part.note_array() + self.folded_note_dict = { + str(n["id"]): float(n["onset_beat"]) for n in folded_notes + } + notated_positions = np.unique(folded_notes["onset_beat"]).astype(np.float32) + + ref_feat = ( + reference_features + if reference_features is not None + else np.empty((0, 12), dtype=np.float32) + ) + dummy_ref_beat = ( + ref_frame_to_beat + if ref_frame_to_beat is not None + else np.zeros(max(1, len(ref_feat)), dtype=np.float32) + ) + + # Build parallel track hypotheses from the score repeat structure + # before the parent __init__ so that reset() always finds them. + self._build_hypotheses() + + super().__init__( + reference_features=ref_feat, + score_positions=notated_positions, + window_size=window_size, + step_size=step_size, + start_window_size=start_window_size, + distance_func=distance_func, + frame_rate=frame_rate, + ref_frame_to_beat=dummy_ref_beat, + queue=queue, + **kwargs, + ) + self.reset() + + def _build_hypotheses(self) -> None: + """Enumerate distinct structural fold clones safely and initialize trackers.""" + import itertools + import sys + + _prev_limit = sys.getrecursionlimit() + clones = [] + try: + sys.setrecursionlimit(max(_prev_limit, 25_000)) + # 1. Maximal unfold (canonical repeat-performed performance) + try: + max_unfolded = pt.score.unfold_part_maximal( + self.score_part, ignore_leaps=False + ) + clones.append( + merge_parts(max_unfolded.parts) + if hasattr(max_unfolded, "parts") + else max_unfolded + ) + except Exception: + pass + + # 2. Minimal / folded score (repeat-skipped performance) + clones.append(self.score_part) + + # 3. Intermediate unfold variants (bounded to prevent combinatorial explosion) + try: + gen = pt.score.iter_unfolded_parts(self.score_part, update_ids=True) + for c in itertools.islice(gen, 4): + clones.append(c) + except Exception: + pass + finally: + sys.setrecursionlimit(_prev_limit) + + # Deduplicate clones by note count and duration + unique_clones = [] + seen = set() + for c in clones: + na = c.note_array() + key = (len(na), round(float(na["onset_beat"].max()), 2)) + if key not in seen: + seen.add(key) + unique_clones.append(c) + + if not unique_clones: + unique_clones = [self.score_part] + + self.hypotheses: List[Dict[str, Any]] = [] + tick = get_ppq(self.score_part) + + for i, clone in enumerate(unique_clones): + # 1. Synthesize audio and extract reference features + clone_audio = generate_score_audio(clone, self.tempo, self.sample_rate) + if self.processor is not None: + features, _ = self.processor((clone_audio.astype(np.float32), 0.0)) + self.processor.reset() + else: + raise ValueError( + "OnlineTimeWarpingArztMultiFold requires an audio processor." + ) + + n_ref_frames = features.shape[0] + + # 2. Frame to unfolded beat + times = ( + (np.arange(n_ref_frames) / self.frame_rate) + * tick + * (self.tempo / 60.0) + ) + ref_frame_to_unfolded_beat = np.array( + [float(clone.beat_map(t)) for t in times], dtype=np.float32 + ) + score_positions_k = np.unique(clone.note_array()["onset_beat"]).astype( + np.float32 + ) + + # 3. Unfolded beat -> Notated beat mapping via note ID correspondence + clone_notes = clone.note_array() + u_beats = [] + n_beats = [] + for n in clone_notes: + orig_id = re.sub(r"-\d+$", "", str(n["id"])) + if orig_id in self.folded_note_dict: + u_beats.append(float(n["onset_beat"])) + n_beats.append(self.folded_note_dict[orig_id]) + + u_beats = np.asarray(u_beats, dtype=float) + n_beats = np.asarray(n_beats, dtype=float) + sort_idx = np.argsort(u_beats) + u_sorted = u_beats[sort_idx] + n_sorted = n_beats[sort_idx] + u_uniq, u_first = np.unique(u_sorted, return_index=True) + n_uniq = n_sorted[u_first] + + # Precompute frame to notated beat directly + if len(n_uniq) > 0: + ref_frame_to_notated_beat = np.interp( + ref_frame_to_unfolded_beat, + u_uniq, + n_uniq, + left=n_uniq[0], + right=n_uniq[-1], + ).astype(np.float32) + else: + ref_frame_to_notated_beat = ref_frame_to_unfolded_beat.astype(np.float32) + + # 4. Instantiate follower + tracker = OnlineTimeWarpingArztFrame( + reference_features=features, + score_positions=score_positions_k, + window_size=self.window_size, + step_size=self.step_size, + start_window_size=self.start_window_size, + distance_func=self.distance_func_name, + frame_rate=self.frame_rate, + ref_frame_to_beat=ref_frame_to_unfolded_beat, + queue=None, + ) + + self.hypotheses.append( + { + "index": i, + "clone": clone, + "tracker": tracker, + "ref_frame_to_notated_beat": ref_frame_to_notated_beat, + "recent_costs": collections.deque( + maxlen=self.selection_window + ), + "mean_recent_cost": float("inf"), + } + ) + + def reset(self) -> None: + super().reset() + self.active_hypothesis_idx = 0 + self.latency_stats = { + "total_latency": 0, + "total_frames": 0, + "max_latency": 0, + "min_latency": float("inf"), + } + for hyp in self.hypotheses: + hyp["tracker"].reset() + hyp["recent_costs"].clear() + hyp["mean_recent_cost"] = float("inf") + + def __call__(self, observation: Any, perf_time: float) -> float: + t0 = time.time() + self.step(observation) + current_beat = self.get_current_position() + self._alignment_path.append((float(perf_time), float(current_beat))) + self.latency_stats = set_latency_stats( + time.time() - t0, self.latency_stats, self.input_index + ) + return current_beat + + def step(self, input_features: NDArray[np.float32]) -> None: + obs_squeezed = input_features.squeeze() + + # 1. Advance every hypothesis tracker + for i, hyp in enumerate(self.hypotheses): + tracker: OnlineTimeWarpingArztFrame = hyp["tracker"] + tracker.step(input_features) + + # Measure local matching error at current hypothesis frame + curr_frame = tracker._current_frame + ref_feat = tracker.reference_features[curr_frame] + loc_cost = float(tracker.distance_func(ref_feat, obs_squeezed)) + + hyp["recent_costs"].append(loc_cost) + mean_cost = float(np.mean(hyp["recent_costs"])) + hyp["mean_recent_cost"] = mean_cost + + # 2. Decision Maker: select winning hypothesis with hysteresis + current_hyp_cost = self.hypotheses[self.active_hypothesis_idx][ + "mean_recent_cost" + ] + min_hyp_idx = min( + range(len(self.hypotheses)), + key=lambda i: self.hypotheses[i]["mean_recent_cost"], + ) + min_hyp_cost = self.hypotheses[min_hyp_idx]["mean_recent_cost"] + + if min_hyp_cost < current_hyp_cost * (1.0 - self.switch_margin): + self.active_hypothesis_idx = min_hyp_idx + + self.input_index += 1 + + def get_current_position(self) -> float: + """Return the current position projected into notated (folded) score beats.""" + winning_hyp = self.hypotheses[self.active_hypothesis_idx] + curr_frame = winning_hyp["tracker"]._current_frame + notated_beat = winning_hyp["ref_frame_to_notated_beat"][ + min(curr_frame, len(winning_hyp["ref_frame_to_notated_beat"]) - 1) + ] + return float(notated_beat) + + def run(self, verbose: bool = True) -> Generator[float, None, NDArray]: + self.reset() + return (yield from super().run(verbose=verbose)) diff --git a/matchmaker/methods.yaml b/matchmaker/methods.yaml index 9b96d1d..cabe986 100644 --- a/matchmaker/methods.yaml +++ b/matchmaker/methods.yaml @@ -169,6 +169,37 @@ methods: start_window_size: 0.1 step_size: 3 + arzt_tempo: + family: oltw + class: matchmaker.dp:OnlineTimeWarpingArztTempoFrame + reference: score_audio + args: *oltw_frame_args + config_passthrough: true + default_kwargs: + window_size: 10 + start_window_size: 0.1 + step_size: 3 + use_tempo_model: true + random_seed: 1984 + + arzt_multi_fold: + family: oltw + class: matchmaker.dp:OnlineTimeWarpingArztMultiFold + reference: score_audio + args: + <<: *oltw_frame_args + score_part: {from: score_part} + tempo: {from: tempo} + sample_rate: {from: sample_rate} + processor: {from: processor} + config_passthrough: true + default_kwargs: + window_size: 10 + start_window_size: 0.1 + step_size: 3 + selection_window: 50 + switch_margin: 0.05 + dixon: family: oltw class: matchmaker.dp:OnlineTimeWarpingDixonFrame diff --git a/tests/test_registry.py b/tests/test_registry.py index 629f627..ef88f08 100644 --- a/tests/test_registry.py +++ b/tests/test_registry.py @@ -97,7 +97,9 @@ def test_default_method_and_processor_exist(self): ) def test_families_are_populated(self): - self.assertEqual(OLTW_METHODS, {"arzt", "dixon"}) + self.assertEqual( + OLTW_METHODS, {"arzt", "dixon", "arzt_tempo", "arzt_multi_fold"} + ) self.assertEqual(PARANGONAR_METHODS, {"OPTM", "OTM", "SLT_OLTW", "SL_OLTW"}) def test_obj_tag_resolves_to_the_real_object(self): diff --git a/tests/test_tempo_models.py b/tests/test_tempo_models.py new file mode 100644 index 0000000..131683a --- /dev/null +++ b/tests/test_tempo_models.py @@ -0,0 +1,201 @@ +#!/usr/bin/env python +# -*- coding: utf-8 -*- +""" +Tests for Arzt (2010) Simple Tempo Model and Raphael/Jiang (2020) SKF. +""" + +import unittest +import numpy as np + +from matchmaker.dp.oltw_arzt import ( + OnlineTimeWarpingArztFrame, + OnlineTimeWarpingArztTempoFrame, +) +from matchmaker.prob.skf import ( + SwitchingKalmanFilterFollower, + build_chord_sequence, + build_spectral_templates, +) +from matchmaker.registry import REGISTRY +from tests.utils import generate_example_sequences + +RNG = np.random.RandomState(1984) + + +class TestArztTempoModel(unittest.TestCase): + """Tests for Arzt & Widmer (2010 SMC) Simple Tempo Model.""" + + def setUp(self): + # Generate synthetic reference and performance sequences + self.frame_rate = 50 + # 10 seconds of score audio -> 500 frames + n_frames = 500 + n_features = 12 + self.X = RNG.randn(n_frames, n_features).astype(np.float32) + # 1 beat per second -> 10 beats total + # Every 50 frames is 1 beat + self.ref_frame_to_beat = (np.arange(n_frames, dtype=np.float32) / self.frame_rate) + # Score positions at beats 0.0, 1.0, 2.0, ..., 9.0 + self.score_positions = np.arange(10, dtype=np.float32) + + def test_initialization(self): + tracker = OnlineTimeWarpingArztTempoFrame( + reference_features=self.X, + score_positions=self.score_positions, + ref_frame_to_beat=self.ref_frame_to_beat, + frame_rate=self.frame_rate, + window_size=5, + step_size=3, + ) + self.assertTrue(tracker.use_tempo_model) + self.assertEqual(tracker.current_relative_tempo, 1.0) + self.assertEqual(tracker._consecutive_alterations, 0) + self.assertEqual(len(tracker._backpointers), 0) + self.assertTrue(tracker.is_onset_frame[0]) + self.assertTrue(tracker.is_onset_frame[50]) + + def test_run_steps(self): + tracker = OnlineTimeWarpingArztTempoFrame( + reference_features=self.X, + score_positions=self.score_positions, + ref_frame_to_beat=self.ref_frame_to_beat, + frame_rate=self.frame_rate, + window_size=5, + step_size=3, + ) + + # Feed 100 frames of observations + for i in range(100): + obs = self.X[min(i, len(self.X) - 1)] + RNG.randn(12).astype(np.float32) * 0.01 + perf_time = i / self.frame_rate + beat = tracker(obs, perf_time) + self.assertIsInstance(beat, float) + self.assertGreaterEqual(beat, 0.0) + + # Check that backpointers were recorded + self.assertEqual(len(tracker._backpointers), 100) + path = tracker.alignment_path + self.assertEqual(path.shape[0], 2) + self.assertEqual(path.shape[1], 100) + + def test_reset(self): + tracker = OnlineTimeWarpingArztTempoFrame( + reference_features=self.X, + score_positions=self.score_positions, + ref_frame_to_beat=self.ref_frame_to_beat, + frame_rate=self.frame_rate, + ) + orig_len = tracker.N_ref + + # Run 50 frames + for i in range(50): + obs = self.X[i] + tracker(obs, i / self.frame_rate) + + # Reset + tracker.reset() + self.assertEqual(tracker.current_index, 0) + self.assertEqual(tracker._current_frame, 0) + self.assertEqual(tracker.current_relative_tempo, 1.0) + self.assertEqual(len(tracker._backpointers), 0) + self.assertEqual(tracker.N_ref, orig_len) + + def test_registry_lookup(self): + spec = REGISTRY.method("audio", "arzt_tempo") + self.assertEqual(spec.cls_path, "matchmaker.dp:OnlineTimeWarpingArztTempoFrame") + self.assertTrue(spec.default_kwargs.get("use_tempo_model")) + + +class TestRaphaelSwitchingStateSpace(unittest.TestCase): + """Tests for Raphael & Jiang (2020 ISMIR) Switching State-Space Model.""" + + def setUp(self): + # Create a simple synthetic note_array + self.note_array = np.zeros( + 4, + dtype=[ + ("pitch", "i4"), + ("onset_beat", "f4"), + ("duration_beat", "f4"), + ], + ) + self.note_array["pitch"] = [60, 64, 67, 72] # C major chord arpeggio + self.note_array["onset_beat"] = [0.0, 1.0, 2.0, 3.0] + self.note_array["duration_beat"] = [1.0, 1.0, 1.0, 1.0] + + def test_build_chord_sequence(self): + chords, lengths, onset_beats = build_chord_sequence(self.note_array) + self.assertEqual(len(chords), 4) + self.assertEqual(chords[0], [60]) + self.assertEqual(chords[1], [64]) + np.testing.assert_array_equal(onset_beats, [0.0, 1.0, 2.0, 3.0]) + # 1 beat = 0.25 whole note + np.testing.assert_allclose(lengths, [0.25, 0.25, 0.25, 0.25]) + + def test_build_spectral_templates(self): + chords, _, _ = build_chord_sequence(self.note_array) + templates = build_spectral_templates(chords, n_fft=512, sample_rate=8000) + self.assertEqual(templates.shape[0], 4) + self.assertEqual(templates.shape[1], 257) + # Each template row should sum to 1 + np.testing.assert_allclose(templates.sum(axis=1), np.ones(4), atol=1e-5) + + def test_initialization_and_step(self): + follower = SwitchingKalmanFilterFollower( + reference_features=self.note_array, + tempo=120.0, + sample_rate=8000, + n_fft=512, + hop_length=128, + max_hypotheses=50, + ) + self.assertEqual(follower.K, 4) + self.assertEqual(follower.current_index, 0) + # Initial tempo: 240 / 120 = 2.0 s/whole-note + self.assertAlmostEqual(follower.init_tempo, 2.0) + + # Process 20 frames + delta = 128 / 8000.0 + for i in range(20): + dummy_spectrum = np.random.rand(257).astype(np.float32) + pos = follower(dummy_spectrum, perf_time=i * delta) + self.assertIsInstance(pos, float) + self.assertGreaterEqual(pos, 0.0) + self.assertLessEqual(pos, 4.0) + + # Verify hypotheses count is bounded by max_hypotheses + self.assertLessEqual(len(follower.hypotheses), 50) + path = follower.alignment_path + self.assertEqual(path.shape, (2, 20)) + + def test_kalman_tempo_adaptation(self): + follower = SwitchingKalmanFilterFollower( + reference_features=self.note_array, + tempo=120.0, + sample_rate=8000, + n_fft=512, + hop_length=128, + ) + # Test transition Kalman update directly: + # Expected duration = length * tempo = 0.25 * 2.0 = 0.5s. + # Suppose performer played chord 0 in 0.25s (twice as fast, tempo should decrease s/wn): + fast_age = int(0.25 / follower.delta) + mu_new, var_new = follower._kalman_update( + mu_t_prev=2.0, var_t_prev=0.04, k_prev=0, age_prev=fast_age + ) + self.assertLess(mu_new, 2.0) + + # Suppose performer played chord 0 in 1.0s (twice as slow, tempo should increase s/wn): + slow_age = int(1.0 / follower.delta) + mu_new_slow, var_new_slow = follower._kalman_update( + mu_t_prev=2.0, var_t_prev=0.04, k_prev=0, age_prev=slow_age + ) + self.assertGreater(mu_new_slow, 2.0) + + def test_registry_lookup(self): + spec = REGISTRY.method("audio", "skf") + self.assertEqual(spec.cls_path, "matchmaker.prob.skf:SwitchingKalmanFilterFollower") + + +if __name__ == "__main__": + unittest.main() From 03318b912e1639468b09a6a299255f41ea9a9d97 Mon Sep 17 00:00:00 2001 From: "jiyun.park" Date: Sat, 3 Oct 2026 11:31:41 +0200 Subject: [PATCH 5/9] run_examples: choose the example piece, custom files, unfolding and method kwargs --- run_examples.py | 72 ++++++++++++++++++++++++++++++++++++------------- 1 file changed, 54 insertions(+), 18 deletions(-) diff --git a/run_examples.py b/run_examples.py index 5665621..242d136 100644 --- a/run_examples.py +++ b/run_examples.py @@ -87,16 +87,17 @@ def plot_alignment_path(perf_sec, score_beat, gt_perf, gt_score, save_path, run_ plt.close(fig) -def save_results(save_dir, run_name, results, perf_sec, score_beat, gt_perf, gt_score): - """Save wp/gt TSVs, results JSON, and the alignment path plot.""" +def save_results(save_dir, run_name, results, perf_sec, score_beat, gt_perf, gt_score, save_plots=True): + """Save alignment data, metrics, and an optional plot.""" save_dir.mkdir(parents=True, exist_ok=True) save_tsv(np.column_stack([perf_sec, score_beat]), save_dir / f"wp_{run_name}.tsv") save_tsv(np.column_stack([gt_perf, gt_score]), save_dir / f"gt_{run_name}.tsv") with open(save_dir / f"{run_name}.json", "w") as f: json.dump(results, f, indent=4) - plot_alignment_path( - perf_sec, score_beat, gt_perf, gt_score, save_dir / f"{run_name}.png", run_name - ) + if save_plots: + plot_alignment_path( + perf_sec, score_beat, gt_perf, gt_score, save_dir / f"{run_name}.png", run_name + ) def main(): @@ -105,18 +106,46 @@ def main(): group = parser.add_mutually_exclusive_group() group.add_argument("--audio", action="store_true", help="Use audio input mode") group.add_argument("--midi", action="store_true", help="Use MIDI input mode") + parser.add_argument( + "--piece", + type=str, + default="simple_mozart", + choices=list(EXAMPLE_PIECES.keys()), + help="Built-in example piece to run (default: simple_mozart)", + ) parser.add_argument( "--method", type=str, default=None, - help="Score following method (e.g., arzt, dixon, outerhmm, audio_outerhmm)", + help="Score following method (e.g., arzt, arzt_tempo, dixon, outerhmm)", ) + parser.add_argument("--score", type=str, default=None, help="Path to custom score XML/MusicXML") + parser.add_argument("--audio-file", type=str, default=None, help="Path to custom performance audio file") + parser.add_argument("--midi-file", type=str, default=None, help="Path to custom performance MIDI file") + parser.add_argument("--match", type=str, default=None, help="Path to custom .match ground truth file") + parser.add_argument("--unfold", action="store_true", help="Unfold score repetitions during load") + parser.add_argument("--no-plots", action="store_true", help="Skip alignment plots") + parser.add_argument("--output-dir", type=Path, default=ROOT_DIR / "results", help="Result directory") + parser.add_argument("--kwargs", type=json.loads, default={}, help="Method option overrides") args = parser.parse_args() input_mode = "midi" if args.midi else "audio" - performance_file = select_performance_file(input_mode) - print(f"Running matchmaker with the score file ({SCORE_FILE.name})...") + # Resolve piece files + if args.score: + score_file = Path(args.score) + performance_file = Path(args.midi_file if input_mode == "midi" else args.audio_file) + match_file = Path(args.match) if args.match else None + run_name = score_file.stem + else: + piece_cfg = EXAMPLE_PIECES[args.piece] + score_file = Path(piece_cfg["score"]) + performance_file = Path(piece_cfg["midi"] if input_mode == "midi" else piece_cfg["audio"]) + match_file = Path(piece_cfg["match"]) if "match" in piece_cfg else None + run_name = args.piece + + print(f"Performance file: {performance_file.name}, with input mode: {input_mode}") + print(f"Running matchmaker with the score file ({score_file.name})...") print("-" * 50) if args.method is not None: @@ -127,10 +156,12 @@ def main(): # Initialize matchmaker (simulation mode) try: mm = Matchmaker( - score_file=SCORE_FILE, + score_file=score_file, performance_file=performance_file, input_type=input_mode, method=method, + unfold_score=args.unfold, + kwargs=args.kwargs or None, ) except Empty as e: print(f"Error initializing Matchmaker: {e}") @@ -141,17 +172,22 @@ def main(): timestamp = datetime.datetime.now().strftime("%H:%M:%S.%f")[:-3] print(f"[{timestamp}] Current beat position: {current_position}") - print("-" * 50) - print(f"Running evaluation using the match file ({MATCH_FILE.name})...") + if match_file and match_file.is_file(): + print("-" * 50) + print(f"Running evaluation using the match file ({match_file.name})...") - results, perf_sec, score_beat, gt_perf, gt_score = evaluate(mm, MATCH_FILE) - print(f"Evaluation Result: {json.dumps(results, indent=4)}") + results, perf_sec, score_beat, gt_perf, gt_score = evaluate(mm, match_file) + print(f"Evaluation Result: {json.dumps(results, indent=4)}") - results_dir = ROOT_DIR / "results" - save_results( - results_dir, "simple_example", results, perf_sec, score_beat, gt_perf, gt_score - ) - print(f"Detailed evaluation results saved in {results_dir}") + results_dir = args.output_dir + save_results( + results_dir, f"{run_name}_{method}", results, perf_sec, score_beat, gt_perf, gt_score, + save_plots=not args.no_plots, + ) + print(f"Detailed evaluation results saved in {results_dir}") + else: + print("\nAlignment finished (no match file provided for quantitative evaluation).") + return if __name__ == "__main__": From 7fbba8ea5e9cd61a9a23b21f2fe7ddae5d3b2cbc Mon Sep 17 00:00:00 2001 From: "jiyun.park" Date: Sat, 3 Oct 2026 22:22:20 +0200 Subject: [PATCH 6/9] GT onset times from note_on, not the float32 onset_sec gt_from_match took performance onsets from note_array()'s onset_sec, which partitura stores as float32. A MIDI follower stamps its updates with the float64 onset it receives, so the GT time of the same onset could fall a few tens of nanoseconds earlier, and the perf -> score lookup ("last position at or before t") then missed the update made at that onset and scored the previous event. In the eval fold about a third of MIDI onsets were scored one event late this way; 6/8 scores, where one event is up to a beat, failed the tracking criterion on paths that are correct. Use the PerformedPart notes' note_on (float64) instead: GT and follower times now coincide exactly. --- matchmaker/utils/eval.py | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/matchmaker/utils/eval.py b/matchmaker/utils/eval.py index 5e02c0a..57623fd 100644 --- a/matchmaker/utils/eval.py +++ b/matchmaker/utils/eval.py @@ -234,7 +234,14 @@ def gt_from_match(match_path, score_notes): ) for n in match_notes } - onset_at = {str(n["id"]): float(n["onset_sec"]) for n in perf.note_array()} + # note_on, not note_array()'s float32 onset_sec: a MIDI follower stamps its + # updates with the float64 onset, and a GT time rounded below it would hide + # the update made at that very onset from the perf -> score lookup. + onset_at = { + str(n["id"]): float(n["note_on"]) + for part in perf.performedparts + for n in part.notes + } beats, secs = [], [] for a in alignment: From d71f74e310403c66ce9e41a757617596653fb3b6 Mon Sep 17 00:00:00 2001 From: "jiyun.park" Date: Mon, 5 Oct 2026 14:55:57 +0200 Subject: [PATCH 7/9] Audio OPHMM: chord durations in quarter notes The bottom-level self-transitions took the chord durations from score beats times 60 / tempo. The tempo is quarter-note BPM, but score beats follow the time signature, so in 6/8 (eighth-note beats) every chord was expected to last twice as long, and in 2/2 half as long. Use onset_quarter; the reported score positions stay in beats. --- matchmaker/prob/outer_product_hmm_audio.py | 21 +++++++++++++++------ 1 file changed, 15 insertions(+), 6 deletions(-) diff --git a/matchmaker/prob/outer_product_hmm_audio.py b/matchmaker/prob/outer_product_hmm_audio.py index 12fffab..2e345c9 100644 --- a/matchmaker/prob/outer_product_hmm_audio.py +++ b/matchmaker/prob/outer_product_hmm_audio.py @@ -306,8 +306,16 @@ def __init__( } # Bottom transitions a_{l',l}^{(i)} and exit probs e_l^{(i)} (Eq.(5)) frame_rate = float(self.sample_rate) / float(self.hop_length) + # chord durations in quarter notes: `tempo` is quarter-note BPM, while + # score beats follow the time signature (an eighth in 6/8, a half in 2/2) + notes = ( + self.reference_features + if isinstance(self.reference_features, np.ndarray) + else self.reference_features.note_array() + ) + quarter_at = dict(zip(notes["onset_beat"].tolist(), notes["onset_quarter"].tolist())) self.a00 = self._compute_chord_self_transition_probs( - unique_onsets=unique_onsets, + onset_quarters=np.array([quarter_at[b] for b in unique_onsets.tolist()]), tempo=tempo, frame_rate=frame_rate, ) @@ -346,20 +354,21 @@ def _pause_self_transition_prob( @staticmethod def _compute_chord_self_transition_probs( - unique_onsets: np.ndarray, + onset_quarters: np.ndarray, tempo: float, frame_rate: float, ) -> np.ndarray: """ Compute self-transition probabilities from chord durations (Eq.5). - a_i = 1 - 1/d_i, where d_i = duration_sec / frame_time. + a_i = 1 - 1/d_i, where d_i = duration_sec / frame_time. ``onset_quarters`` + are the chord onsets in quarter notes and ``tempo`` is quarter-note BPM. """ - N = len(unique_onsets) + N = len(onset_quarters) frame_time = 1.0 / max(frame_rate, 1e-6) - # Convert onset beats to seconds, then compute inter-onset durations - onset_sec = unique_onsets * (60.0 / tempo) + # Convert onset quarters to seconds, then compute inter-onset durations + onset_sec = np.asarray(onset_quarters, dtype=float) * (60.0 / tempo) dur_sec = np.zeros(N, dtype=float) if N >= 2: dur_sec[:-1] = np.diff(onset_sec) From 77614b7ac6639370858642bb4d637789400642c4 Mon Sep 17 00:00:00 2001 From: "jiyun.park" Date: Wed, 7 Oct 2026 15:21:40 +0200 Subject: [PATCH 8/9] GT onset times from the streamed performance when it holds the match notes A .match file and the MIDI file a follower streams can store the same notes at different tick resolutions ((n)ASAP: 480 vs 384 ppq), so their onsets differ by up to half a tick. A GT time a fraction of a millisecond before the follower's update was scored against the position before that update. resolve_gt/gt_from_match take the streamed performance and use its onsets when every matched note has the same id and pitch in it. --- matchmaker/utils/eval.py | 51 ++++++++++++++++++++++++++++++++++--- tests/test_utils_eval_gt.py | 41 +++++++++++++++++++++++++++++ 2 files changed, 88 insertions(+), 4 deletions(-) create mode 100644 tests/test_utils_eval_gt.py diff --git a/matchmaker/utils/eval.py b/matchmaker/utils/eval.py index 57623fd..1c2e864 100644 --- a/matchmaker/utils/eval.py +++ b/matchmaker/utils/eval.py @@ -214,7 +214,39 @@ def evaluate_alignment( return {"beat": beat_results, "ms": ms_results} -def gt_from_match(match_path, score_notes): +def _streamed_onsets(performance, match_perf, performance_ids): + """Onset times of ``performance`` by note id, if it holds the match's notes. + + A .match file and the MIDI file a follower streams can store the same notes + with different tick resolutions (e.g. (n)ASAP: 480 vs 384 ppq), so their + times differ by up to half a tick. The follower stamps its updates with the + times it receives, so GT must use them: a GT time a fraction of a + millisecond early would be scored against the position before that update. + Returns None unless every matched note has the same id and pitch in both. + """ + if performance is None: + return None + if isinstance(performance, (str, Path)): + performance = pt.load_performance(str(performance)) + parts = getattr(performance, "performedparts", [performance]) + streamed = {str(n["id"]): n for part in parts for n in part.notes} + pitch_in_match = { + str(n["id"]): int(n["pitch"]) for part in match_perf.performedparts for n in part.notes + } + for pid in performance_ids: + note = streamed.get(pid) + if note is None or int(note["pitch"]) != pitch_in_match[pid]: + return None + return {pid: float(note["note_on"]) for pid, note in streamed.items()} + + +def gt_from_match(match_path, score_notes, performance=None): + """GT (perf_sec, score_beat) from a .match file. + + ``performance`` is the performance the follower receives (a MIDI path or a + partitura performance); when its notes are those of the match file, their + onset times are used instead of the match file's own. + """ perf, alignment, match_score = pt.load_match(str(match_path), create_score=True) match_notes = match_score.note_array() shared_ids = {str(n["id"]) for n in score_notes} & { @@ -242,6 +274,14 @@ def gt_from_match(match_path, score_notes): for part in perf.performedparts for n in part.notes } + matched_ids = [ + str(a["performance_id"]) + for a in alignment + if a.get("label") == "match" and str(a["performance_id"]) in onset_at + ] + streamed = _streamed_onsets(performance, perf, matched_ids) + if streamed is not None: + onset_at = streamed beats, secs = [], [] for a in alignment: @@ -261,14 +301,17 @@ def gt_from_match(match_path, score_notes): return secs[keep], beats[keep] -def resolve_gt(gt, score_notes): - """Return GT as (perf_sec, score_beat) arrays, from an ndarray, .match, or .tsv.""" +def resolve_gt(gt, score_notes, performance=None): + """Return GT as (perf_sec, score_beat) arrays, from an ndarray, .match, or .tsv. + + ``performance``: the performance the follower receives; see gt_from_match. + """ if isinstance(gt, np.ndarray): # gt array columns: [perf_sec, score_beat] arr = np.asarray(gt, dtype=float) return arr[:, 0], arr[:, 1] if Path(gt).suffix.lower() == ".match": - return gt_from_match(gt, score_notes) + return gt_from_match(gt, score_notes, performance) with open(gt) as f: header = f.readline().rstrip("\n").split("\t") data = np.loadtxt(gt, delimiter="\t", skiprows=1, ndmin=2) diff --git a/tests/test_utils_eval_gt.py b/tests/test_utils_eval_gt.py new file mode 100644 index 0000000..9983d1c --- /dev/null +++ b/tests/test_utils_eval_gt.py @@ -0,0 +1,41 @@ +import copy +import unittest + +import numpy as np +import partitura as pt + +from matchmaker.utils.eval import gt_from_match, resolve_gt + +MATCH = "matchmaker/assets/Bach-fugue_bwv_858.match" + + +class TestGTFromStreamedPerformance(unittest.TestCase): + """GT times come from the performance the follower streams when it holds the same notes.""" + + @classmethod + def setUpClass(cls): + cls.perf, _, score = pt.load_match(MATCH, create_score=True) + cls.score_notes = score.note_array() + cls.secs, cls.beats = gt_from_match(MATCH, cls.score_notes) + + def shifted(self, offset): + perf = copy.deepcopy(self.perf) + for note in perf.performedparts[0].notes: + note["note_on"] = float(note["note_on"]) + offset + return perf + + def test_streamed_onsets_replace_the_match_times(self): + secs, beats = resolve_gt(MATCH, self.score_notes, performance=self.shifted(4e-4)) + np.testing.assert_allclose(secs, self.secs + 4e-4) + np.testing.assert_array_equal(beats, self.beats) + + def test_other_notes_fall_back_to_the_match_times(self): + perf = self.shifted(4e-4) + perf.performedparts[0].notes[0]["pitch"] += 1 + secs, beats = gt_from_match(MATCH, self.score_notes, performance=perf) + np.testing.assert_array_equal(secs, self.secs) + np.testing.assert_array_equal(beats, self.beats) + + +if __name__ == "__main__": + unittest.main() From 8e2348859ea141e7ee1f777f5a06cf130be3598a Mon Sep 17 00:00:00 2001 From: "jiyun.park" Date: Wed, 7 Oct 2026 15:21:40 +0200 Subject: [PATCH 9/9] PF MIDI processor: stamp an observation with its first onset ParticleFilterMidiProcessor returned the middle of the polling window, which precedes the onsets in the window's second half and follows those in its first half. Use the first onset of the frame, as the other MIDI processors do. --- matchmaker/features/midi.py | 11 +++++++++-- 1 file changed, 9 insertions(+), 2 deletions(-) diff --git a/matchmaker/features/midi.py b/matchmaker/features/midi.py index d56d721..555b46e 100644 --- a/matchmaker/features/midi.py +++ b/matchmaker/features/midi.py @@ -525,13 +525,19 @@ def __call__( Returns ------- - (KorzeniowskiObservation, frame_time) + (KorzeniowskiObservation, time) + ``time`` is the first onset in the frame, or the frame time when + the frame has none. The frame time is the middle of the polling + window, which lies before the onsets in its second half and after + those in its first half; the other MIDI processors also stamp an + observation with its first note. """ data, f_time = frame # Reset onset list for this frame. self.current_onsets = [] + onset_time = None for msg, m_time in data: @@ -540,6 +546,7 @@ def __call__( # new onset in this frame self.current_onsets.append(msg.note) self.velocities[msg.note] = msg.velocity + onset_time = m_time if onset_time is None else min(onset_time, m_time) self.active_notes[msg.note] = ( msg.velocity, @@ -590,7 +597,7 @@ def __call__( loudness=self.compute_loudness(), ) - return observation, f_time + return observation, (f_time if onset_time is None else onset_time) def compute_loudness(self) -> float: """