gabriel / muse public
note_log.py python
176 lines 6.0 KB
26a36470 feat(music): 9 new semantic commands — version control that understands music Gabriel Cardona <gabriel@tellurstori.com> 5d ago
1 """muse note-log — note-level commit history for a MIDI track.
2
3 Walks the commit history and shows exactly which notes were added and
4 removed in each commit that touched a specific MIDI track. Every change
5 is expressed in musical notation, not as a binary blob diff.
6
7 Usage::
8
9 muse note-log tracks/melody.mid
10 muse note-log tracks/melody.mid --from HEAD~10
11 muse note-log tracks/melody.mid --json
12
13 Output::
14
15 Note history: tracks/melody.mid
16 Commits analysed: 12
17
18 cb4afaed 2026-03-16 "Perf: vectorise melody" (3 changes)
19 + C4 vel=80 @beat=1.00 dur=1.00 ch 0
20 + E4 vel=75 @beat=2.00 dur=0.50 ch 0
21 - D4 vel=72 @beat=2.00 dur=0.50 ch 0 (removed)
22
23 1d2e3faa 2026-03-15 "Add bridge section" (4 changes)
24 + A4 vel=78 @beat=9.00 dur=1.00 ch 0
25 + B4 vel=75 @beat=10.00 dur=1.00 ch 0
26 ...
27 """
28 from __future__ import annotations
29
30 import json
31 import logging
32 import pathlib
33
34 import typer
35
36 from muse.core.errors import ExitCode
37 from muse.core.repo import require_repo
38 from muse.core.store import resolve_commit_ref
39 from muse.domain import DomainOp
40 from muse.plugins.music._query import (
41 NoteInfo,
42 load_track,
43 walk_commits_for_track,
44 )
45 from muse.plugins.music.midi_diff import NoteKey, _note_summary, extract_notes
46 from muse.core.object_store import read_object
47
48 logger = logging.getLogger(__name__)
49
50 app = typer.Typer()
51
52
53 def _read_repo_id(root: pathlib.Path) -> str:
54 return str(json.loads((root / ".muse" / "repo.json").read_text())["repo_id"])
55
56
57 def _read_branch(root: pathlib.Path) -> str:
58 head_ref = (root / ".muse" / "HEAD").read_text().strip()
59 return head_ref.removeprefix("refs/heads/").strip()
60
61
62 def _flat_ops(ops: list[DomainOp]) -> list[DomainOp]:
63 """Flatten PatchOp child_ops for the given track."""
64 result: list[DomainOp] = []
65 for op in ops:
66 if op["op"] == "patch":
67 result.extend(op["child_ops"])
68 else:
69 result.append(op)
70 return result
71
72
73 @app.callback(invoke_without_command=True)
74 def note_log(
75 ctx: typer.Context,
76 track: str = typer.Argument(..., metavar="TRACK", help="Workspace-relative path to a .mid file."),
77 from_ref: str | None = typer.Option(
78 None, "--from", metavar="REF",
79 help="Start walking from this commit (default: HEAD).",
80 ),
81 max_commits: int = typer.Option(
82 50, "--max", "-n", metavar="N",
83 help="Maximum number of commits to walk (default: 50).",
84 ),
85 as_json: bool = typer.Option(False, "--json", help="Emit results as JSON."),
86 ) -> None:
87 """Show the note-level commit history for a MIDI track.
88
89 ``muse note-log`` walks the commit history and, for each commit that
90 touched *TRACK*, shows exactly which notes were added and removed —
91 expressed in musical notation (pitch name, beat position, velocity,
92 duration), not as a binary diff.
93
94 This is the music-domain equivalent of ``muse symbol-log``: a
95 semantic history of a single artefact, at the level of individual notes.
96
97 Use ``--from`` to start at a different point in history. Use ``--json``
98 to pipe the output to an agent for further processing.
99 """
100 root = require_repo()
101 repo_id = _read_repo_id(root)
102 branch = _read_branch(root)
103
104 start_commit = resolve_commit_ref(root, repo_id, branch, from_ref)
105 if start_commit is None:
106 typer.echo(f"❌ Commit '{from_ref or 'HEAD'}' not found.", err=True)
107 raise typer.Exit(code=ExitCode.USER_ERROR)
108
109 commits_with_manifest = walk_commits_for_track(
110 root, start_commit.commit_id, track, max_commits=max_commits
111 )
112
113 # Collect events: (commit, note_summary, op_kind) per commit that touched the track.
114 EventEntry = tuple[str, str, str, str, str, list[tuple[str, str]]]
115 events: list[EventEntry] = []
116
117 for commit, manifest in commits_with_manifest:
118 if commit.structured_delta is None:
119 continue
120 # Find the PatchOp for this track.
121 track_ops: list[DomainOp] = []
122 for op in commit.structured_delta["ops"]:
123 if op["address"] == track:
124 if op["op"] == "patch":
125 track_ops.extend(op["child_ops"])
126 else:
127 # File-level insert/delete/replace — not note-level.
128 track_ops.append(op)
129
130 if not track_ops:
131 continue
132
133 note_changes: list[tuple[str, str]] = []
134 for op in track_ops:
135 if op["op"] == "insert":
136 note_changes.append(("+", op.get("content_summary", op["address"])))
137 elif op["op"] == "delete":
138 note_changes.append(("-", op.get("content_summary", op["address"])))
139
140 if note_changes:
141 date_str = commit.committed_at.strftime("%Y-%m-%d")
142 events.append((
143 commit.commit_id[:8],
144 date_str,
145 commit.message,
146 commit.author or "unknown",
147 commit.commit_id,
148 note_changes,
149 ))
150
151 if as_json:
152 out: list[dict[str, str | list[dict[str, str]]]] = []
153 for short_id, date, msg, author, full_id, changes in events:
154 out.append({
155 "commit_id": full_id,
156 "date": date,
157 "message": msg,
158 "author": author,
159 "changes": [{"op": op, "note": note} for op, note in changes],
160 })
161 typer.echo(json.dumps({"track": track, "events": out}, indent=2))
162 return
163
164 typer.echo(f"\nNote history: {track}")
165 typer.echo(f"Commits analysed: {len(commits_with_manifest)}")
166
167 if not events:
168 typer.echo("\n (no note-level changes found for this track)")
169 return
170
171 for short_id, date, msg, author, _full_id, changes in events:
172 typer.echo(f"\n{short_id} {date} \"{msg}\" ({len(changes)} change(s))")
173 for op_kind, note_summary in changes:
174 prefix = " +" if op_kind == "+" else " -"
175 suffix = " (removed)" if op_kind == "-" else ""
176 typer.echo(f"{prefix} {note_summary}{suffix}")