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/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( 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: """ 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/matchmaker/prob/outer_product_hmm_audio.py b/matchmaker/prob/outer_product_hmm_audio.py index f325e7c..2e345c9 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: @@ -310,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, ) @@ -350,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) @@ -511,12 +516,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,) 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/eval.py b/matchmaker/utils/eval.py index 5e02c0a..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} & { @@ -234,7 +266,22 @@ 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 + } + 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: @@ -254,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/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 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__": 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() 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()