aaronrene / gabriel-muse public
manifest.py python
339 lines 11.0 KB
Raw
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6 chore(tests): add docstring to tests/__init__.py so rc14 tr… Human 32 days ago
1 """Hierarchical chunk manifests for the Muse MIDI plugin.
2
3 Evolves the flat ``{"files": {"track.mid": "<sha256>"}}`` snapshot beyond
4 a single content hash per file to a rich, per-bar, per-track manifest that
5 enables:
6
7 - **Partial diff** — compare only the bars that changed.
8 - **Query acceleration** — answer note queries without reading full MIDI blobs.
9 - **Targeted merge** — attempt to merge only the bars with conflicts.
10 - **Historical analytics** — aggregate statistics over commit history without
11 re-parsing MIDI bytes on every query.
12
13 Backward compatibility
14 ----------------------
15 The ``MusicManifest.files`` field is identical to the standard
16 ``SnapshotManifest.files`` so that the core Muse engine and all existing
17 commands continue to work unchanged. The ``tracks`` field is additive
18 metadata stored as a sidecar under ``.muse/music_manifests/`` — never
19 replacing the canonical flat manifest.
20
21 Storage layout::
22
23 .muse/music_manifests/
24 <snapshot_id>.json — full MusicManifest for this snapshot
25 (rebuildable from history; add to .museignore [domain.midi] in CI)
26
27 Public API
28 ----------
29 - :class:`BarChunk` — per-bar chunk descriptor.
30 - :class:`TrackManifest` — rich metadata for one MIDI track.
31 - :class:`MusicManifest` — top-level sidecar manifest.
32 - :func:`build_bar_chunk` — build a :class:`BarChunk` from a bar's notes.
33 - :func:`build_track_manifest` — build a :class:`TrackManifest`.
34 - :func:`build_music_manifest` — build the full :class:`MusicManifest`.
35 - :func:`write_music_manifest` — persist to ``.muse/music_manifests/``.
36 - :func:`read_music_manifest` — load from the sidecar store.
37 """
38
39 from __future__ import annotations
40
41 import hashlib
42 import json
43 import logging
44 import pathlib
45 from typing import Literal, TypedDict
46
47 from muse._version import __version__
48 from muse.core._types import Manifest
49 from muse.plugins.midi._query import (
50 NoteInfo,
51 detect_chord,
52 key_signature_guess,
53 notes_by_bar,
54 )
55 from muse.plugins.midi.midi_diff import extract_notes
56
57 logger = logging.getLogger(__name__)
58
59 type _BarMap = dict[str, "BarChunk"]
60 type _TrackMap = dict[str, "TrackManifest"]
61 type _ChangeMap = dict[str, list[int]]
62
63 _MANIFEST_DIR = ".muse/music_manifests"
64
65
66 # ---------------------------------------------------------------------------
67 # Types
68 # ---------------------------------------------------------------------------
69
70
71 class BarChunk(TypedDict):
72 """Descriptor for one bar's worth of note events in a MIDI track.
73
74 ``bar`` 1-indexed bar number (assumes 4/4 time).
75 ``chunk_hash`` SHA-256 of the canonical JSON of all notes in this bar.
76 Used for per-bar change detection without re-parsing MIDI.
77 ``note_count`` Number of notes in this bar.
78 ``chord`` Best-guess chord name for this bar (e.g. ``"Cmaj"``).
79 ``pitch_range`` ``[min_pitch, max_pitch]`` MIDI pitch values in this bar.
80 """
81
82 bar: int
83 chunk_hash: str
84 note_count: int
85 chord: str
86 pitch_range: list[int]
87
88
89 class TrackManifest(TypedDict):
90 """Rich metadata descriptor for one MIDI track at a specific snapshot.
91
92 ``track_id`` Stable identifier for this track (SHA-256 of file path).
93 Stable across renames if you track by content; changes
94 on rename. Use entity IDs in the entity index for
95 true cross-rename continuity.
96 ``file_path`` Workspace-relative MIDI file path.
97 ``content_hash`` SHA-256 of the full MIDI file bytes (same as the flat
98 manifest entry — the canonical content address).
99 ``bars`` Mapping from ``str(bar_number)`` → :class:`BarChunk`.
100 JSON keys are always strings; callers convert to int.
101 ``ticks_per_beat`` MIDI ticks per quarter note for this file.
102 ``note_count`` Total note count across all bars.
103 ``key_guess`` Krumhansl-Schmuckler key estimate (e.g. ``"G major"``).
104 ``bar_count`` Number of bars with at least one note.
105 """
106
107 track_id: str
108 file_path: str
109 content_hash: str
110 bars: _BarMap
111 ticks_per_beat: int
112 note_count: int
113 key_guess: str
114 bar_count: int
115
116
117 class MusicManifest(TypedDict):
118 """Top-level hierarchical manifest for a music snapshot.
119
120 This is the sidecar companion to the standard :class:`~muse.domain.SnapshotManifest`.
121 The ``files`` field is identical to the flat manifest — the core engine
122 reads only ``files`` for content addressing. The ``tracks`` field is
123 additive richness for music-domain queries, diff, and merge.
124
125 ``schema_version`` The Muse package version (read from ``muse._version``).
126 ``snapshot_id`` The snapshot this manifest belongs to.
127 ``files`` Standard flat ``{path: sha256}`` manifest (compat layer).
128 ``tracks`` ``{path: TrackManifest}`` for each MIDI file.
129 """
130
131 domain: Literal["midi"]
132 schema_version: str
133 snapshot_id: str
134 files: Manifest
135 tracks: _TrackMap
136
137
138 # ---------------------------------------------------------------------------
139 # Builders
140 # ---------------------------------------------------------------------------
141
142
143 def _bar_chunk_hash(notes: list[NoteInfo]) -> str:
144 """Return a SHA-256 of the canonical JSON of a bar's notes."""
145 payload = json.dumps(
146 [
147 {
148 "pitch": n.pitch,
149 "velocity": n.velocity,
150 "start_tick": n.start_tick,
151 "duration_ticks": n.duration_ticks,
152 "channel": n.channel,
153 }
154 for n in sorted(notes, key=lambda n: (n.start_tick, n.pitch))
155 ],
156 sort_keys=True,
157 separators=(",", ":"),
158 ).encode()
159 return hashlib.sha256(payload).hexdigest()
160
161
162 def build_bar_chunk(bar_num: int, notes: list[NoteInfo]) -> BarChunk:
163 """Build a :class:`BarChunk` descriptor for *bar_num*.
164
165 Args:
166 bar_num: 1-indexed bar number.
167 notes: All notes in this bar.
168
169 Returns:
170 A populated :class:`BarChunk`.
171 """
172 pcs = frozenset(n.pitch_class for n in notes)
173 chord = detect_chord(pcs)
174 pitches = [n.pitch for n in notes]
175 pitch_range: list[int] = [min(pitches), max(pitches)] if pitches else [0, 0]
176 return BarChunk(
177 bar=bar_num,
178 chunk_hash=_bar_chunk_hash(notes),
179 note_count=len(notes),
180 chord=chord,
181 pitch_range=pitch_range,
182 )
183
184
185 def build_track_manifest(
186 notes: list[NoteInfo],
187 file_path: str,
188 content_hash: str,
189 ticks_per_beat: int,
190 ) -> TrackManifest:
191 """Build a :class:`TrackManifest` from a parsed note list.
192
193 Args:
194 notes: All notes extracted from the MIDI file.
195 file_path: Workspace-relative MIDI file path.
196 content_hash: SHA-256 of the MIDI file bytes (from the flat manifest).
197 ticks_per_beat: MIDI timing resolution.
198
199 Returns:
200 A populated :class:`TrackManifest`.
201 """
202 track_id = hashlib.sha256(file_path.encode()).hexdigest()
203 bars_map = notes_by_bar(notes)
204 bars: _BarMap = {}
205 for bar_num, bar_notes in sorted(bars_map.items()):
206 bars[str(bar_num)] = build_bar_chunk(bar_num, bar_notes)
207
208 key_guess = key_signature_guess(notes)
209
210 return TrackManifest(
211 track_id=track_id,
212 file_path=file_path,
213 content_hash=content_hash,
214 bars=bars,
215 ticks_per_beat=ticks_per_beat,
216 note_count=len(notes),
217 key_guess=key_guess,
218 bar_count=len(bars),
219 )
220
221
222
223 # ---------------------------------------------------------------------------
224 # Persistence
225 # ---------------------------------------------------------------------------
226
227
228 def _manifest_path(repo_root: pathlib.Path, snapshot_id: str) -> pathlib.Path:
229 return repo_root / _MANIFEST_DIR / f"{snapshot_id}.json"
230
231
232 def write_music_manifest(
233 repo_root: pathlib.Path,
234 manifest: MusicManifest,
235 ) -> pathlib.Path:
236 """Persist *manifest* to ``.muse/music_manifests/<snapshot_id>.json``.
237
238 Args:
239 repo_root: Repository root.
240 manifest: The manifest to write.
241
242 Returns:
243 Path to the written file.
244 """
245 snapshot_id = manifest.get("snapshot_id", "")
246 if not snapshot_id:
247 raise ValueError("MusicManifest.snapshot_id must be non-empty")
248 path = _manifest_path(repo_root, snapshot_id)
249 path.parent.mkdir(parents=True, exist_ok=True)
250 path.write_text(json.dumps(manifest, indent=2) + "\n")
251 logger.debug(
252 "✅ Music manifest written: %s (%d tracks)",
253 snapshot_id[:8],
254 len(manifest["tracks"]),
255 )
256 return path
257
258
259 def read_music_manifest(
260 repo_root: pathlib.Path,
261 snapshot_id: str,
262 ) -> MusicManifest | None:
263 """Load the music manifest for *snapshot_id*, or ``None`` if absent.
264
265 Args:
266 repo_root: Repository root.
267 snapshot_id: Snapshot ID.
268
269 Returns:
270 The :class:`MusicManifest`, or ``None`` when the sidecar file does
271 not exist.
272 """
273 path = _manifest_path(repo_root, snapshot_id)
274 if not path.exists():
275 return None
276 try:
277 raw: MusicManifest = json.loads(path.read_text())
278 return raw
279 except (json.JSONDecodeError, KeyError) as exc:
280 logger.warning("⚠️ Corrupt music manifest %s: %s", path, exc)
281 return None
282
283
284 # ---------------------------------------------------------------------------
285 # Partial diff helper
286 # ---------------------------------------------------------------------------
287
288
289 def diff_manifests_by_bar(
290 base: MusicManifest,
291 target: MusicManifest,
292 ) -> _ChangeMap:
293 """Return a per-track list of bars that changed between two manifests.
294
295 Uses the per-bar ``chunk_hash`` values to detect changes without
296 loading any MIDI bytes.
297
298 Args:
299 base: Ancestor manifest.
300 target: Newer manifest.
301
302 Returns:
303 ``{track_path: [changed_bar_numbers]}`` for all tracks where at
304 least one bar differs. Tracks added or removed appear with ``[-1]``
305 as a sentinel indicating the whole track changed.
306 """
307 changed: _ChangeMap = {}
308
309 all_tracks = set(base["tracks"]) | set(target["tracks"])
310
311 for track in sorted(all_tracks):
312 base_track = base["tracks"].get(track)
313 target_track = target["tracks"].get(track)
314
315 if base_track is None or target_track is None:
316 changed[track] = [-1]
317 continue
318
319 if base_track["content_hash"] == target_track["content_hash"]:
320 continue
321
322 # Content changed — find which bars.
323 base_bars = base_track["bars"]
324 target_bars = target_track["bars"]
325 all_bar_keys = set(base_bars) | set(target_bars)
326
327 changed_bars: list[int] = []
328 for bar_key in sorted(all_bar_keys, key=lambda k: int(k)):
329 base_chunk = base_bars.get(bar_key)
330 target_chunk = target_bars.get(bar_key)
331 if base_chunk is None or target_chunk is None:
332 changed_bars.append(int(bar_key))
333 elif base_chunk["chunk_hash"] != target_chunk["chunk_hash"]:
334 changed_bars.append(int(bar_key))
335
336 if changed_bars:
337 changed[track] = sorted(changed_bars)
338
339 return changed
File History 2 commits
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6 chore(tests): add docstring to tests/__init__.py so rc14 tr… Human 32 days ago
sha256:d8316ffae901be06347e16ab55be11868eb519dd16ade3e8aa16a99e662f7e62 baseline: rc14 re-baseline after rc3 store corruption recovery Human patch 32 days ago