_query.py
python
sha256:18b983389ee1b55900fcd799bfbb496552d2e3ecded9d18cefbfef188947a12e
chore: remove blob-debug test marker file
Sonnet 4.6
1 day ago
| 1 | """Shared music-domain query helpers for the Muse CLI. |
| 2 | |
| 3 | Provides the low-level primitives that music-domain commands share: |
| 4 | note extraction from the object store, bar-level grouping, chord detection, |
| 5 | and commit-graph walking specific to MIDI tracks. |
| 6 | |
| 7 | Nothing here belongs in the public ``MidiPlugin`` API. These are CLI-layer |
| 8 | helpers — thin adapters over ``midi_diff.extract_notes`` and the core store. |
| 9 | """ |
| 10 | |
| 11 | import logging |
| 12 | import pathlib |
| 13 | from typing import NamedTuple |
| 14 | |
| 15 | from muse.core.object_store import read_object |
| 16 | from muse.core.commits import ( |
| 17 | CommitRecord, |
| 18 | read_commit, |
| 19 | ) |
| 20 | from muse.core.snapshots import get_commit_snapshot_manifest |
| 21 | from muse.plugins.midi.midi_diff import NoteKey, _pitch_name, extract_notes # noqa: PLC2701 |
| 22 | from muse.core.types import Manifest, short_id |
| 23 | |
| 24 | logger = logging.getLogger(__name__) |
| 25 | |
| 26 | # --------------------------------------------------------------------------- |
| 27 | # Pitch / music-theory constants |
| 28 | # --------------------------------------------------------------------------- |
| 29 | |
| 30 | _PITCH_CLASSES = ["C", "C#", "D", "D#", "E", "F", "F#", "G", "G#", "A", "A#", "B"] |
| 31 | |
| 32 | # Chord templates: frozenset of pitch-class offsets (root = 0). |
| 33 | _CHORD_TEMPLATES: list[tuple[str, frozenset[int]]] = [ |
| 34 | ("maj", frozenset({0, 4, 7})), |
| 35 | ("min", frozenset({0, 3, 7})), |
| 36 | ("dim", frozenset({0, 3, 6})), |
| 37 | ("aug", frozenset({0, 4, 8})), |
| 38 | ("sus2", frozenset({0, 2, 7})), |
| 39 | ("sus4", frozenset({0, 5, 7})), |
| 40 | ("dom7", frozenset({0, 4, 7, 10})), |
| 41 | ("maj7", frozenset({0, 4, 7, 11})), |
| 42 | ("min7", frozenset({0, 3, 7, 10})), |
| 43 | ("dim7", frozenset({0, 3, 6, 9})), |
| 44 | ("5", frozenset({0, 7})), # power chord |
| 45 | ] |
| 46 | |
| 47 | # --------------------------------------------------------------------------- |
| 48 | # NoteInfo — enriched note for display |
| 49 | # --------------------------------------------------------------------------- |
| 50 | |
| 51 | class NoteInfo(NamedTuple): |
| 52 | """A ``NoteKey`` with derived musical fields for display.""" |
| 53 | |
| 54 | pitch: int |
| 55 | velocity: int |
| 56 | start_tick: int |
| 57 | duration_ticks: int |
| 58 | channel: int |
| 59 | ticks_per_beat: int |
| 60 | |
| 61 | @property |
| 62 | def pitch_name(self) -> str: |
| 63 | return _pitch_name(self.pitch) |
| 64 | |
| 65 | @property |
| 66 | def beat(self) -> float: |
| 67 | return self.start_tick / max(self.ticks_per_beat, 1) |
| 68 | |
| 69 | @property |
| 70 | def beat_duration(self) -> float: |
| 71 | return self.duration_ticks / max(self.ticks_per_beat, 1) |
| 72 | |
| 73 | @property |
| 74 | def bar(self) -> int: |
| 75 | """1-indexed bar number (assumes 4/4 time).""" |
| 76 | return int(self.start_tick // (4 * max(self.ticks_per_beat, 1))) + 1 |
| 77 | |
| 78 | @property |
| 79 | def beat_in_bar(self) -> float: |
| 80 | """Beat position within the bar (1-indexed).""" |
| 81 | tpb = max(self.ticks_per_beat, 1) |
| 82 | bar_tick = (self.bar - 1) * 4 * tpb |
| 83 | return (self.start_tick - bar_tick) / tpb + 1 |
| 84 | |
| 85 | @property |
| 86 | def pitch_class(self) -> int: |
| 87 | return self.pitch % 12 |
| 88 | |
| 89 | @property |
| 90 | def pitch_class_name(self) -> str: |
| 91 | return _PITCH_CLASSES[self.pitch_class] |
| 92 | |
| 93 | @classmethod |
| 94 | def from_note_key(cls, note: NoteKey, ticks_per_beat: int) -> "NoteInfo": |
| 95 | return cls( |
| 96 | pitch=note["pitch"], |
| 97 | velocity=note["velocity"], |
| 98 | start_tick=note["start_tick"], |
| 99 | duration_ticks=note["duration_ticks"], |
| 100 | channel=note["channel"], |
| 101 | ticks_per_beat=ticks_per_beat, |
| 102 | ) |
| 103 | |
| 104 | # --------------------------------------------------------------------------- |
| 105 | # Track loading from the object store |
| 106 | # --------------------------------------------------------------------------- |
| 107 | |
| 108 | def load_track( |
| 109 | root: pathlib.Path, |
| 110 | commit_id: str, |
| 111 | track_path: str, |
| 112 | ) -> tuple[list[NoteInfo], int] | None: |
| 113 | """Load notes for *track_path* from the snapshot at *commit_id*. |
| 114 | |
| 115 | Args: |
| 116 | root: Repository root. |
| 117 | commit_id: SHA-256 commit ID. |
| 118 | track_path: Workspace-relative path to the ``.mid`` file. |
| 119 | |
| 120 | Returns: |
| 121 | ``(notes, ticks_per_beat)`` on success, ``None`` when the track is |
| 122 | not in the snapshot or the object is missing / unparseable. |
| 123 | """ |
| 124 | manifest: Manifest = get_commit_snapshot_manifest(root, commit_id) or {} |
| 125 | object_id = manifest.get(track_path) |
| 126 | if object_id is None: |
| 127 | return None |
| 128 | raw = read_object(root, object_id) |
| 129 | if raw is None: |
| 130 | return None |
| 131 | try: |
| 132 | keys, tpb = extract_notes(raw) |
| 133 | except ValueError as exc: |
| 134 | logger.debug("Cannot parse MIDI %r from commit %s: %s", track_path, short_id(commit_id, strip=True), exc) |
| 135 | return None |
| 136 | notes = [NoteInfo.from_note_key(k, tpb) for k in keys] |
| 137 | return notes, tpb |
| 138 | |
| 139 | def load_track_from_workdir( |
| 140 | root: pathlib.Path, |
| 141 | track_path: str, |
| 142 | ) -> tuple[list[NoteInfo], int] | None: |
| 143 | """Load notes for *track_path* from ``state/`` (live working tree). |
| 144 | |
| 145 | Args: |
| 146 | root: Repository root. |
| 147 | track_path: Workspace-relative path to the ``.mid`` file. |
| 148 | |
| 149 | Returns: |
| 150 | ``(notes, ticks_per_beat)`` on success, ``None`` when unreadable. |
| 151 | """ |
| 152 | work_path = root / track_path |
| 153 | if not work_path.exists(): |
| 154 | return None |
| 155 | raw = work_path.read_bytes() |
| 156 | try: |
| 157 | keys, tpb = extract_notes(raw) |
| 158 | except ValueError as exc: |
| 159 | logger.debug("Cannot parse MIDI %r from workdir: %s", track_path, exc) |
| 160 | return None |
| 161 | notes = [NoteInfo.from_note_key(k, tpb) for k in keys] |
| 162 | return notes, tpb |
| 163 | |
| 164 | # --------------------------------------------------------------------------- |
| 165 | # Musical analysis helpers |
| 166 | # --------------------------------------------------------------------------- |
| 167 | |
| 168 | def notes_by_bar(notes: list[NoteInfo]) -> dict[int, list[NoteInfo]]: |
| 169 | """Group *notes* by 1-indexed bar number (assumes 4/4 time).""" |
| 170 | bars: dict[int, list[NoteInfo]] = {} |
| 171 | for note in sorted(notes, key=lambda n: (n.start_tick, n.pitch)): |
| 172 | bars.setdefault(note.bar, []).append(note) |
| 173 | return bars |
| 174 | |
| 175 | def detect_chord(pitch_classes: frozenset[int]) -> str: |
| 176 | """Return the best chord name for a set of pitch classes. |
| 177 | |
| 178 | Tries every chromatic root and every chord template. Returns the |
| 179 | name of the best match (most pitch classes covered) as ``"RootQuality"`` |
| 180 | e.g. ``"Cmaj"``, ``"Fmin7"``. Returns ``"??"`` when fewer than two |
| 181 | distinct pitch classes are present. |
| 182 | """ |
| 183 | if len(pitch_classes) < 2: |
| 184 | return "??" |
| 185 | best_name = "??" |
| 186 | best_score = 0 |
| 187 | for root in range(12): |
| 188 | normalized = frozenset((pc - root) % 12 for pc in pitch_classes) |
| 189 | for quality, template in _CHORD_TEMPLATES: |
| 190 | overlap = len(normalized & template) |
| 191 | if overlap > best_score or ( |
| 192 | overlap == best_score and overlap == len(template) |
| 193 | ): |
| 194 | best_score = overlap |
| 195 | root_name = _PITCH_CLASSES[root] |
| 196 | best_name = f"{root_name}{quality}" |
| 197 | return best_name |
| 198 | |
| 199 | def key_signature_guess(notes: list[NoteInfo]) -> str: |
| 200 | """Guess the key signature from pitch class frequencies. |
| 201 | |
| 202 | Uses the Krumhansl-Schmuckler key-finding algorithm with simplified |
| 203 | major and minor profiles. Returns a string like ``"G major"`` or |
| 204 | ``"D minor"``. |
| 205 | """ |
| 206 | if not notes: |
| 207 | return "unknown" |
| 208 | |
| 209 | # Build pitch class histogram. |
| 210 | histogram = [0] * 12 |
| 211 | for note in notes: |
| 212 | histogram[note.pitch_class] += 1 |
| 213 | |
| 214 | # Krumhansl-Schmuckler major and minor profiles (normalized). |
| 215 | major_profile = [ |
| 216 | 6.35, 2.23, 3.48, 2.33, 4.38, 4.09, |
| 217 | 2.52, 5.19, 2.39, 3.66, 2.29, 2.88, |
| 218 | ] |
| 219 | minor_profile = [ |
| 220 | 6.33, 2.68, 3.52, 5.38, 2.60, 3.53, |
| 221 | 2.54, 4.75, 3.98, 2.69, 3.34, 3.17, |
| 222 | ] |
| 223 | |
| 224 | total = max(sum(histogram), 1) |
| 225 | h_norm = [v / total for v in histogram] |
| 226 | |
| 227 | best_key = "" |
| 228 | best_score = -999.0 |
| 229 | |
| 230 | for root in range(12): |
| 231 | for mode, profile in [("major", major_profile), ("minor", minor_profile)]: |
| 232 | # Rotate profile to this root. |
| 233 | score = sum( |
| 234 | h_norm[(root + i) % 12] * profile[i] for i in range(12) |
| 235 | ) |
| 236 | if score > best_score: |
| 237 | best_score = score |
| 238 | best_key = f"{_PITCH_CLASSES[root]} {mode}" |
| 239 | |
| 240 | return best_key |
| 241 | |
| 242 | # --------------------------------------------------------------------------- |
| 243 | # Commit-graph walking (music-domain specific) |
| 244 | # --------------------------------------------------------------------------- |
| 245 | |
| 246 | def walk_commits_for_track( |
| 247 | root: pathlib.Path, |
| 248 | start_commit_id: str, |
| 249 | track_path: str, |
| 250 | max_commits: int = 10_000, |
| 251 | ) -> list[tuple[CommitRecord, dict[str, str] | None]]: |
| 252 | """Walk the parent chain from *start_commit_id*, collecting snapshot manifests. |
| 253 | |
| 254 | Returns ``(commit, manifest)`` pairs where ``manifest`` may be ``None`` |
| 255 | when the commit has no snapshot. Only commits where the track appears |
| 256 | in the manifest (or in its parent's manifest) are useful for note-level |
| 257 | queries, but we return all so callers can filter. |
| 258 | """ |
| 259 | from muse.core.graph import iter_ancestors |
| 260 | result: list[tuple[CommitRecord, dict[str, str] | None]] = [] |
| 261 | for commit in iter_ancestors(root, start_commit_id, first_parent_only=True, max_commits=max_commits): |
| 262 | manifest = get_commit_snapshot_manifest(root, commit.commit_id) or None |
| 263 | result.append((commit, manifest)) |
| 264 | return result |
| 265 | |
| 266 | # --------------------------------------------------------------------------- |
| 267 | # MIDI reconstruction helper (for transpose / mix) |
| 268 | # --------------------------------------------------------------------------- |
| 269 | |
| 270 | def notes_to_midi_bytes(notes: list[NoteInfo], ticks_per_beat: int) -> bytes: |
| 271 | """Reconstruct a MIDI file from a list of ``NoteInfo`` objects. |
| 272 | |
| 273 | Produces a Type-0 single-track MIDI file with one note_on / note_off |
| 274 | pair per note. Delegates to |
| 275 | :func:`~muse.plugins.midi.midi_diff.reconstruct_midi`. |
| 276 | """ |
| 277 | from muse.plugins.midi.midi_diff import NoteKey, reconstruct_midi |
| 278 | |
| 279 | keys = [ |
| 280 | NoteKey( |
| 281 | pitch=n.pitch, |
| 282 | velocity=n.velocity, |
| 283 | start_tick=n.start_tick, |
| 284 | duration_ticks=n.duration_ticks, |
| 285 | channel=n.channel, |
| 286 | ) |
| 287 | for n in notes |
| 288 | ] |
| 289 | return reconstruct_midi(keys, ticks_per_beat=ticks_per_beat) |
File History
7 commits
sha256:18b983389ee1b55900fcd799bfbb496552d2e3ecded9d18cefbfef188947a12e
chore: remove blob-debug test marker file
Sonnet 4.6
1 day ago
sha256:e452ad9a6ace6ccc6d875a35e06caf9da5576a970c1c36133b69a891ce5fefa8
chore: prebuild timing test
Sonnet 4.6
8 days ago
sha256:0008ab6695e3e064b3e236b24fd19e538fef6a588eb0d211622f4466d919c0b1
merge: pull staging/dev — advance to 0.2.0rc12
Sonnet 4.6
patch
10 days ago
sha256:9c33d61749fff814c5226d5386aa2af7064c2c02788594a25fdd709358132eea
fix: _PROPOSAL_PREFIX_RESOLVE_LIMIT 200 → 100 to match hub …
Sonnet 4.6
21 days ago
sha256:36c3cb3e76619d4c30a6d9bf81b5ec4ff148e30dcfed913e3114ca7b43b81c7e
fix: rename objects→blobs in push client and all stale test…
Sonnet 4.6
patch
24 days ago
sha256:c06a9b9b9fee26c68ea725b44d54b2c0a171301ce9de746d5b656617b4463a9a
fix: repair four test failures from post-migration audit
Sonnet 4.6
patch
30 days ago
sha256:1900655993c83c4107067375548a7be823e471d2515830842f1a12cba4bd3cdf
fix: unified object store migration — idempotent writes, JS…
Sonnet 4.6
minor
⚠
31 days ago