aaronrene / gabriel-muse public
arpeggiate.py python
166 lines 6.1 KB
Raw
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6 chore(tests): add docstring to tests/__init__.py so rc14 tr… Human 33 days ago
1 """muse arpeggiate — convert simultaneous chord notes into a sequential arpeggio.
2
3 Takes notes that overlap in time (chord voicings) and spreads them out
4 sequentially at a specified rhythmic rate. Agents that receive a chord-pad
5 track and want to convert it into a rolling arpeggio pattern can do this in
6 one command.
7
8 Usage::
9
10 muse arpeggiate tracks/chords.mid --rate 16th
11 muse arpeggiate tracks/pads.mid --rate 8th --order up
12 muse arpeggiate tracks/piano.mid --rate 8th --order down
13 muse arpeggiate tracks/chords.mid --rate 16th --order random --seed 7
14 muse arpeggiate tracks/chords.mid --rate 16th --dry-run
15
16 Order values: up (low→high), down (high→low), up-down, random
17
18 Output::
19
20 ✅ Arpeggiated tracks/chords.mid (16th-note rate, up order)
21 12 chord clusters → 48 arpeggio notes
22 Run `muse status` to review, then `muse commit`
23 """
24
25 from __future__ import annotations
26
27 import argparse
28 import logging
29 import pathlib
30 import random
31 import sys
32
33 from muse.core.errors import ExitCode
34 from muse.core.validation import contain_path
35 from muse.core.repo import require_repo
36 from muse.plugins.midi._query import NoteInfo, load_track_from_workdir, notes_to_midi_bytes
37
38
39 type _FloatMap = dict[str, float]
40 logger = logging.getLogger(__name__)
41
42 _RATE_FRACTIONS: _FloatMap = {
43 "quarter": 1.0,
44 "8th": 0.5,
45 "16th": 0.25,
46 "32nd": 0.125,
47 }
48
49 _VALID_ORDERS = ("up", "down", "up-down", "random")
50
51
52 def _cluster_notes(notes: list[NoteInfo]) -> list[list[NoteInfo]]:
53 """Group notes into time-overlapping clusters (chords)."""
54 by_time = sorted(notes, key=lambda n: n.start_tick)
55 clusters: list[list[NoteInfo]] = []
56 current: list[NoteInfo] = []
57 window = max(n.ticks_per_beat // 8 for n in notes) if notes else 1
58
59 for note in by_time:
60 if current and note.start_tick > current[0].start_tick + window:
61 clusters.append(current)
62 current = [note]
63 else:
64 current.append(note)
65 if current:
66 clusters.append(current)
67 return clusters
68
69
70 def _order_cluster(cluster: list[NoteInfo], order: str, rng: random.Random) -> list[NoteInfo]:
71 s = sorted(cluster, key=lambda n: n.pitch)
72 if order == "up":
73 return s
74 if order == "down":
75 return list(reversed(s))
76 if order == "up-down":
77 return s + list(reversed(s[1:-1]))
78 # random
79 shuffled = list(s)
80 rng.shuffle(shuffled)
81 return shuffled
82
83
84 def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None:
85 """Register the arpeggiate subcommand."""
86 parser = subparsers.add_parser("arpeggiate", help="Spread chord voicings into a sequential arpeggio pattern.", description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
87 parser.add_argument("track", metavar="TRACK", help="Workspace-relative path to a .mid file.")
88 parser.add_argument("--rate", "-r", metavar="RATE", default="16th", help="Arpeggio note rate: quarter, 8th, 16th, 32nd.")
89 parser.add_argument("--order", "-o", metavar="ORDER", default="up", help="Arpeggio order: up, down, up-down, random.")
90 parser.add_argument("--seed", metavar="INT", type=int, default=None, help="Random seed (for --order random).")
91 parser.add_argument("--dry-run", "-n", action="store_true", help="Preview without writing.")
92 parser.set_defaults(func=run)
93
94
95 def run(args: argparse.Namespace) -> None:
96 """Spread chord voicings into a sequential arpeggio pattern.
97
98 ``muse arpeggiate`` groups overlapping notes into chord clusters, then
99 replaces each cluster with an arpeggio — sequential notes at the specified
100 rhythmic rate in the specified pitch order.
101
102 Durations are set to one grid step; original velocities are preserved.
103 Use ``--order up-down`` for a ping-pong arpeggio.
104 """
105 track: str = args.track
106 rate: str = args.rate
107 order: str = args.order
108 seed: int | None = args.seed
109 dry_run: bool = args.dry_run
110
111 if rate not in _RATE_FRACTIONS:
112 print(f"❌ Unknown rate '{rate}'. Valid: {', '.join(_RATE_FRACTIONS)}", file=sys.stderr)
113 raise SystemExit(ExitCode.USER_ERROR)
114 if order not in _VALID_ORDERS:
115 print(f"❌ Unknown order '{order}'. Valid: {', '.join(_VALID_ORDERS)}", file=sys.stderr)
116 raise SystemExit(ExitCode.USER_ERROR)
117
118 root = require_repo()
119 result = load_track_from_workdir(root, track)
120 if result is None:
121 print(f"❌ Track '{track}' not found or not a valid MIDI file.", file=sys.stderr)
122 raise SystemExit(ExitCode.USER_ERROR)
123
124 notes, tpb = result
125 if not notes:
126 print(f" (track '{track}' contains no notes — nothing to arpeggiate)")
127 return
128
129 rng = random.Random(seed)
130 step = max(1, round(tpb * _RATE_FRACTIONS[rate]))
131 clusters = _cluster_notes(notes)
132 arpeggiated: list[NoteInfo] = []
133
134 for cluster in clusters:
135 ordered = _order_cluster(cluster, order, rng)
136 base_tick = ordered[0].start_tick
137 for i, note in enumerate(ordered):
138 arpeggiated.append(NoteInfo(
139 pitch=note.pitch,
140 velocity=note.velocity,
141 start_tick=base_tick + i * step,
142 duration_ticks=step,
143 channel=note.channel,
144 ticks_per_beat=note.ticks_per_beat,
145 ))
146
147 if dry_run:
148 print(f"\n[dry-run] Would arpeggiate {track} ({rate}-note rate, {order} order)")
149 print(f" Chord clusters: {len(clusters)}")
150 print(f" Output notes: {len(arpeggiated)}")
151 print(" No changes written (--dry-run).")
152 return
153
154 midi_bytes = notes_to_midi_bytes(arpeggiated, tpb)
155 workdir = root
156 try:
157 work_path = contain_path(workdir, track)
158 except ValueError as exc:
159 print(f"❌ Invalid track path: {exc}")
160 raise SystemExit(ExitCode.USER_ERROR)
161 work_path.parent.mkdir(parents=True, exist_ok=True)
162 work_path.write_bytes(midi_bytes)
163
164 print(f"\n✅ Arpeggiated {track} ({rate}-note rate, {order} order)")
165 print(f" {len(clusters)} chord clusters → {len(arpeggiated)} arpeggio notes")
166 print(" Run `muse status` to review, then `muse commit`")
File History 2 commits
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6 chore(tests): add docstring to tests/__init__.py so rc14 tr… Human 33 days ago
sha256:d8316ffae901be06347e16ab55be11868eb519dd16ade3e8aa16a99e662f7e62 baseline: rc14 re-baseline after rc3 store corruption recovery Human patch 33 days ago