aaronrene / gabriel-muse public
midi_query.py python
148 lines 5.1 KB
Raw
sha256:cf6265cea8c21d9228d90dec13ef6ec2dab5103d466db9cc4590681832de4bf8 docs(KD-STAGING): sync governance after KD-6b DONE Human 13 days ago
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
30 from __future__ import annotations
31
32 import argparse
33 import json
34 import logging
35 import pathlib
36 import sys
37
38 from muse.core.repo import require_repo
39 from muse.core.store import get_head_commit_id, read_commit, read_current_branch
40 from muse.plugins.midi._midi_query import run_query
41 from muse.core.validation import clamp_int
42
43 logger = logging.getLogger(__name__)
44
45
46 def _read_branch(root: pathlib.Path) -> str:
47 return read_current_branch(root)
48
49
50 def _resolve_head(root: pathlib.Path, alias: str | None = None) -> str | None:
51 """Resolve ``None``, ``HEAD``, or ``HEAD~N`` to a concrete commit ID."""
52 branch = _read_branch(root)
53 commit_id = get_head_commit_id(root, branch)
54 if commit_id is None:
55 return None
56 if alias is None or alias == "HEAD":
57 return commit_id
58
59 # Handle HEAD~N.
60 parts = alias.split("~")
61 if len(parts) != 2:
62 return alias
63 try:
64 steps = int(parts[1])
65 except ValueError:
66 return alias
67
68 current: str | None = commit_id
69 for _ in range(steps):
70 if current is None:
71 break
72 commit = read_commit(root, current)
73 if commit is None:
74 break
75 current = commit.parent_commit_id
76
77 return current or alias
78
79
80 def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None:
81 """Register the midi-query subcommand."""
82 parser = subparsers.add_parser("query", help="Query the MIDI note history using a MIDI DSL predicate.", description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
83 parser.add_argument("query_expr", metavar="QUERY", help=(
84 "Music query DSL expression. Examples: "
85 "\"note.pitch_class == 'Eb'\", "
86 "\"harmony.quality == 'dim' and bar == 8\", "
87 "\"agent_id == 'my-bot' and note.velocity > 80\""
88 ))
89 parser.add_argument("--track", "-t", metavar="PATH", default=None, help="Restrict search to a single MIDI file path.")
90 parser.add_argument("--from", "-f", metavar="COMMIT", default=None, dest="start", help="Start commit (default: HEAD).")
91 parser.add_argument("--to", metavar="COMMIT", default=None, dest="stop", help="Stop before this commit (exclusive).")
92 parser.add_argument("--max-results", "-n", metavar="N", type=int, default=100, help="Maximum number of matches to return.")
93 parser.add_argument("--json", action="store_true", dest="as_json", help="Output machine-readable JSON instead of formatted text.")
94 parser.set_defaults(func=run)
95
96
97 def run(args: argparse.Namespace) -> None:
98 """Query the MIDI note history using a MIDI DSL predicate."""
99 query_expr: str = args.query_expr
100 track: str | None = args.track
101 start: str | None = args.start
102 stop: str | None = args.stop
103 max_results: int = clamp_int(args.max_results, 1, 10000, 'max_results')
104 as_json: bool = args.as_json
105
106 root = require_repo()
107
108 start_id = _resolve_head(root, start)
109 if start_id is None:
110 print("❌ No commits in this repository.", file=sys.stderr)
111 raise SystemExit(1)
112
113 try:
114 matches = run_query(
115 query_expr,
116 root,
117 start_id,
118 track_filter=track,
119 from_commit_id=stop,
120 max_results=max_results,
121 )
122 except ValueError as exc:
123 print(f"❌ Query parse error: {exc}", file=sys.stderr)
124 raise SystemExit(1)
125
126 if not matches:
127 print("No matches found.")
128 return
129
130 if as_json:
131 sys.stdout.write(json.dumps(matches, indent=2) + "\n")
132 return
133
134 for m in matches:
135 print(
136 f"commit {m['commit_short']} {m['committed_at'][:19]} "
137 f"author={m['author']} agent={m['agent_id'] or '—'}"
138 )
139 print(f" track={m['track']} bar={m['bar']} chord={m['chord'] or '—'}")
140 for n in m["notes"]:
141 print(
142 f" {n['pitch_class']:3} (MIDI {n['pitch']:3}) "
143 f"vel={n['velocity']:3} ch={n['channel']} "
144 f"beat={n['beat']:.2f} dur={n['duration_beats']:.2f}"
145 )
146 print("")
147
148 print(f"— {len(matches)} match{'es' if len(matches) != 1 else ''} —")
File History 1 commit
sha256:cf6265cea8c21d9228d90dec13ef6ec2dab5103d466db9cc4590681832de4bf8 docs(KD-STAGING): sync governance after KD-6b DONE Human 13 days ago