cgcardona / muse public
motif.py python
453 lines 15.2 KB
Raw
12901c5a Initial extraction from tellurstori/maestro cgcardona <gabriel@tellurstori.com> 4d ago
1 """muse motif — identify, track, and compare recurring melodic motifs.
2
3 A motif is a short melodic or rhythmic idea that reappears and transforms
4 throughout a composition. This command group surfaces motif-level analysis
5 over the Muse VCS commit history.
6
7 Subcommands
8 -----------
9
10 ``muse motif find [<commit>]``
11 Detect recurring melodic/rhythmic patterns in a commit (default: HEAD).
12
13 ``muse motif track <pattern>``
14 Search all commits for appearances of a specific motif. The pattern
15 is expressed as space-separated note names (``"C D E G"``) or MIDI
16 numbers (``"60 62 64 67"``). Transpositions and standard transformations
17 (inversion, retrograde, retrograde-inversion) are detected automatically.
18
19 ``muse motif diff <commit-a> <commit-b>``
20 Show how the dominant motif transformed between two commits.
21
22 ``muse motif list``
23 List all named motifs stored in ``.muse/motifs/``.
24
25 Flags on ``find``
26 -----------------
27 --min-length N Minimum motif length in notes (default: 3).
28 --section TEXT Scope to a named section/region.
29 --track TEXT Scope to a named MIDI track.
30 --json Machine-readable JSON output.
31
32 All subcommands support ``--json`` for agent consumption.
33
34 Output example (find, default)::
35
36 Recurring motifs — commit a1b2c3d4 (HEAD -> main)
37 ── stub mode: full MIDI analysis pending ──
38
39 # Fingerprint Contour Count
40 - ------------------- ---------------- -----
41 1 [+2, +2, -1, +2] ascending-step 3
42 2 [-2, -2, +1, -2] descending-step 2
43 3 [+4, -2, +3] arch 2
44
45 3 motifs found (min-length 3)
46 """
47 from __future__ import annotations
48
49 import asyncio
50 import json
51 import logging
52 import pathlib
53 from typing import Optional
54
55 import typer
56 from typing_extensions import Annotated
57
58 from maestro.muse_cli._repo import require_repo
59 from maestro.muse_cli.db import open_session
60 from maestro.muse_cli.errors import ExitCode
61 from maestro.services.muse_motif import (
62 MotifDiffResult,
63 MotifFindResult,
64 MotifListResult,
65 MotifOccurrence,
66 MotifTrackResult,
67 diff_motifs,
68 find_motifs,
69 list_motifs,
70 track_motif,
71 )
72
73 logger = logging.getLogger(__name__)
74
75 app = typer.Typer(no_args_is_help=True)
76
77 # ---------------------------------------------------------------------------
78 # Output formatters
79 # ---------------------------------------------------------------------------
80
81
82 def _format_find(result: MotifFindResult, *, as_json: bool) -> str:
83 """Render a find result as a human-readable table or JSON."""
84 if as_json:
85 payload = {
86 "commit": result.commit_id,
87 "branch": result.branch,
88 "min_length": result.min_length,
89 "total_found": result.total_found,
90 "source": result.source,
91 "motifs": [
92 {
93 "fingerprint": list(g.fingerprint),
94 "label": g.label,
95 "count": g.count,
96 "occurrences": [
97 {
98 "commit": o.commit_id,
99 "track": o.track,
100 "section": o.section,
101 "start_position": o.start_position,
102 "transformation": o.transformation.value,
103 "pitches": list(o.pitch_sequence),
104 }
105 for o in g.occurrences
106 ],
107 }
108 for g in result.motifs
109 ],
110 }
111 return json.dumps(payload, indent=2)
112
113 lines: list[str] = [
114 f"Recurring motifs — commit {result.commit_id} (HEAD -> {result.branch})",
115 ]
116 if result.source == "stub":
117 lines.append("── stub mode: full MIDI analysis pending ──")
118 lines.append("")
119 lines.append(f"{'#':<3} {'Fingerprint':<22} {'Contour':<18} {'Count':>5}")
120 lines.append(f"{'-':<3} {'-'*22} {'-'*18} {'-'*5}")
121 for idx, group in enumerate(result.motifs, start=1):
122 fp_str = "[" + ", ".join(f"{'+' if i >= 0 else ''}{i}" for i in group.fingerprint) + "]"
123 lines.append(
124 f"{idx:<3} {fp_str:<22} {group.label:<18} {group.count:>5}"
125 )
126 lines.append("")
127 lines.append(f"{result.total_found} motif(s) found (min-length {result.min_length})")
128 return "\n".join(lines)
129
130
131 def _format_track(result: MotifTrackResult, *, as_json: bool) -> str:
132 """Render a track result as a human-readable table or JSON."""
133 if as_json:
134 payload = {
135 "pattern": result.pattern,
136 "fingerprint": list(result.fingerprint),
137 "total_commits_scanned": result.total_commits_scanned,
138 "source": result.source,
139 "occurrences": [
140 {
141 "commit": o.commit_id,
142 "track": o.track,
143 "section": o.section,
144 "start_position": o.start_position,
145 "transformation": o.transformation.value,
146 "pitches": list(o.pitch_sequence),
147 }
148 for o in result.occurrences
149 ],
150 }
151 return json.dumps(payload, indent=2)
152
153 fp_str = "[" + ", ".join(f"{'+' if i >= 0 else ''}{i}" for i in result.fingerprint) + "]"
154 lines: list[str] = [
155 f"Tracking motif: {result.pattern!r}",
156 f"Fingerprint: {fp_str}",
157 f"Commits scanned: {result.total_commits_scanned}",
158 "",
159 ]
160 if result.source == "stub":
161 lines.append("── stub mode: full history scan pending ──")
162 lines.append("")
163 if not result.occurrences:
164 lines.append("No occurrences found.")
165 return "\n".join(lines)
166
167 lines.append(f"{'Commit':<10} {'Track':<12} {'Transform':<14} {'Position':>8}")
168 lines.append(f"{'-'*10} {'-'*12} {'-'*14} {'-'*8}")
169 for occ in result.occurrences:
170 lines.append(
171 f"{occ.commit_id:<10} {occ.track:<12} "
172 f"{occ.transformation.value:<14} {occ.start_position:>8}"
173 )
174 lines.append("")
175 lines.append(f"{len(result.occurrences)} occurrence(s) found.")
176 return "\n".join(lines)
177
178
179 def _format_diff(result: MotifDiffResult, *, as_json: bool) -> str:
180 """Render a diff result as human-readable text or JSON."""
181 if as_json:
182 payload = {
183 "transformation": result.transformation.value,
184 "description": result.description,
185 "source": result.source,
186 "commit_a": {
187 "commit": result.commit_a.commit_id,
188 "fingerprint": list(result.commit_a.fingerprint),
189 "label": result.commit_a.label,
190 "pitches": list(result.commit_a.pitch_sequence),
191 },
192 "commit_b": {
193 "commit": result.commit_b.commit_id,
194 "fingerprint": list(result.commit_b.fingerprint),
195 "label": result.commit_b.label,
196 "pitches": list(result.commit_b.pitch_sequence),
197 },
198 }
199 return json.dumps(payload, indent=2)
200
201 def _fp(intervals: tuple[int, ...]) -> str:
202 return "[" + ", ".join(f"{'+' if i >= 0 else ''}{i}" for i in intervals) + "]"
203
204 lines: list[str] = [
205 f"Motif diff: {result.commit_a.commit_id} → {result.commit_b.commit_id}",
206 "",
207 f" A ({result.commit_a.commit_id}): {_fp(result.commit_a.fingerprint)} [{result.commit_a.label}]",
208 f" B ({result.commit_b.commit_id}): {_fp(result.commit_b.fingerprint)} [{result.commit_b.label}]",
209 "",
210 f"Transformation: {result.transformation.value.upper()}",
211 f"{result.description}",
212 ]
213 if result.source == "stub":
214 lines.append("")
215 lines.append("── stub mode: full MIDI analysis pending ──")
216 return "\n".join(lines)
217
218
219 def _format_list(result: MotifListResult, *, as_json: bool) -> str:
220 """Render a list result as a human-readable table or JSON."""
221 if as_json:
222 payload = {
223 "source": result.source,
224 "motifs": [
225 {
226 "name": m.name,
227 "fingerprint": list(m.fingerprint),
228 "created_at": m.created_at,
229 "description": m.description,
230 }
231 for m in result.motifs
232 ],
233 }
234 return json.dumps(payload, indent=2)
235
236 if not result.motifs:
237 return "No named motifs saved. Use `muse motif find` to discover them."
238
239 lines: list[str] = ["Named motifs:", ""]
240 lines.append(f"{'Name':<20} {'Fingerprint':<22} {'Created':<24} Description")
241 lines.append(f"{'-'*20} {'-'*22} {'-'*24} {'-'*30}")
242 for m in result.motifs:
243 fp_str = "[" + ", ".join(f"{'+' if i >= 0 else ''}{i}" for i in m.fingerprint) + "]"
244 desc = (m.description or "")[:30]
245 lines.append(f"{m.name:<20} {fp_str:<22} {m.created_at:<24} {desc}")
246 return "\n".join(lines)
247
248
249 def _resolve_head(root: pathlib.Path) -> tuple[str, str]:
250 """Return (short_commit_id, branch) for the current HEAD.
251
252 Args:
253 root: Repository root (directory containing ``.muse/``).
254
255 Returns:
256 A ``(commit_id, branch)`` pair. ``commit_id`` is at most 8 chars;
257 ``branch`` is the branch name extracted from the HEAD ref.
258 """
259 muse_dir = root / ".muse"
260 head_ref = (muse_dir / "HEAD").read_text().strip()
261 branch = head_ref.rsplit("/", 1)[-1] if "/" in head_ref else head_ref
262 ref_path = muse_dir / pathlib.Path(head_ref)
263 head_sha = ref_path.read_text().strip() if ref_path.exists() else "0000000"
264 return head_sha[:8], branch
265
266
267 # ---------------------------------------------------------------------------
268 # Subcommand: find
269 # ---------------------------------------------------------------------------
270
271
272 @app.command(name="find")
273 def motif_find(
274 commit: Annotated[
275 Optional[str],
276 typer.Argument(
277 help="Commit SHA to analyse. Defaults to HEAD.",
278 show_default=False,
279 ),
280 ] = None,
281 min_length: Annotated[
282 int,
283 typer.Option(
284 "--min-length",
285 help="Minimum motif length in notes (default: 3).",
286 min=2,
287 ),
288 ] = 3,
289 track: Annotated[
290 Optional[str],
291 typer.Option(
292 "--track",
293 help="Restrict analysis to a named MIDI track.",
294 show_default=False,
295 ),
296 ] = None,
297 section: Annotated[
298 Optional[str],
299 typer.Option(
300 "--section",
301 help="Restrict analysis to a named section/region.",
302 show_default=False,
303 ),
304 ] = None,
305 as_json: Annotated[
306 bool,
307 typer.Option("--json", help="Emit machine-readable JSON output."),
308 ] = False,
309 ) -> None:
310 """Detect recurring melodic/rhythmic patterns in a commit (default: HEAD)."""
311 root = require_repo()
312
313 async def _run() -> None:
314 async with open_session() as session: # noqa: F841 — reserved for DB queries
315 commit_id, branch = _resolve_head(root)
316 resolved = commit or commit_id
317 result = await find_motifs(
318 commit_id=resolved,
319 branch=branch,
320 min_length=min_length,
321 track=track,
322 section=section,
323 as_json=as_json,
324 )
325 typer.echo(_format_find(result, as_json=as_json))
326
327 try:
328 asyncio.run(_run())
329 except typer.Exit:
330 raise
331 except Exception as exc:
332 typer.echo(f"❌ muse motif find failed: {exc}")
333 logger.error("❌ muse motif find error: %s", exc, exc_info=True)
334 raise typer.Exit(code=ExitCode.INTERNAL_ERROR)
335
336
337 # ---------------------------------------------------------------------------
338 # Subcommand: track
339 # ---------------------------------------------------------------------------
340
341
342 @app.command(name="track")
343 def motif_track(
344 pattern: Annotated[
345 str,
346 typer.Argument(
347 help=(
348 "Motif to track — space-separated note names (e.g. 'C D E G') "
349 "or MIDI numbers (e.g. '60 62 64 67')."
350 ),
351 ),
352 ],
353 as_json: Annotated[
354 bool,
355 typer.Option("--json", help="Emit machine-readable JSON output."),
356 ] = False,
357 ) -> None:
358 """Search all commits for appearances of a specific motif.
359
360 Detects the motif and its common transformations: transposition,
361 inversion, retrograde, and retrograde-inversion.
362 """
363 root = require_repo()
364
365 async def _run() -> None:
366 async with open_session() as session: # noqa: F841 — reserved for DB queries
367 muse_dir = root / ".muse"
368 commit_ids: list[str] = []
369 head_ref = (muse_dir / "HEAD").read_text().strip()
370 ref_path = muse_dir / pathlib.Path(head_ref)
371 if ref_path.exists():
372 commit_ids = [ref_path.read_text().strip()]
373
374 result = await track_motif(pattern=pattern, commit_ids=commit_ids)
375 typer.echo(_format_track(result, as_json=as_json))
376
377 try:
378 asyncio.run(_run())
379 except ValueError as exc:
380 typer.echo(f"❌ {exc}")
381 raise typer.Exit(code=ExitCode.USER_ERROR)
382 except typer.Exit:
383 raise
384 except Exception as exc:
385 typer.echo(f"❌ muse motif track failed: {exc}")
386 logger.error("❌ muse motif track error: %s", exc, exc_info=True)
387 raise typer.Exit(code=ExitCode.INTERNAL_ERROR)
388
389
390 # ---------------------------------------------------------------------------
391 # Subcommand: diff
392 # ---------------------------------------------------------------------------
393
394
395 @app.command(name="diff")
396 def motif_diff(
397 commit_a: Annotated[
398 str,
399 typer.Argument(help="First (earlier) commit SHA.", metavar="COMMIT-A"),
400 ],
401 commit_b: Annotated[
402 str,
403 typer.Argument(help="Second (later) commit SHA.", metavar="COMMIT-B"),
404 ],
405 as_json: Annotated[
406 bool,
407 typer.Option("--json", help="Emit machine-readable JSON output."),
408 ] = False,
409 ) -> None:
410 """Show how the dominant motif transformed between two commits."""
411 require_repo()
412
413 async def _run() -> None:
414 result = await diff_motifs(commit_a_id=commit_a, commit_b_id=commit_b)
415 typer.echo(_format_diff(result, as_json=as_json))
416
417 try:
418 asyncio.run(_run())
419 except typer.Exit:
420 raise
421 except Exception as exc:
422 typer.echo(f"❌ muse motif diff failed: {exc}")
423 logger.error("❌ muse motif diff error: %s", exc, exc_info=True)
424 raise typer.Exit(code=ExitCode.INTERNAL_ERROR)
425
426
427 # ---------------------------------------------------------------------------
428 # Subcommand: list
429 # ---------------------------------------------------------------------------
430
431
432 @app.command(name="list")
433 def motif_list(
434 as_json: Annotated[
435 bool,
436 typer.Option("--json", help="Emit machine-readable JSON output."),
437 ] = False,
438 ) -> None:
439 """List all named motifs stored in ``.muse/motifs/``."""
440 root = require_repo()
441
442 async def _run() -> None:
443 result = await list_motifs(muse_dir_path=str(root / ".muse"))
444 typer.echo(_format_list(result, as_json=as_json))
445
446 try:
447 asyncio.run(_run())
448 except typer.Exit:
449 raise
450 except Exception as exc:
451 typer.echo(f"❌ muse motif list failed: {exc}")
452 logger.error("❌ muse motif list error: %s", exc, exc_info=True)
453 raise typer.Exit(code=ExitCode.INTERNAL_ERROR)