midi_query.py
python
| 1 | """``muse midi-query`` — MIDI DSL query over commit history. |
| 2 | |
| 3 | Evaluates a predicate expression against the note content of all MIDI tracks |
| 4 | across the commit history and returns matching bars with chord annotations, |
| 5 | agent provenance, and note tables. |
| 6 | |
| 7 | Usage:: |
| 8 | |
| 9 | muse midi-query "note.pitch_class == 'Eb' and bar == 12" |
| 10 | muse midi-query "note.velocity > 100" --track piano.mid |
| 11 | muse midi-query "agent_id == 'counterpoint-bot'" --from HEAD~10 |
| 12 | muse midi-query "harmony.quality == 'dim'" --json |
| 13 | |
| 14 | Grammar:: |
| 15 | |
| 16 | query = or_expr |
| 17 | or_expr = and_expr ( 'or' and_expr )* |
| 18 | and_expr = not_expr ( 'and' not_expr )* |
| 19 | not_expr = 'not' not_expr | atom |
| 20 | atom = '(' query ')' | FIELD OP VALUE |
| 21 | FIELD = note.pitch | note.pitch_class | note.velocity | |
| 22 | note.channel | note.duration | bar | track | |
| 23 | harmony.chord | harmony.quality | |
| 24 | author | agent_id | model_id | toolchain_id |
| 25 | OP = == | != | > | < | >= | <= |
| 26 | |
| 27 | See ``muse/plugins/midi/_midi_query.py`` for the full grammar reference. |
| 28 | """ |
| 29 | from __future__ import annotations |
| 30 | |
| 31 | import json |
| 32 | import logging |
| 33 | import pathlib |
| 34 | import sys |
| 35 | |
| 36 | import typer |
| 37 | |
| 38 | from muse.core.repo import require_repo |
| 39 | from muse.core.store import get_head_commit_id, read_commit |
| 40 | from muse.plugins.midi._midi_query import run_query |
| 41 | |
| 42 | logger = logging.getLogger(__name__) |
| 43 | |
| 44 | app = typer.Typer(no_args_is_help=True) |
| 45 | |
| 46 | |
| 47 | def _read_branch(root: pathlib.Path) -> str: |
| 48 | head_ref = (root / ".muse" / "HEAD").read_text().strip() |
| 49 | return head_ref.removeprefix("refs/heads/").strip() |
| 50 | |
| 51 | |
| 52 | def _resolve_head(root: pathlib.Path, alias: str | None = None) -> str | None: |
| 53 | """Resolve ``None``, ``HEAD``, or ``HEAD~N`` to a concrete commit ID.""" |
| 54 | branch = _read_branch(root) |
| 55 | commit_id = get_head_commit_id(root, branch) |
| 56 | if commit_id is None: |
| 57 | return None |
| 58 | if alias is None or alias == "HEAD": |
| 59 | return commit_id |
| 60 | |
| 61 | # Handle HEAD~N. |
| 62 | parts = alias.split("~") |
| 63 | if len(parts) != 2: |
| 64 | return alias |
| 65 | try: |
| 66 | steps = int(parts[1]) |
| 67 | except ValueError: |
| 68 | return alias |
| 69 | |
| 70 | current: str | None = commit_id |
| 71 | for _ in range(steps): |
| 72 | if current is None: |
| 73 | break |
| 74 | commit = read_commit(root, current) |
| 75 | if commit is None: |
| 76 | break |
| 77 | current = commit.parent_commit_id |
| 78 | |
| 79 | return current or alias |
| 80 | |
| 81 | |
| 82 | @app.command(name="midi-query") |
| 83 | def midi_query_cmd( |
| 84 | query_expr: str = typer.Argument( |
| 85 | ..., |
| 86 | metavar="QUERY", |
| 87 | help=( |
| 88 | "Music query DSL expression. Examples: " |
| 89 | "\"note.pitch_class == 'Eb'\", " |
| 90 | "\"harmony.quality == 'dim' and bar == 8\", " |
| 91 | "\"agent_id == 'my-bot' and note.velocity > 80\"" |
| 92 | ), |
| 93 | ), |
| 94 | track: str | None = typer.Option( |
| 95 | None, |
| 96 | "--track", |
| 97 | "-t", |
| 98 | metavar="PATH", |
| 99 | help="Restrict search to a single MIDI file path.", |
| 100 | ), |
| 101 | start: str | None = typer.Option( |
| 102 | None, |
| 103 | "--from", |
| 104 | "-f", |
| 105 | metavar="COMMIT", |
| 106 | help="Start commit (default: HEAD).", |
| 107 | ), |
| 108 | stop: str | None = typer.Option( |
| 109 | None, |
| 110 | "--to", |
| 111 | metavar="COMMIT", |
| 112 | help="Stop before this commit (exclusive).", |
| 113 | ), |
| 114 | max_results: int = typer.Option( |
| 115 | 100, |
| 116 | "--max-results", |
| 117 | "-n", |
| 118 | metavar="N", |
| 119 | help="Maximum number of matches to return.", |
| 120 | ), |
| 121 | as_json: bool = typer.Option( |
| 122 | False, |
| 123 | "--json", |
| 124 | help="Output machine-readable JSON instead of formatted text.", |
| 125 | ), |
| 126 | ) -> None: |
| 127 | """Query the MIDI note history using a MIDI DSL predicate.""" |
| 128 | root = require_repo() |
| 129 | |
| 130 | start_id = _resolve_head(root, start) |
| 131 | if start_id is None: |
| 132 | typer.echo("❌ No commits in this repository.", err=True) |
| 133 | raise typer.Exit(1) |
| 134 | |
| 135 | try: |
| 136 | matches = run_query( |
| 137 | query_expr, |
| 138 | root, |
| 139 | start_id, |
| 140 | track_filter=track, |
| 141 | from_commit_id=stop, |
| 142 | max_results=max_results, |
| 143 | ) |
| 144 | except ValueError as exc: |
| 145 | typer.echo(f"❌ Query parse error: {exc}", err=True) |
| 146 | raise typer.Exit(1) |
| 147 | |
| 148 | if not matches: |
| 149 | typer.echo("No matches found.") |
| 150 | return |
| 151 | |
| 152 | if as_json: |
| 153 | sys.stdout.write(json.dumps(matches, indent=2) + "\n") |
| 154 | return |
| 155 | |
| 156 | for m in matches: |
| 157 | typer.echo( |
| 158 | f"commit {m['commit_short']} {m['committed_at'][:19]} " |
| 159 | f"author={m['author']} agent={m['agent_id'] or '—'}" |
| 160 | ) |
| 161 | typer.echo(f" track={m['track']} bar={m['bar']} chord={m['chord'] or '—'}") |
| 162 | for n in m["notes"]: |
| 163 | typer.echo( |
| 164 | f" {n['pitch_class']:3} (MIDI {n['pitch']:3}) " |
| 165 | f"vel={n['velocity']:3} ch={n['channel']} " |
| 166 | f"beat={n['beat']:.2f} dur={n['duration_beats']:.2f}" |
| 167 | ) |
| 168 | typer.echo("") |
| 169 | |
| 170 | typer.echo(f"— {len(matches)} match{'es' if len(matches) != 1 else ''} —") |