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