cgcardona / muse public
show.py python
483 lines 15.8 KB
Raw
12901c5a Initial extraction from tellurstori/maestro cgcardona <gabriel@tellurstori.com> 4d ago
1 """muse show <commit> — music-aware commit inspection.
2
3 The musician's equivalent of ``git show``: displays metadata, snapshot
4 contents, and optional music-native views for any historical commit.
5
6 **Default output** (human-readable)::
7
8 commit a1b2c3d4e5f6...
9 Branch: main
10 Author: producer@stori.app
11 Date: 2026-02-27 17:30:00
12
13 Add bridge section with Rhodes keys
14
15 Snapshot: 3 files
16 beat.mid
17 keys.mid
18 bass.mid
19
20 **Flag summary:**
21
22 - ``--json`` — full commit metadata + snapshot manifest as JSON
23 - ``--diff`` — path-level diff vs parent commit (A/M/D markers)
24 - ``--midi`` — list MIDI files in the snapshot
25 - ``--audio-preview`` — generate and open audio preview of the snapshot (macOS)
26
27 **Commit resolution** (same strategy as ``muse arrange``):
28
29 1. If the ref looks like a hex string (4–64 chars) → prefix match in the DB.
30 2. Otherwise → treat as a branch name and read from ``.muse/refs/heads/``.
31 3. ``HEAD`` → read from current ``HEAD`` ref pointer.
32 """
33 from __future__ import annotations
34
35 import asyncio
36 import json
37 import logging
38 import pathlib
39 import subprocess
40 from typing import Optional
41
42 import typer
43 from sqlalchemy import select
44 from sqlalchemy.ext.asyncio import AsyncSession
45 from typing_extensions import TypedDict
46
47 from maestro.muse_cli._repo import require_repo
48 from maestro.muse_cli.db import open_session
49 from maestro.muse_cli.errors import ExitCode
50 from maestro.muse_cli.models import MuseCliCommit, MuseCliSnapshot
51
52 logger = logging.getLogger(__name__)
53
54 app = typer.Typer()
55
56 _HEX_CHARS = frozenset("0123456789abcdef")
57
58 # MIDI file extensions recognised by the show command
59 _MIDI_EXTENSIONS = frozenset({".mid", ".midi", ".smf"})
60
61
62 # ---------------------------------------------------------------------------
63 # Result types
64 # ---------------------------------------------------------------------------
65
66
67 class ShowCommitResult(TypedDict):
68 """Full commit metadata for ``muse show``.
69
70 Returned by ``_show_async`` and consumed by both the human-readable
71 renderer and the JSON serialiser. Includes the snapshot manifest so
72 callers can list files, MIDI files, and compute diffs without a second
73 DB round-trip.
74
75 Music-domain fields are surfaced at the top level for easy agent
76 consumption (sourced from ``commit_metadata`` in the DB). All are
77 ``None`` when the commit was created without the corresponding flag.
78 """
79
80 commit_id: str
81 branch: str
82 parent_commit_id: Optional[str]
83 parent2_commit_id: Optional[str]
84 message: str
85 author: str
86 committed_at: str
87 snapshot_id: str
88 snapshot_manifest: dict[str, str]
89 # Music-domain metadata (from commit_metadata JSON blob)
90 section: Optional[str]
91 track: Optional[str]
92 emotion: Optional[str]
93
94
95 class ShowDiffResult(TypedDict):
96 """Path-level diff of a commit vs its parent.
97
98 Produced by ``_diff_vs_parent`` and used by ``--diff`` rendering.
99 """
100
101 commit_id: str
102 parent_commit_id: Optional[str]
103 added: list[str]
104 modified: list[str]
105 removed: list[str]
106 total_changed: int
107
108
109 # ---------------------------------------------------------------------------
110 # Commit resolution helpers
111 # ---------------------------------------------------------------------------
112
113
114 def _looks_like_hex_prefix(s: str) -> bool:
115 """Return True if *s* is a 4–64 character lowercase hex string."""
116 lower = s.lower()
117 return 4 <= len(lower) <= 64 and all(c in _HEX_CHARS for c in lower)
118
119
120 async def _resolve_commit(
121 session: AsyncSession,
122 muse_dir: pathlib.Path,
123 ref: str,
124 ) -> MuseCliCommit:
125 """Resolve a commit reference to a ``MuseCliCommit`` row.
126
127 Resolution order:
128 1. ``HEAD`` (case-insensitive) → follow the HEAD ref file.
129 2. Hex prefix (4–64 chars) → prefix match against ``commit_id`` in DB.
130 3. Anything else → treat as a branch name and read the tip ref file.
131
132 Raises ``typer.Exit`` with ``USER_ERROR`` when the ref cannot be resolved.
133 """
134 if ref.upper() == "HEAD" or not _looks_like_hex_prefix(ref):
135 # Branch name or HEAD
136 if ref.upper() == "HEAD":
137 head_ref_text = (muse_dir / "HEAD").read_text().strip()
138 ref_path = muse_dir / pathlib.Path(head_ref_text)
139 else:
140 ref_path = muse_dir / "refs" / "heads" / ref
141
142 if not ref_path.exists():
143 typer.echo(f"❌ Reference '{ref}' not found.")
144 raise typer.Exit(code=ExitCode.USER_ERROR)
145
146 commit_id = ref_path.read_text().strip()
147 if not commit_id:
148 typer.echo(f"❌ Reference '{ref}' has no commits yet.")
149 raise typer.Exit(code=ExitCode.USER_ERROR)
150
151 commit = await session.get(MuseCliCommit, commit_id)
152 if commit is None:
153 typer.echo(f"❌ Commit {commit_id[:8]} not found in database.")
154 raise typer.Exit(code=ExitCode.USER_ERROR)
155 return commit
156
157 # Hex prefix: try exact match first, then startswith
158 exact = await session.get(MuseCliCommit, ref)
159 if exact is not None:
160 return exact
161
162 prefix = ref.lower()
163 result = await session.execute(
164 select(MuseCliCommit).where(MuseCliCommit.commit_id.startswith(prefix))
165 )
166 matches = list(result.scalars().all())
167
168 if not matches:
169 typer.echo(f"❌ No commit found matching '{prefix[:8]}'.")
170 raise typer.Exit(code=ExitCode.USER_ERROR)
171 if len(matches) > 1:
172 typer.echo(f"❌ Ambiguous prefix '{prefix[:8]}' matches {len(matches)} commits:")
173 for c in matches:
174 typer.echo(f" {c.commit_id[:8]} {c.message[:60]}")
175 raise typer.Exit(code=ExitCode.USER_ERROR)
176
177 return matches[0]
178
179
180 async def _load_snapshot(
181 session: AsyncSession, commit: MuseCliCommit
182 ) -> dict[str, str]:
183 """Load the snapshot manifest for *commit*.
184
185 Returns an empty dict when the snapshot is missing (shouldn't happen in a
186 consistent DB, but handled gracefully to avoid crashing the display path).
187 """
188 snapshot = await session.get(MuseCliSnapshot, commit.snapshot_id)
189 if snapshot is None:
190 logger.warning(
191 "⚠️ Snapshot %s for commit %s missing from DB",
192 commit.snapshot_id[:8],
193 commit.commit_id[:8],
194 )
195 return {}
196 return dict(snapshot.manifest)
197
198
199 # ---------------------------------------------------------------------------
200 # Core async logic — fully injectable for tests
201 # ---------------------------------------------------------------------------
202
203
204 async def _show_async(
205 *,
206 session: AsyncSession,
207 muse_dir: pathlib.Path,
208 ref: str,
209 ) -> ShowCommitResult:
210 """Load commit metadata and snapshot manifest for *ref*.
211
212 Used by the Typer command and directly by tests. All I/O goes through
213 *session* — no filesystem side-effects beyond reading ``.muse/`` refs.
214 """
215 commit = await _resolve_commit(session, muse_dir, ref)
216 manifest = await _load_snapshot(session, commit)
217
218 # Extract music-domain metadata from the extensible JSON blob.
219 raw_metadata: dict[str, object] = dict(commit.commit_metadata or {})
220
221 return ShowCommitResult(
222 commit_id=commit.commit_id,
223 branch=commit.branch,
224 parent_commit_id=commit.parent_commit_id,
225 parent2_commit_id=commit.parent2_commit_id,
226 message=commit.message,
227 author=commit.author,
228 committed_at=commit.committed_at.strftime("%Y-%m-%d %H:%M:%S"),
229 snapshot_id=commit.snapshot_id,
230 snapshot_manifest=manifest,
231 section=str(raw_metadata["section"]) if "section" in raw_metadata else None,
232 track=str(raw_metadata["track"]) if "track" in raw_metadata else None,
233 emotion=str(raw_metadata["emotion"]) if "emotion" in raw_metadata else None,
234 )
235
236
237 async def _diff_vs_parent_async(
238 *,
239 session: AsyncSession,
240 muse_dir: pathlib.Path,
241 ref: str,
242 ) -> ShowDiffResult:
243 """Compute the path-level diff of *ref* vs its parent commit.
244
245 For the root commit (no parent) every path in the snapshot is "added".
246 """
247 commit = await _resolve_commit(session, muse_dir, ref)
248 manifest = await _load_snapshot(session, commit)
249
250 parent_manifest: dict[str, str] = {}
251 if commit.parent_commit_id:
252 parent_commit = await session.get(MuseCliCommit, commit.parent_commit_id)
253 if parent_commit is not None:
254 parent_manifest = await _load_snapshot(session, parent_commit)
255 else:
256 logger.warning(
257 "⚠️ Parent %s not found; treating as empty",
258 commit.parent_commit_id[:8],
259 )
260
261 all_paths = sorted(set(manifest) | set(parent_manifest))
262 added: list[str] = []
263 modified: list[str] = []
264 removed: list[str] = []
265
266 for path in all_paths:
267 cur = manifest.get(path)
268 par = parent_manifest.get(path)
269 if par is None:
270 added.append(path)
271 elif cur is None:
272 removed.append(path)
273 elif cur != par:
274 modified.append(path)
275
276 return ShowDiffResult(
277 commit_id=commit.commit_id,
278 parent_commit_id=commit.parent_commit_id,
279 added=added,
280 modified=modified,
281 removed=removed,
282 total_changed=len(added) + len(modified) + len(removed),
283 )
284
285
286 def _midi_files_in_manifest(manifest: dict[str, str]) -> list[str]:
287 """Return the subset of manifest paths whose extension is a MIDI extension."""
288 return sorted(
289 path for path in manifest if pathlib.Path(path).suffix.lower() in _MIDI_EXTENSIONS
290 )
291
292
293 # ---------------------------------------------------------------------------
294 # Renderers
295 # ---------------------------------------------------------------------------
296
297
298 def _render_show(result: ShowCommitResult) -> None:
299 """Print commit metadata in ``git show`` style."""
300 typer.echo(f"commit {result['commit_id']}")
301 typer.echo(f"Branch: {result['branch']}")
302 if result["author"]:
303 typer.echo(f"Author: {result['author']}")
304 typer.echo(f"Date: {result['committed_at']}")
305 if result["parent_commit_id"]:
306 typer.echo(f"Parent: {result['parent_commit_id'][:8]}")
307 if result["parent2_commit_id"]:
308 typer.echo(f"Parent2: {result['parent2_commit_id'][:8]}")
309 # Music-domain metadata (only shown when present)
310 if result["section"]:
311 typer.echo(f"Section: {result['section']}")
312 if result["track"]:
313 typer.echo(f"Track: {result['track']}")
314 if result["emotion"]:
315 typer.echo(f"Emotion: {result['emotion']}")
316 typer.echo("")
317 typer.echo(f" {result['message']}")
318 typer.echo("")
319
320 manifest = result["snapshot_manifest"]
321 paths = sorted(manifest)
322 typer.echo(f"Snapshot: {len(paths)} file{'s' if len(paths) != 1 else ''}")
323 for p in paths:
324 typer.echo(f" {p}")
325
326
327 def _render_diff(diff: ShowDiffResult) -> None:
328 """Print path-level diff vs parent in ``git diff --name-status`` style."""
329 short = diff["commit_id"][:8]
330 parent_short = diff["parent_commit_id"][:8] if diff["parent_commit_id"] else "(root)"
331 typer.echo(f"diff {parent_short}..{short}")
332 typer.echo("")
333
334 for p in diff["added"]:
335 typer.echo(f"A {p}")
336 for p in diff["modified"]:
337 typer.echo(f"M {p}")
338 for p in diff["removed"]:
339 typer.echo(f"D {p}")
340
341 if diff["total_changed"] == 0:
342 typer.echo("(no changes vs parent)")
343 else:
344 typer.echo(f"\n{diff['total_changed']} path(s) changed")
345
346
347 def _render_midi(manifest: dict[str, str], commit_id: str) -> None:
348 """List MIDI files contained in the snapshot."""
349 midi_files = _midi_files_in_manifest(manifest)
350 short = commit_id[:8]
351 if not midi_files:
352 typer.echo(f"No MIDI files in snapshot {short}.")
353 return
354
355 typer.echo(f"MIDI files in snapshot {short} ({len(midi_files)}):")
356 for path in midi_files:
357 obj_id = manifest[path]
358 typer.echo(f" {path} ({obj_id[:8]})")
359
360
361 def _render_audio_preview(commit_id: str, root: pathlib.Path) -> None:
362 """Trigger an audio preview for the commit's snapshot (macOS, stub).
363
364 The full implementation would call the Storpheus render-preview pipeline
365 and stream the result to ``afplay``. This stub prints the resolved path
366 and launches ``afplay`` on any pre-rendered WAV file in the export cache,
367 falling back to a clear help message when nothing is cached.
368 """
369 short = commit_id[:8]
370 export_dir = root / ".muse" / "exports" / short
371 if not export_dir.exists():
372 typer.echo(
373 f"⚠️ No cached audio preview for commit {short}.\n"
374 f" Run: muse export {short} --wav to render first, then retry."
375 )
376 return
377
378 wav_files = sorted(export_dir.glob("*.wav"))
379 if not wav_files:
380 typer.echo(
381 f"⚠️ Export directory exists but contains no WAV files for {short}.\n"
382 f" Run: muse export {short} --wav to regenerate."
383 )
384 return
385
386 wav = wav_files[0]
387 typer.echo(f"▶ Playing {wav.name} (commit {short}) …")
388 try:
389 subprocess.run(["afplay", str(wav)], check=True)
390 except FileNotFoundError:
391 typer.echo("❌ afplay not found — audio preview requires macOS.")
392 except subprocess.CalledProcessError as exc:
393 typer.echo(f"❌ afplay exited with code {exc.returncode}.")
394
395
396 # ---------------------------------------------------------------------------
397 # Typer command
398 # ---------------------------------------------------------------------------
399
400
401 @app.callback(invoke_without_command=True)
402 def show(
403 ctx: typer.Context,
404 commit: Optional[str] = typer.Argument(
405 default=None,
406 help=(
407 "Commit ID (full or prefix), branch name, or HEAD. "
408 "Defaults to HEAD when omitted."
409 ),
410 metavar="COMMIT",
411 ),
412 as_json: bool = typer.Option(
413 False,
414 "--json",
415 help="Output complete commit metadata and snapshot manifest as JSON.",
416 ),
417 diff: bool = typer.Option(
418 False,
419 "--diff",
420 help="Show path-level diff vs parent commit (A/M/D markers).",
421 ),
422 midi: bool = typer.Option(
423 False,
424 "--midi",
425 help="List MIDI files contained in the commit snapshot.",
426 ),
427 audio_preview: bool = typer.Option(
428 False,
429 "--audio-preview",
430 help="Open cached audio preview for this snapshot (macOS). "
431 "Run `muse export <commit> --wav` first to render.",
432 ),
433 ) -> None:
434 """Inspect a commit: metadata, snapshot, diff, and music-native views.
435
436 Equivalent to ``git show`` — lets you inspect any historical creative
437 decision in the Muse VCS. The ``--midi`` and ``--audio-preview`` flags
438 make it music-native, allowing direct playback of historical snapshots.
439
440 Without flags, prints commit metadata and snapshot file list.
441 Flags can be combined: ``muse show abc1234 --diff --midi``.
442 """
443 if ctx.invoked_subcommand is not None:
444 return
445
446 ref = commit or "HEAD"
447 root = require_repo()
448 muse_dir = root / ".muse"
449
450 async def _run() -> None:
451 async with open_session() as session:
452 result = await _show_async(session=session, muse_dir=muse_dir, ref=ref)
453
454 if as_json:
455 typer.echo(json.dumps(dict(result), indent=2))
456 return
457
458 # Default metadata view (always shown unless --json)
459 _render_show(result)
460
461 if diff:
462 diff_result = await _diff_vs_parent_async(
463 session=session, muse_dir=muse_dir, ref=ref
464 )
465 typer.echo("")
466 _render_diff(diff_result)
467
468 if midi:
469 typer.echo("")
470 _render_midi(result["snapshot_manifest"], result["commit_id"])
471
472 if audio_preview:
473 typer.echo("")
474 _render_audio_preview(result["commit_id"], root)
475
476 try:
477 asyncio.run(_run())
478 except typer.Exit:
479 raise
480 except Exception as exc:
481 typer.echo(f"❌ muse show failed: {exc}")
482 logger.error("❌ muse show error: %s", exc, exc_info=True)
483 raise typer.Exit(code=ExitCode.INTERNAL_ERROR)