cgcardona / muse public
muse_motif.py python
655 lines 21.7 KB
Raw
12901c5a Initial extraction from tellurstori/maestro cgcardona <gabriel@tellurstori.com> 4d ago
1 """Muse Motif Engine — identify, track, and compare recurring melodic motifs.
2
3 A *motif* is a short melodic or rhythmic idea — a sequence of pitches and/or
4 durations — that reappears and transforms throughout a composition. This
5 module implements the core analysis engine used by ``muse motif find``,
6 ``muse motif track``, and ``muse motif diff``.
7
8 Design
9 ------
10 - Melodic identity is encoded as **interval sequences** (signed semitone
11 differences between consecutive pitches) so that transpositions of the same
12 motif hash to the same fingerprint.
13 - Rhythmic identity is encoded as **relative duration ratios** normalised
14 against the shortest note in the sequence so that augmented / diminished
15 versions are detectable.
16 - A motif fingerprint is the concatenation of its interval sequence, allowing
17 fast set-based matching across commits.
18 - Transformation detection (inversion, retrograde, augmentation, diminution)
19 compares the found fingerprint against the query fingerprint's variants.
20
21 Boundary rules
22 --------------
23 - Must NOT import StateStore, executor, MCP tools, or route handlers.
24 - May import ``muse_cli.{db, models}``.
25 - Pure helpers (fingerprint computation, transformation detection) are
26 synchronous and fully testable without a DB.
27 """
28 from __future__ import annotations
29
30 import logging
31 from dataclasses import dataclass
32 from enum import Enum
33 from typing import Optional
34
35 logger = logging.getLogger(__name__)
36
37 # ---------------------------------------------------------------------------
38 # Primitive types
39 # ---------------------------------------------------------------------------
40
41 #: A melodic motif expressed as a sequence of semitone intervals.
42 #: e.g. [2, 2, -1, 2] for a 4-note motif with those ascending/descending steps.
43 IntervalSequence = tuple[int, ...]
44
45 #: A rhythmic motif expressed as relative duration ratios (float).
46 #: e.g. (1.0, 2.0, 1.0) means short–long–short relative durations.
47 RhythmSequence = tuple[float, ...]
48
49
50 # ---------------------------------------------------------------------------
51 # Transformation vocabulary
52 # ---------------------------------------------------------------------------
53
54
55 class MotifTransformation(str, Enum):
56 """Detected relationship between a found motif and the query motif.
57
58 - ``EXACT`` — identical interval sequence (possibly transposed).
59 - ``INVERSION`` — each interval negated (melodic mirror).
60 - ``RETROGRADE`` — interval sequence reversed.
61 - ``RETRO_INV`` — retrograde + inversion combined.
62 - ``AUGMENTED`` — same intervals; note durations scaled up.
63 - ``DIMINISHED`` — same intervals; note durations scaled down.
64 - ``APPROXIMATE`` — similar contour but not an exact variant.
65 """
66
67 EXACT = "exact"
68 INVERSION = "inversion"
69 RETROGRADE = "retrograde"
70 RETRO_INV = "retro_inv"
71 AUGMENTED = "augmented"
72 DIMINISHED = "diminished"
73 APPROXIMATE = "approximate"
74
75
76 # ---------------------------------------------------------------------------
77 # Named result types (public API contract)
78 # ---------------------------------------------------------------------------
79
80
81 @dataclass(frozen=True)
82 class MotifOccurrence:
83 """A single occurrence of a motif within a commit or pattern search.
84
85 Attributes:
86 commit_id: Short commit SHA where the motif was found.
87 track: Track name (e.g. ``"melody"``, ``"bass"``).
88 section: Named section the occurrence falls in (optional).
89 start_position: Position index of the first note of the motif.
90 transformation: Relationship to the query motif.
91 pitch_sequence: Literal pitch values at this occurrence (MIDI note numbers).
92 interval_fingerprint: Normalised interval sequence used for matching.
93 """
94
95 commit_id: str
96 track: str
97 section: Optional[str]
98 start_position: int
99 transformation: MotifTransformation
100 pitch_sequence: tuple[int, ...]
101 interval_fingerprint: IntervalSequence
102
103
104 @dataclass(frozen=True)
105 class MotifFindResult:
106 """Results from ``muse motif find`` — recurring patterns in a single commit.
107
108 Attributes:
109 commit_id: Short commit SHA analysed.
110 branch: Branch name.
111 min_length: Minimum motif length requested.
112 motifs: Detected recurring motif groups, sorted by occurrence count desc.
113 total_found: Total number of distinct recurring motifs identified.
114 source: ``"stub"`` until full MIDI analysis is wired; ``"live"`` thereafter.
115 """
116
117 commit_id: str
118 branch: str
119 min_length: int
120 motifs: tuple[MotifGroup, ...]
121 total_found: int
122 source: str
123
124
125 @dataclass(frozen=True)
126 class MotifGroup:
127 """A single recurring motif and all its occurrences in the scanned commit.
128
129 Attributes:
130 fingerprint: Normalised interval sequence (the motif's identity).
131 count: Number of times the motif appears.
132 occurrences: All detected occurrences.
133 label: Human-readable contour label (e.g. ``"ascending-step"``,
134 ``"arch"``, ``"descending-leap"``).
135 """
136
137 fingerprint: IntervalSequence
138 count: int
139 occurrences: tuple[MotifOccurrence, ...]
140 label: str
141
142
143 @dataclass(frozen=True)
144 class MotifTrackResult:
145 """Results from ``muse motif track`` — appearances of a pattern across history.
146
147 Attributes:
148 pattern: The query pattern (as a space-separated pitch string or
149 interval fingerprint).
150 fingerprint: Normalised interval sequence derived from the pattern.
151 occurrences: All commits where the motif (or a transformation) was found.
152 total_commits_scanned: How many commits were searched.
153 source: ``"stub"`` or ``"live"``.
154 """
155
156 pattern: str
157 fingerprint: IntervalSequence
158 occurrences: tuple[MotifOccurrence, ...]
159 total_commits_scanned: int
160 source: str
161
162
163 @dataclass(frozen=True)
164 class MotifDiffEntry:
165 """One side of a motif diff comparison.
166
167 Attributes:
168 commit_id: Short commit SHA.
169 fingerprint: Interval sequence at this commit.
170 label: Contour label.
171 pitch_sequence: Literal pitches (if available).
172 """
173
174 commit_id: str
175 fingerprint: IntervalSequence
176 label: str
177 pitch_sequence: tuple[int, ...]
178
179
180 @dataclass(frozen=True)
181 class MotifDiffResult:
182 """Results from ``muse motif diff`` — how a motif transformed between commits.
183
184 Attributes:
185 commit_a: Analysis of the motif at the first commit.
186 commit_b: Analysis of the motif at the second commit.
187 transformation: How the motif changed from commit A to commit B.
188 description: Human-readable description of the transformation.
189 source: ``"stub"`` or ``"live"``.
190 """
191
192 commit_a: MotifDiffEntry
193 commit_b: MotifDiffEntry
194 transformation: MotifTransformation
195 description: str
196 source: str
197
198
199 @dataclass(frozen=True)
200 class SavedMotif:
201 """A named motif stored in ``.muse/motifs/``.
202
203 Attributes:
204 name: User-assigned motif name (e.g. ``"main-theme"``).
205 fingerprint: Stored interval fingerprint.
206 created_at: ISO-8601 timestamp when the motif was named.
207 description: Optional free-text annotation.
208 """
209
210 name: str
211 fingerprint: IntervalSequence
212 created_at: str
213 description: Optional[str]
214
215
216 @dataclass(frozen=True)
217 class MotifListResult:
218 """Results from ``muse motif list`` — all named motifs in the repository.
219
220 Attributes:
221 motifs: All saved named motifs.
222 source: ``"stub"`` or ``"live"``.
223 """
224
225 motifs: tuple[SavedMotif, ...]
226 source: str
227
228
229 # ---------------------------------------------------------------------------
230 # Pure fingerprint helpers
231 # ---------------------------------------------------------------------------
232
233
234 def pitches_to_intervals(pitches: tuple[int, ...]) -> IntervalSequence:
235 """Convert a sequence of MIDI pitch numbers to a signed semitone interval sequence.
236
237 The interval representation is transposition-invariant — the same motif
238 at different pitch levels produces identical fingerprints.
239
240 Args:
241 pitches: Sequence of MIDI note numbers (0–127), length ≥ 2.
242
243 Returns:
244 Tuple of signed semitone differences, length = ``len(pitches) - 1``.
245 Returns an empty tuple for inputs shorter than 2 notes.
246 """
247 if len(pitches) < 2:
248 return ()
249 return tuple(pitches[i + 1] - pitches[i] for i in range(len(pitches) - 1))
250
251
252 def invert_intervals(intervals: IntervalSequence) -> IntervalSequence:
253 """Return the melodic inversion of an interval sequence (negate each step).
254
255 Inversion mirrors the motif around its starting pitch so that what went
256 up now goes down by the same interval.
257
258 Args:
259 intervals: Normalised interval sequence from :func:`pitches_to_intervals`.
260
261 Returns:
262 Interval sequence with each element negated.
263 """
264 return tuple(-i for i in intervals)
265
266
267 def retrograde_intervals(intervals: IntervalSequence) -> IntervalSequence:
268 """Return the retrograde (reverse) of an interval sequence.
269
270 Note: reversing the intervals gives the same pitches played backward,
271 which is not the same as reversing the pitch list directly.
272
273 Args:
274 intervals: Normalised interval sequence.
275
276 Returns:
277 Reversed interval sequence, with each element negated (retrograde
278 of pitches is the negation of reversed intervals).
279 """
280 return tuple(-i for i in reversed(intervals))
281
282
283 def detect_transformation(
284 query: IntervalSequence,
285 candidate: IntervalSequence,
286 ) -> Optional[MotifTransformation]:
287 """Determine the transformation relationship between a query and candidate motif.
288
289 Checks for exact match (transposition), inversion, retrograde, and the
290 combined retrograde-inversion.
291
292 Args:
293 query: The reference interval sequence.
294 candidate: The interval sequence to test.
295
296 Returns:
297 The :class:`MotifTransformation` if a relationship is detected, or
298 ``None`` if no recognised transformation applies.
299 """
300 if candidate == query:
301 return MotifTransformation.EXACT
302 if candidate == invert_intervals(query):
303 return MotifTransformation.INVERSION
304 if candidate == retrograde_intervals(query):
305 return MotifTransformation.RETROGRADE
306 if candidate == invert_intervals(retrograde_intervals(query)):
307 return MotifTransformation.RETRO_INV
308 return None
309
310
311 def contour_label(intervals: IntervalSequence) -> str:
312 """Assign a human-readable contour label to an interval sequence.
313
314 Labels encode the overall melodic direction and whether movement is
315 predominantly stepwise (≤2 semitones) or leap-based (>2 semitones).
316
317 Args:
318 intervals: Normalised interval sequence. Empty sequences return ``"static"``.
319
320 Returns:
321 One of: ``"ascending-step"``, ``"ascending-leap"``,
322 ``"descending-step"``, ``"descending-leap"``, ``"arch"``,
323 ``"valley"``, ``"oscillating"``, or ``"static"``.
324 """
325 if not intervals:
326 return "static"
327 net = sum(intervals)
328 max_step = max(abs(i) for i in intervals)
329 direction_changes = sum(
330 1
331 for j in range(len(intervals) - 1)
332 if (intervals[j] > 0) != (intervals[j + 1] > 0)
333 )
334 if direction_changes >= len(intervals) // 2:
335 return "oscillating"
336 ups = sum(1 for i in intervals if i > 0)
337 downs = sum(1 for i in intervals if i < 0)
338 if ups > 0 and downs > 0:
339 if ups > downs:
340 return "arch"
341 if downs > ups:
342 return "valley"
343 # Equal ups and downs: arch if motion starts upward, valley if downward.
344 return "arch" if intervals[0] > 0 else "valley"
345 stepwise = max_step <= 2
346 if net > 0:
347 return "ascending-step" if stepwise else "ascending-leap"
348 if net < 0:
349 return "descending-step" if stepwise else "descending-leap"
350 return "static"
351
352
353 def parse_pitch_string(pattern: str) -> tuple[int, ...]:
354 """Parse a space-separated pitch-name or MIDI-number string into pitch values.
355
356 Supports:
357 - MIDI integers: ``"60 62 64 67"``
358 - Note names: ``"C D E G"`` (middle octave assumed, sharps as ``C#``/``Cs``)
359
360 Args:
361 pattern: Space-separated pitch tokens.
362
363 Returns:
364 Tuple of MIDI note numbers (0–127).
365
366 Raises:
367 ValueError: If any token cannot be parsed as a MIDI number or note name.
368 """
369 _NOTE_MAP: dict[str, int] = {
370 "C": 60, "C#": 61, "CS": 61, "DB": 61,
371 "D": 62, "D#": 63, "DS": 63, "EB": 63,
372 "E": 64, "F": 65, "F#": 66, "FS": 66, "GB": 66,
373 "G": 67, "G#": 68, "GS": 68, "AB": 68,
374 "A": 69, "A#": 70, "AS": 70, "BB": 70,
375 "B": 71,
376 }
377 result: list[int] = []
378 for token in pattern.strip().split():
379 upper = token.upper().replace("-", "")
380 if upper in _NOTE_MAP:
381 result.append(_NOTE_MAP[upper])
382 else:
383 try:
384 midi = int(token)
385 if not 0 <= midi <= 127:
386 raise ValueError(f"MIDI pitch {midi} out of range [0, 127]")
387 result.append(midi)
388 except ValueError as exc:
389 raise ValueError(f"Cannot parse pitch token {token!r}") from exc
390 return tuple(result)
391
392
393 # ---------------------------------------------------------------------------
394 # Stub data helpers
395 # ---------------------------------------------------------------------------
396
397 _STUB_MOTIFS: list[tuple[IntervalSequence, str, int]] = [
398 ((2, 2, -1, 2), "ascending-step", 3),
399 ((-2, -2, 1, -2), "descending-step", 2),
400 ((4, -2, 3), "arch", 2),
401 ]
402
403
404 def _stub_motif_groups(
405 commit_id: str,
406 track: Optional[str],
407 min_length: int,
408 ) -> tuple[MotifGroup, ...]:
409 """Return placeholder MotifGroup entries for stub mode.
410
411 Args:
412 commit_id: Short commit SHA to embed in occurrences.
413 track: Track filter (if provided, used as the occurrence track name).
414 min_length: Minimum motif length filter (intervals of length ≥ min_length - 1).
415
416 Returns:
417 Tuple of :class:`MotifGroup` objects filtered to min_length.
418 """
419 groups: list[MotifGroup] = []
420 for fp, label, count in _STUB_MOTIFS:
421 if len(fp) + 1 < min_length:
422 continue
423 track_name = track or "melody"
424 pitches = _intervals_to_pitches(fp, start=60)
425 occurrence = MotifOccurrence(
426 commit_id=commit_id,
427 track=track_name,
428 section=None,
429 start_position=0,
430 transformation=MotifTransformation.EXACT,
431 pitch_sequence=pitches,
432 interval_fingerprint=fp,
433 )
434 groups.append(
435 MotifGroup(
436 fingerprint=fp,
437 count=count,
438 occurrences=(occurrence,) * count,
439 label=label,
440 )
441 )
442 return tuple(sorted(groups, key=lambda g: g.count, reverse=True))
443
444
445 def _intervals_to_pitches(
446 intervals: IntervalSequence,
447 start: int = 60,
448 ) -> tuple[int, ...]:
449 """Reconstruct a pitch sequence from an interval sequence starting at *start*.
450
451 Args:
452 intervals: Signed semitone intervals.
453 start: MIDI pitch of the first note (default: 60 = middle C).
454
455 Returns:
456 Tuple of MIDI pitch values.
457 """
458 pitches: list[int] = [start]
459 for step in intervals:
460 pitches.append(pitches[-1] + step)
461 return tuple(pitches)
462
463
464 # ---------------------------------------------------------------------------
465 # Public async API (stub implementations — contract-correct)
466 # ---------------------------------------------------------------------------
467
468
469 async def find_motifs(
470 *,
471 commit_id: str,
472 branch: str,
473 min_length: int = 3,
474 track: Optional[str] = None,
475 section: Optional[str] = None,
476 as_json: bool = False,
477 ) -> MotifFindResult:
478 """Detect recurring melodic/rhythmic patterns in a single commit.
479
480 Scans the MIDI data at *commit_id* for note sequences that appear more
481 than once within the commit, groups them by their transposition-invariant
482 fingerprint, and returns them sorted by occurrence count.
483
484 Args:
485 commit_id: Short or full commit SHA to analyse.
486 branch: Branch name (for context in the result).
487 min_length: Minimum motif length in notes (default: 3). Shorter
488 motifs tend to be musically trivial.
489 track: Restrict analysis to a single named track, or ``None`` for all.
490 section: Restrict to a named section/region, or ``None`` for all.
491 as_json: Unused here — rendered by the CLI layer.
492
493 Returns:
494 A :class:`MotifFindResult` with all detected recurring motifs.
495 """
496 logger.info("✅ muse motif find: commit=%s min_length=%d", commit_id[:8], min_length)
497 groups = _stub_motif_groups(commit_id[:8], track=track, min_length=min_length)
498 return MotifFindResult(
499 commit_id=commit_id[:8],
500 branch=branch,
501 min_length=min_length,
502 motifs=groups,
503 total_found=len(groups),
504 source="stub",
505 )
506
507
508 async def track_motif(
509 *,
510 pattern: str,
511 commit_ids: list[str],
512 ) -> MotifTrackResult:
513 """Search all commits for appearances of a specific motif pattern.
514
515 Parses *pattern* as a sequence of pitch names or MIDI numbers, derives the
516 transposition-invariant interval fingerprint, and scans each commit in
517 *commit_ids* for exact matches or recognised transformations (inversion,
518 retrograde, retrograde-inversion).
519
520 Args:
521 pattern: Space-separated pitch names (e.g. ``"C D E G"``) or MIDI
522 numbers (e.g. ``"60 62 64 67"``).
523 commit_ids: Ordered list of commit SHAs to scan (newest first).
524
525 Returns:
526 A :class:`MotifTrackResult` with all occurrences found.
527
528 Raises:
529 ValueError: If *pattern* cannot be parsed into a valid pitch sequence.
530 """
531 pitches = parse_pitch_string(pattern)
532 fingerprint = pitches_to_intervals(pitches)
533 logger.info(
534 "✅ muse motif track: pattern=%r fingerprint=%r commits=%d",
535 pattern,
536 fingerprint,
537 len(commit_ids),
538 )
539
540 occurrences: list[MotifOccurrence] = []
541 for cid in commit_ids:
542 short = cid[:8]
543 occ = MotifOccurrence(
544 commit_id=short,
545 track="melody",
546 section=None,
547 start_position=0,
548 transformation=MotifTransformation.EXACT,
549 pitch_sequence=pitches,
550 interval_fingerprint=fingerprint,
551 )
552 occurrences.append(occ)
553
554 return MotifTrackResult(
555 pattern=pattern,
556 fingerprint=fingerprint,
557 occurrences=tuple(occurrences),
558 total_commits_scanned=len(commit_ids),
559 source="stub",
560 )
561
562
563 async def diff_motifs(
564 *,
565 commit_a_id: str,
566 commit_b_id: str,
567 ) -> MotifDiffResult:
568 """Show how the dominant motif transformed between two commits.
569
570 Extracts the most prominent motif from each commit and computes the
571 transformation relationship between them (exact, inversion, retrograde, etc.).
572
573 Args:
574 commit_a_id: Short or full SHA of the first (earlier) commit.
575 commit_b_id: Short or full SHA of the second (later) commit.
576
577 Returns:
578 A :class:`MotifDiffResult` describing the transformation.
579 """
580 fp_a: IntervalSequence = (2, 2, -1, 2)
581 fp_b: IntervalSequence = (-2, -2, 1, -2)
582
583 transformation = detect_transformation(fp_a, fp_b) or MotifTransformation.APPROXIMATE
584
585 descriptions: dict[MotifTransformation, str] = {
586 MotifTransformation.EXACT: "The motif is transposition-equivalent — same shape, different pitch level.",
587 MotifTransformation.INVERSION: "The motif was inverted — ascending intervals became descending.",
588 MotifTransformation.RETROGRADE: "The motif was played in retrograde — same pitches reversed.",
589 MotifTransformation.RETRO_INV: "The motif was retrograde-inverted — reversed and mirrored.",
590 MotifTransformation.AUGMENTED: "The motif was augmented — note durations scaled up.",
591 MotifTransformation.DIMINISHED: "The motif was diminished — note durations compressed.",
592 MotifTransformation.APPROXIMATE: "The motif contour changed significantly between commits.",
593 }
594
595 entry_a = MotifDiffEntry(
596 commit_id=commit_a_id[:8],
597 fingerprint=fp_a,
598 label=contour_label(fp_a),
599 pitch_sequence=_intervals_to_pitches(fp_a),
600 )
601 entry_b = MotifDiffEntry(
602 commit_id=commit_b_id[:8],
603 fingerprint=fp_b,
604 label=contour_label(fp_b),
605 pitch_sequence=_intervals_to_pitches(fp_b),
606 )
607
608 logger.info(
609 "✅ muse motif diff: %s → %s, transformation=%s",
610 commit_a_id[:8],
611 commit_b_id[:8],
612 transformation.value,
613 )
614
615 return MotifDiffResult(
616 commit_a=entry_a,
617 commit_b=entry_b,
618 transformation=transformation,
619 description=descriptions[transformation],
620 source="stub",
621 )
622
623
624 async def list_motifs(
625 *,
626 muse_dir_path: str,
627 ) -> MotifListResult:
628 """List all named motifs stored in ``.muse/motifs/``.
629
630 Named motifs are user-annotated melodic ideas saved for future recall.
631 This command surfaces them in a structured format suitable for both
632 human review and agent consumption.
633
634 Args:
635 muse_dir_path: Absolute path to the ``.muse/`` directory.
636
637 Returns:
638 A :class:`MotifListResult` with all saved named motifs.
639 """
640 logger.info("✅ muse motif list: scanning %s", muse_dir_path)
641 stub_motifs: tuple[SavedMotif, ...] = (
642 SavedMotif(
643 name="main-theme",
644 fingerprint=(2, 2, -1, 2),
645 created_at="2026-01-15T10:30:00Z",
646 description="The central ascending motif introduced in the opening.",
647 ),
648 SavedMotif(
649 name="bass-riff",
650 fingerprint=(-2, -3, 2),
651 created_at="2026-01-20T14:15:00Z",
652 description="Chromatic bass figure used throughout the bridge.",
653 ),
654 )
655 return MotifListResult(motifs=stub_motifs, source="stub")