_analysis.py
python
sha256:ae1eca169c5efe00a6415971ed0e7259f68df3e3ce23935c4b1b714c2df6a240
Merge branch 'dev' into main
Human
22 days ago
| 1 | """Pure musical analysis helpers for the Muse MIDI plugin. |
| 2 | |
| 3 | All functions accept ``list[NoteInfo]`` and return typed results. |
| 4 | No I/O, no store access — composable building blocks for semantic commands. |
| 5 | |
| 6 | Design rule: every public function returns a TypedDict or a list thereof. |
| 7 | No ``Any``, no bare collections, no untyped parameters. |
| 8 | """ |
| 9 | |
| 10 | import math |
| 11 | from collections import Counter |
| 12 | from typing import TypedDict |
| 13 | |
| 14 | from muse.plugins.midi._query import NoteInfo, _PITCH_CLASSES, detect_chord, notes_by_bar |
| 15 | |
| 16 | # --------------------------------------------------------------------------- |
| 17 | # Scale detection |
| 18 | # --------------------------------------------------------------------------- |
| 19 | |
| 20 | _SCALES: list[tuple[str, frozenset[int]]] = [ |
| 21 | ("major", frozenset({0, 2, 4, 5, 7, 9, 11})), |
| 22 | ("natural minor", frozenset({0, 2, 3, 5, 7, 8, 10})), |
| 23 | ("harmonic minor", frozenset({0, 2, 3, 5, 7, 8, 11})), |
| 24 | ("melodic minor", frozenset({0, 2, 3, 5, 7, 9, 11})), |
| 25 | ("dorian", frozenset({0, 2, 3, 5, 7, 9, 10})), |
| 26 | ("phrygian", frozenset({0, 1, 3, 5, 7, 8, 10})), |
| 27 | ("lydian", frozenset({0, 2, 4, 6, 7, 9, 11})), |
| 28 | ("mixolydian", frozenset({0, 2, 4, 5, 7, 9, 10})), |
| 29 | ("locrian", frozenset({0, 1, 3, 5, 6, 8, 10})), |
| 30 | ("major pentatonic", frozenset({0, 2, 4, 7, 9})), |
| 31 | ("minor pentatonic", frozenset({0, 3, 5, 7, 10})), |
| 32 | ("blues", frozenset({0, 3, 5, 6, 7, 10})), |
| 33 | ("whole tone", frozenset({0, 2, 4, 6, 8, 10})), |
| 34 | ("diminished", frozenset({0, 2, 3, 5, 6, 8, 9, 11})), |
| 35 | ("chromatic", frozenset(range(12))), |
| 36 | ] |
| 37 | |
| 38 | class ScaleMatch(TypedDict): |
| 39 | """Best-fit scale result.""" |
| 40 | |
| 41 | root: str |
| 42 | name: str |
| 43 | confidence: float |
| 44 | out_of_scale_notes: int |
| 45 | |
| 46 | def detect_scale(notes: list[NoteInfo]) -> list[ScaleMatch]: |
| 47 | """Return the top-5 scale matches sorted by confidence. |
| 48 | |
| 49 | Confidence is the fraction of note *weight* (note count) covered by |
| 50 | the scale's pitch classes. ``out_of_scale_notes`` is the number of |
| 51 | notes whose pitch class falls outside the scale. |
| 52 | """ |
| 53 | if not notes: |
| 54 | return [] |
| 55 | histogram = [0] * 12 |
| 56 | for n in notes: |
| 57 | histogram[n.pitch_class] += 1 |
| 58 | total = max(sum(histogram), 1) |
| 59 | |
| 60 | results: list[ScaleMatch] = [] |
| 61 | for root in range(12): |
| 62 | for scale_name, scale_pcs in _SCALES: |
| 63 | absolute_pcs = frozenset((root + pc) % 12 for pc in scale_pcs) |
| 64 | covered = sum(histogram[pc] for pc in absolute_pcs) |
| 65 | out_of_scale = sum(histogram[pc] for pc in range(12) if pc not in absolute_pcs) |
| 66 | results.append(ScaleMatch( |
| 67 | root=_PITCH_CLASSES[root], |
| 68 | name=scale_name, |
| 69 | confidence=round(covered / total, 3), |
| 70 | out_of_scale_notes=out_of_scale, |
| 71 | )) |
| 72 | |
| 73 | results.sort(key=lambda r: (-r["confidence"], r["out_of_scale_notes"])) |
| 74 | seen: set[tuple[str, str]] = set() |
| 75 | unique: list[ScaleMatch] = [] |
| 76 | for r in results: |
| 77 | key = (r["root"], r["name"]) |
| 78 | if key not in seen: |
| 79 | seen.add(key) |
| 80 | unique.append(r) |
| 81 | return unique[:5] |
| 82 | |
| 83 | # --------------------------------------------------------------------------- |
| 84 | # Rhythm analysis |
| 85 | # --------------------------------------------------------------------------- |
| 86 | |
| 87 | class RhythmAnalysis(TypedDict): |
| 88 | """Summary of rhythmic properties.""" |
| 89 | |
| 90 | total_notes: int |
| 91 | bars: int |
| 92 | notes_per_bar_avg: float |
| 93 | syncopation_score: float |
| 94 | quantization_score: float |
| 95 | swing_ratio: float |
| 96 | dominant_subdivision: str |
| 97 | |
| 98 | def analyze_rhythm(notes: list[NoteInfo]) -> RhythmAnalysis: |
| 99 | """Compute rhythmic metrics: syncopation, quantisation, swing. |
| 100 | |
| 101 | *syncopation_score*: fraction of notes landing on off-beat positions. |
| 102 | *quantization_score*: 1.0 = perfectly on the 16th-note grid, 0.0 = random. |
| 103 | *swing_ratio*: ratio of even/odd 8th-note IOI durations (1.0 = straight, |
| 104 | >1.3 = swung). |
| 105 | """ |
| 106 | if not notes: |
| 107 | return RhythmAnalysis( |
| 108 | total_notes=0, bars=0, notes_per_bar_avg=0.0, |
| 109 | syncopation_score=0.0, quantization_score=1.0, |
| 110 | swing_ratio=1.0, dominant_subdivision="quarter", |
| 111 | ) |
| 112 | |
| 113 | tpb = max(notes[0].ticks_per_beat, 1) |
| 114 | sorted_notes = sorted(notes, key=lambda n: n.start_tick) |
| 115 | bars = notes_by_bar(notes) |
| 116 | num_bars = max(bars.keys()) if bars else 1 |
| 117 | |
| 118 | # Syncopation: fraction on weak sub-beats |
| 119 | half_beat = tpb // 2 |
| 120 | synco_count = sum( |
| 121 | 1 for n in notes |
| 122 | if half_beat // 4 < n.start_tick % tpb < tpb - half_beat // 4 |
| 123 | ) |
| 124 | syncopation_score = synco_count / max(len(notes), 1) |
| 125 | |
| 126 | # Quantisation vs 16th-note grid |
| 127 | grid = max(tpb // 4, 1) |
| 128 | q_dists = [min(n.start_tick % grid, grid - n.start_tick % grid) / grid for n in notes] |
| 129 | quantization_score = 1.0 - (sum(q_dists) / max(len(q_dists), 1)) |
| 130 | |
| 131 | # Swing ratio between even/odd 8th-note slots |
| 132 | grid_8 = max(tpb // 2, 1) |
| 133 | even_durs: list[int] = [] |
| 134 | odd_durs: list[int] = [] |
| 135 | for n in sorted_notes: |
| 136 | if (n.start_tick // grid_8) % 2 == 0: |
| 137 | even_durs.append(n.duration_ticks) |
| 138 | else: |
| 139 | odd_durs.append(n.duration_ticks) |
| 140 | if even_durs and odd_durs: |
| 141 | swing_ratio = (sum(even_durs) / len(even_durs)) / max( |
| 142 | sum(odd_durs) / len(odd_durs), 1 |
| 143 | ) |
| 144 | else: |
| 145 | swing_ratio = 1.0 |
| 146 | |
| 147 | # Dominant note length |
| 148 | dur_buckets: Counter[str] = Counter() |
| 149 | for n in notes: |
| 150 | beats = n.duration_ticks / tpb |
| 151 | if beats >= 3.5: |
| 152 | dur_buckets["whole"] += 1 |
| 153 | elif beats >= 1.75: |
| 154 | dur_buckets["half"] += 1 |
| 155 | elif beats >= 0.875: |
| 156 | dur_buckets["quarter"] += 1 |
| 157 | elif beats >= 0.4: |
| 158 | dur_buckets["eighth"] += 1 |
| 159 | else: |
| 160 | dur_buckets["sixteenth"] += 1 |
| 161 | dominant_subdivision = dur_buckets.most_common(1)[0][0] if dur_buckets else "quarter" |
| 162 | |
| 163 | return RhythmAnalysis( |
| 164 | total_notes=len(notes), |
| 165 | bars=num_bars, |
| 166 | notes_per_bar_avg=round(len(notes) / max(num_bars, 1), 2), |
| 167 | syncopation_score=round(syncopation_score, 3), |
| 168 | quantization_score=round(quantization_score, 3), |
| 169 | swing_ratio=round(swing_ratio, 3), |
| 170 | dominant_subdivision=dominant_subdivision, |
| 171 | ) |
| 172 | |
| 173 | # --------------------------------------------------------------------------- |
| 174 | # Melodic contour |
| 175 | # --------------------------------------------------------------------------- |
| 176 | |
| 177 | class ContourAnalysis(TypedDict): |
| 178 | """Melodic contour shape and statistics.""" |
| 179 | |
| 180 | shape: str |
| 181 | intervals: list[int] |
| 182 | range_semitones: int |
| 183 | direction_changes: int |
| 184 | avg_interval_size: float |
| 185 | highest_pitch: str |
| 186 | lowest_pitch: str |
| 187 | |
| 188 | def analyze_contour(notes: list[NoteInfo]) -> ContourAnalysis: |
| 189 | """Analyse the melodic contour of a pitch sequence. |
| 190 | |
| 191 | *shape* is one of: ascending, descending, arch, valley, wave, flat. |
| 192 | *intervals* is the semitone interval sequence between consecutive notes. |
| 193 | """ |
| 194 | from muse.plugins.midi.midi_diff import _pitch_name |
| 195 | |
| 196 | sorted_notes = sorted(notes, key=lambda n: n.start_tick) |
| 197 | if len(sorted_notes) < 2: |
| 198 | pitch = sorted_notes[0].pitch if sorted_notes else 60 |
| 199 | pn = _pitch_name(pitch) |
| 200 | return ContourAnalysis( |
| 201 | shape="flat", intervals=[], range_semitones=0, |
| 202 | direction_changes=0, avg_interval_size=0.0, |
| 203 | highest_pitch=pn, lowest_pitch=pn, |
| 204 | ) |
| 205 | |
| 206 | intervals = [ |
| 207 | sorted_notes[i + 1].pitch - sorted_notes[i].pitch |
| 208 | for i in range(len(sorted_notes) - 1) |
| 209 | ] |
| 210 | pitches = [n.pitch for n in sorted_notes] |
| 211 | range_semitones = max(pitches) - min(pitches) |
| 212 | avg_interval_size = round(sum(abs(iv) for iv in intervals) / max(len(intervals), 1), 2) |
| 213 | |
| 214 | # Count direction changes |
| 215 | direction_changes = 0 |
| 216 | prev_dir = 0 |
| 217 | for iv in intervals: |
| 218 | cur_dir = 1 if iv > 0 else (-1 if iv < 0 else 0) |
| 219 | if cur_dir != 0 and prev_dir != 0 and cur_dir != prev_dir: |
| 220 | direction_changes += 1 |
| 221 | if cur_dir != 0: |
| 222 | prev_dir = cur_dir |
| 223 | |
| 224 | # Shape classification |
| 225 | n = len(pitches) |
| 226 | first_avg = sum(pitches[: n // 2]) / max(n // 2, 1) |
| 227 | second_avg = sum(pitches[n // 2 :]) / max(n - n // 2, 1) |
| 228 | mid_avg = sum(pitches[n // 4 : 3 * n // 4]) / max(n // 2, 1) |
| 229 | overall_avg = sum(pitches) / n |
| 230 | |
| 231 | if direction_changes == 0: |
| 232 | if second_avg > first_avg + 0.5: |
| 233 | shape = "ascending" |
| 234 | elif first_avg > second_avg + 0.5: |
| 235 | shape = "descending" |
| 236 | else: |
| 237 | shape = "flat" |
| 238 | elif direction_changes == 1: |
| 239 | # One direction change: arch (up-then-down) or valley (down-then-up) |
| 240 | if mid_avg > overall_avg + 0.5: |
| 241 | shape = "arch" |
| 242 | elif mid_avg < overall_avg - 0.5: |
| 243 | shape = "valley" |
| 244 | elif second_avg > first_avg + 0.5: |
| 245 | shape = "ascending" |
| 246 | elif first_avg > second_avg + 0.5: |
| 247 | shape = "descending" |
| 248 | else: |
| 249 | shape = "flat" |
| 250 | elif mid_avg > overall_avg + 0.5: |
| 251 | shape = "arch" |
| 252 | elif mid_avg < overall_avg - 0.5: |
| 253 | shape = "valley" |
| 254 | else: |
| 255 | shape = "wave" |
| 256 | |
| 257 | return ContourAnalysis( |
| 258 | shape=shape, |
| 259 | intervals=intervals[:32], |
| 260 | range_semitones=range_semitones, |
| 261 | direction_changes=direction_changes, |
| 262 | avg_interval_size=avg_interval_size, |
| 263 | highest_pitch=_pitch_name(max(pitches)), |
| 264 | lowest_pitch=_pitch_name(min(pitches)), |
| 265 | ) |
| 266 | |
| 267 | # --------------------------------------------------------------------------- |
| 268 | # Density |
| 269 | # --------------------------------------------------------------------------- |
| 270 | |
| 271 | class BarDensity(TypedDict): |
| 272 | """Note density for one bar.""" |
| 273 | |
| 274 | bar: int |
| 275 | note_count: int |
| 276 | notes_per_beat: float |
| 277 | |
| 278 | def analyze_density(notes: list[NoteInfo]) -> list[BarDensity]: |
| 279 | """Return note density (notes per beat, assuming 4/4) per bar.""" |
| 280 | bars = notes_by_bar(notes) |
| 281 | return [ |
| 282 | BarDensity( |
| 283 | bar=bar_num, |
| 284 | note_count=len(bar_notes), |
| 285 | notes_per_beat=round(len(bar_notes) / 4.0, 2), |
| 286 | ) |
| 287 | for bar_num, bar_notes in sorted(bars.items()) |
| 288 | ] |
| 289 | |
| 290 | # --------------------------------------------------------------------------- |
| 291 | # Harmonic tension |
| 292 | # --------------------------------------------------------------------------- |
| 293 | |
| 294 | # Dissonance weight per semitone interval class (0=unison, 1=m2, …, 11=M7) |
| 295 | _INTERVAL_DISSONANCE = [0, 10, 5, 3, 2, 1, 8, 0, 2, 3, 5, 8] |
| 296 | |
| 297 | class BarTension(TypedDict): |
| 298 | """Harmonic tension for one bar.""" |
| 299 | |
| 300 | bar: int |
| 301 | tension: float |
| 302 | label: str |
| 303 | |
| 304 | def compute_tension(notes: list[NoteInfo]) -> list[BarTension]: |
| 305 | """Compute harmonic tension per bar (0 = consonant, 1 = very dissonant).""" |
| 306 | bars = notes_by_bar(notes) |
| 307 | result: list[BarTension] = [] |
| 308 | for bar_num, bar_notes in sorted(bars.items()): |
| 309 | pitches = [n.pitch for n in bar_notes] |
| 310 | if len(pitches) < 2: |
| 311 | result.append(BarTension(bar=bar_num, tension=0.0, label="consonant")) |
| 312 | continue |
| 313 | intervals = [ |
| 314 | abs(pitches[i] - pitches[j]) % 12 |
| 315 | for i in range(len(pitches)) |
| 316 | for j in range(i + 1, len(pitches)) |
| 317 | ] |
| 318 | raw = sum(_INTERVAL_DISSONANCE[iv] for iv in intervals) / (len(intervals) * 10) |
| 319 | tension = round(min(1.0, raw), 3) |
| 320 | label = "consonant" if tension < 0.2 else "mild" if tension < 0.5 else "tense" |
| 321 | result.append(BarTension(bar=bar_num, tension=tension, label=label)) |
| 322 | return result |
| 323 | |
| 324 | # --------------------------------------------------------------------------- |
| 325 | # Cadence detection |
| 326 | # --------------------------------------------------------------------------- |
| 327 | |
| 328 | class Cadence(TypedDict): |
| 329 | """A detected cadence at a phrase boundary.""" |
| 330 | |
| 331 | bar: int |
| 332 | cadence_type: str |
| 333 | from_chord: str |
| 334 | to_chord: str |
| 335 | |
| 336 | def detect_cadences(notes: list[NoteInfo]) -> list[Cadence]: |
| 337 | """Detect cadences by examining chord motions at phrase boundaries (every 4 bars).""" |
| 338 | bars = notes_by_bar(notes) |
| 339 | bar_nums = sorted(bars.keys()) |
| 340 | if len(bar_nums) < 2: |
| 341 | return [] |
| 342 | |
| 343 | bar_chords: dict[int, str] = { |
| 344 | bn: detect_chord(frozenset(n.pitch_class for n in bar_notes)) |
| 345 | for bn, bar_notes in bars.items() |
| 346 | } |
| 347 | |
| 348 | cadences: list[Cadence] = [] |
| 349 | for i in range(len(bar_nums) - 1): |
| 350 | bn = bar_nums[i] |
| 351 | next_bn = bar_nums[i + 1] |
| 352 | # Only at phrase endings (bar before a multiple of 4 or 8) |
| 353 | if next_bn % 4 != 1 and next_bn % 8 != 1: |
| 354 | continue |
| 355 | from_chord = bar_chords.get(bn, "??") |
| 356 | to_chord = bar_chords.get(next_bn, "??") |
| 357 | cadence_type = _classify_cadence(from_chord, to_chord) |
| 358 | if cadence_type: |
| 359 | cadences.append(Cadence( |
| 360 | bar=next_bn, |
| 361 | cadence_type=cadence_type, |
| 362 | from_chord=from_chord, |
| 363 | to_chord=to_chord, |
| 364 | )) |
| 365 | return cadences |
| 366 | |
| 367 | def _classify_cadence(from_chord: str, to_chord: str) -> str | None: |
| 368 | """Heuristically classify a two-chord motion.""" |
| 369 | fc, tc = from_chord.lower(), to_chord.lower() |
| 370 | if "dom7" in fc and "maj" in tc: |
| 371 | return "authentic" |
| 372 | if "dom7" in fc and "min" in tc: |
| 373 | return "deceptive" |
| 374 | if ("maj" in fc or "min" in fc) and "dom7" in tc: |
| 375 | return "half" |
| 376 | if "min" in fc and "maj" in tc: |
| 377 | return "plagal" |
| 378 | return None |
| 379 | |
| 380 | # --------------------------------------------------------------------------- |
| 381 | # Motif detection |
| 382 | # --------------------------------------------------------------------------- |
| 383 | |
| 384 | class Motif(TypedDict): |
| 385 | """A recurring melodic interval pattern.""" |
| 386 | |
| 387 | id: int |
| 388 | interval_pattern: list[int] |
| 389 | occurrences: int |
| 390 | bars: list[int] |
| 391 | first_pitch: str |
| 392 | |
| 393 | def find_motifs( |
| 394 | notes: list[NoteInfo], |
| 395 | min_length: int = 3, |
| 396 | min_occurrences: int = 2, |
| 397 | ) -> list[Motif]: |
| 398 | """Find recurring melodic interval patterns (motifs) in the note sequence. |
| 399 | |
| 400 | Scans for repeated subsequences of semitone intervals. Returns up to 8 |
| 401 | non-overlapping patterns sorted by occurrence count. |
| 402 | """ |
| 403 | from muse.plugins.midi.midi_diff import _pitch_name |
| 404 | |
| 405 | sorted_notes = sorted(notes, key=lambda n: n.start_tick) |
| 406 | if len(sorted_notes) < min_length + 1: |
| 407 | return [] |
| 408 | |
| 409 | intervals = [ |
| 410 | sorted_notes[i + 1].pitch - sorted_notes[i].pitch |
| 411 | for i in range(len(sorted_notes) - 1) |
| 412 | ] |
| 413 | |
| 414 | pattern_count: Counter[tuple[int, ...]] = Counter() |
| 415 | for length in range(min_length, min(min_length + 4, len(intervals))): |
| 416 | for i in range(len(intervals) - length + 1): |
| 417 | pattern_count[tuple(intervals[i : i + length])] += 1 |
| 418 | |
| 419 | motifs: list[Motif] = [] |
| 420 | seen: set[tuple[int, ...]] = set() |
| 421 | motif_id = 0 |
| 422 | |
| 423 | for pat, count in pattern_count.most_common(20): |
| 424 | if count < min_occurrences or len(motifs) >= 8: |
| 425 | break |
| 426 | # Skip sub-patterns of already-found patterns |
| 427 | is_sub = any( |
| 428 | len(seen_pat) >= len(pat) and any( |
| 429 | seen_pat[k : k + len(pat)] == pat |
| 430 | for k in range(len(seen_pat) - len(pat) + 1) |
| 431 | ) |
| 432 | for seen_pat in seen |
| 433 | ) |
| 434 | if is_sub: |
| 435 | continue |
| 436 | seen.add(pat) |
| 437 | |
| 438 | bars_found: list[int] = [] |
| 439 | first_pitch = "" |
| 440 | for i in range(len(intervals) - len(pat) + 1): |
| 441 | if tuple(intervals[i : i + len(pat)]) == pat: |
| 442 | bars_found.append(sorted_notes[i].bar) |
| 443 | if not first_pitch: |
| 444 | first_pitch = _pitch_name(sorted_notes[i].pitch) |
| 445 | |
| 446 | motifs.append(Motif( |
| 447 | id=motif_id, |
| 448 | interval_pattern=list(pat), |
| 449 | occurrences=count, |
| 450 | bars=bars_found[:8], |
| 451 | first_pitch=first_pitch, |
| 452 | )) |
| 453 | motif_id += 1 |
| 454 | |
| 455 | return motifs |
| 456 | |
| 457 | # --------------------------------------------------------------------------- |
| 458 | # Voice-leading |
| 459 | # --------------------------------------------------------------------------- |
| 460 | |
| 461 | class VoiceLeadingIssue(TypedDict): |
| 462 | """A detected voice-leading problem.""" |
| 463 | |
| 464 | bar: int |
| 465 | issue_type: str |
| 466 | description: str |
| 467 | |
| 468 | def check_voice_leading(notes: list[NoteInfo]) -> list[VoiceLeadingIssue]: |
| 469 | """Detect parallel fifths/octaves and large leaps in the top voice.""" |
| 470 | bars = notes_by_bar(notes) |
| 471 | bar_nums = sorted(bars.keys()) |
| 472 | issues: list[VoiceLeadingIssue] = [] |
| 473 | |
| 474 | for i in range(len(bar_nums) - 1): |
| 475 | bn = bar_nums[i] |
| 476 | next_bn = bar_nums[i + 1] |
| 477 | cur = sorted(n.pitch for n in bars[bn]) |
| 478 | nxt = sorted(n.pitch for n in bars.get(next_bn, [])) |
| 479 | if len(cur) < 2 or len(nxt) < 2: |
| 480 | continue |
| 481 | |
| 482 | for vi in range(min(len(cur), len(nxt)) - 1): |
| 483 | ci = (cur[vi + 1] - cur[vi]) % 12 |
| 484 | ni = (nxt[vi + 1] - nxt[vi]) % 12 |
| 485 | if ci == 7 and ni == 7: |
| 486 | issues.append(VoiceLeadingIssue( |
| 487 | bar=next_bn, |
| 488 | issue_type="parallel_fifths", |
| 489 | description=f"voices {vi}–{vi+1}: parallel perfect fifths", |
| 490 | )) |
| 491 | if ci == 0 and ni == 0: |
| 492 | issues.append(VoiceLeadingIssue( |
| 493 | bar=next_bn, |
| 494 | issue_type="parallel_octaves", |
| 495 | description=f"voices {vi}–{vi+1}: parallel octaves", |
| 496 | )) |
| 497 | |
| 498 | if cur and nxt: |
| 499 | leap = abs(nxt[-1] - cur[-1]) |
| 500 | if leap > 9: |
| 501 | issues.append(VoiceLeadingIssue( |
| 502 | bar=next_bn, |
| 503 | issue_type="large_leap", |
| 504 | description=f"top voice: leap of {leap} semitones", |
| 505 | )) |
| 506 | |
| 507 | return issues |
| 508 | |
| 509 | # --------------------------------------------------------------------------- |
| 510 | # Tempo estimation |
| 511 | # --------------------------------------------------------------------------- |
| 512 | |
| 513 | class TempoEstimate(TypedDict): |
| 514 | """Estimated tempo from note onset spacing.""" |
| 515 | |
| 516 | estimated_bpm: float |
| 517 | ticks_per_beat: int |
| 518 | confidence: str |
| 519 | method: str |
| 520 | |
| 521 | def estimate_tempo(notes: list[NoteInfo]) -> TempoEstimate: |
| 522 | """Estimate BPM from inter-onset intervals. |
| 523 | |
| 524 | Uses the most common IOI (inter-onset interval) as the beat estimate. |
| 525 | Confidence is "high" when many notes agree on the same IOI. |
| 526 | """ |
| 527 | if not notes: |
| 528 | return TempoEstimate( |
| 529 | estimated_bpm=120.0, |
| 530 | ticks_per_beat=480, |
| 531 | confidence="none", |
| 532 | method="default", |
| 533 | ) |
| 534 | tpb = max(notes[0].ticks_per_beat, 1) |
| 535 | sorted_notes = sorted(notes, key=lambda n: n.start_tick) |
| 536 | iois = [ |
| 537 | sorted_notes[i + 1].start_tick - sorted_notes[i].start_tick |
| 538 | for i in range(len(sorted_notes) - 1) |
| 539 | if sorted_notes[i + 1].start_tick > sorted_notes[i].start_tick |
| 540 | ] |
| 541 | if not iois: |
| 542 | return TempoEstimate( |
| 543 | estimated_bpm=120.0, |
| 544 | ticks_per_beat=tpb, |
| 545 | confidence="none", |
| 546 | method="no_ioi", |
| 547 | ) |
| 548 | |
| 549 | # Snap IOIs to beat multiples and find most common beat length |
| 550 | beat_counts: Counter[int] = Counter() |
| 551 | for ioi in iois: |
| 552 | for div in [1, 2, 4]: |
| 553 | candidate = round(ioi / div / tpb) * tpb |
| 554 | if candidate > 0: |
| 555 | beat_counts[candidate] += 1 |
| 556 | |
| 557 | if not beat_counts: |
| 558 | return TempoEstimate( |
| 559 | estimated_bpm=120.0, ticks_per_beat=tpb, |
| 560 | confidence="low", method="ioi_fallback", |
| 561 | ) |
| 562 | |
| 563 | beat_ticks, vote_count = beat_counts.most_common(1)[0] |
| 564 | bpm = 60.0 * tpb / max(beat_ticks, 1) |
| 565 | confidence = "high" if vote_count >= len(iois) * 0.4 else "medium" if vote_count >= 3 else "low" |
| 566 | |
| 567 | return TempoEstimate( |
| 568 | estimated_bpm=round(bpm, 1), |
| 569 | ticks_per_beat=tpb, |
| 570 | confidence=confidence, |
| 571 | method="ioi_voting", |
| 572 | ) |
| 573 | |
| 574 | # --------------------------------------------------------------------------- |
| 575 | # Phrase-level similarity (for find-phrase) |
| 576 | # --------------------------------------------------------------------------- |
| 577 | |
| 578 | def pitch_interval_fingerprint(notes: list[NoteInfo]) -> tuple[int, ...]: |
| 579 | """Return the semitone-interval fingerprint of a sorted note sequence.""" |
| 580 | s = sorted(notes, key=lambda n: n.start_tick) |
| 581 | return tuple(s[i + 1].pitch - s[i].pitch for i in range(len(s) - 1)) |
| 582 | |
| 583 | def _cosine_similarity(a: list[float], b: list[float]) -> float: |
| 584 | if len(a) != len(b): |
| 585 | return 0.0 |
| 586 | dot = sum(x * y for x, y in zip(a, b)) |
| 587 | mag_a = math.sqrt(sum(x * x for x in a)) |
| 588 | mag_b = math.sqrt(sum(x * x for x in b)) |
| 589 | return dot / max(mag_a * mag_b, 1e-9) |
| 590 | |
| 591 | def phrase_similarity( |
| 592 | query_notes: list[NoteInfo], |
| 593 | candidate_notes: list[NoteInfo], |
| 594 | ) -> float: |
| 595 | """Return a similarity score [0, 1] between two note sequences. |
| 596 | |
| 597 | Uses interval fingerprints and pitch-class histogram cosine similarity. |
| 598 | """ |
| 599 | if not query_notes or not candidate_notes: |
| 600 | return 0.0 |
| 601 | |
| 602 | # Interval fingerprint similarity (rhythmically normalised) |
| 603 | q_fp = list(pitch_interval_fingerprint(query_notes)) |
| 604 | c_fp = list(pitch_interval_fingerprint(candidate_notes)) |
| 605 | if q_fp and c_fp: |
| 606 | min_len = min(len(q_fp), len(c_fp)) |
| 607 | interval_sim = _cosine_similarity( |
| 608 | [float(x) for x in q_fp[:min_len]], |
| 609 | [float(x) for x in c_fp[:min_len]], |
| 610 | ) |
| 611 | else: |
| 612 | interval_sim = 0.0 |
| 613 | |
| 614 | # Pitch-class histogram similarity |
| 615 | q_hist = [0.0] * 12 |
| 616 | c_hist = [0.0] * 12 |
| 617 | for n in query_notes: |
| 618 | q_hist[n.pitch_class] += 1.0 |
| 619 | for n in candidate_notes: |
| 620 | c_hist[n.pitch_class] += 1.0 |
| 621 | pc_sim = _cosine_similarity(q_hist, c_hist) |
| 622 | |
| 623 | return round(0.6 * interval_sim + 0.4 * pc_sim, 3) |
File History
6 commits
sha256:ae1eca169c5efe00a6415971ed0e7259f68df3e3ce23935c4b1b714c2df6a240
Merge branch 'dev' into main
Human
22 days ago
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6
chore(tests): add docstring to tests/__init__.py so rc14 tr…
Human
30 days ago
sha256:d8316ffae901be06347e16ab55be11868eb519dd16ade3e8aa16a99e662f7e62
baseline: rc14 re-baseline after rc3 store corruption recovery
Human
patch
30 days ago
sha256:50d413a635f57761c3fce1f70251b957b6423821525bf330747ef4007339222e
Merge branch 'dev' into main
Human
33 days ago
sha256:fb67fed5a4d3e40de84bdd163de94ef1386570bef1dd1a020a732c8a038962ce
Merge branch 'dev' into main
Human
52 days ago
sha256:1c4b3e3a9a1f300774c3ee662b572a698d5fd405bf765a71e6011a2e9c3eaaaa
feat: Muse — version control for the agent era
Human
104 days ago