cgcardona / muse public
humanize.py python
569 lines 17.7 KB
Raw
12901c5a Initial extraction from tellurstori/maestro cgcardona <gabriel@tellurstori.com> 4d ago
1 """muse humanize — apply micro-timing and velocity humanization to quantized MIDI.
2
3 Machine-quantized MIDI sounds robotic. This command adds realistic human-performance
4 variation — subtle micro-timing offsets and velocity fluctuations — drawn from a
5 configurable distribution, producing a new Muse commit that sounds natural while
6 preserving the musical identity of the original.
7
8 Presets
9 -------
10 ``--tight`` Subtle humanization: timing ±5 ms, velocity ±5.
11 ``--natural`` Moderate humanization: timing ±12 ms, velocity ±10. (default)
12 ``--loose`` Heavy humanization: timing ±20 ms, velocity ±15.
13
14 Custom control
15 --------------
16 ``--factor FLOAT`` 0.0 = no change; 1.0 = maximum natural variation (maps to
17 the ``--loose`` ceiling).
18 ``--timing-only`` Apply timing variation only; preserve all velocities.
19 ``--velocity-only`` Apply velocity variation only; preserve all note positions.
20
21 Scoping
22 -------
23 ``--track TEXT`` Restrict humanization to a single named track.
24 ``--section TEXT`` Restrict humanization to a named section/region.
25
26 Reproducibility
27 ---------------
28 ``--seed N`` Fix the random seed so the same invocation produces identical
29 output every time. Without ``--seed``, each run is stochastic.
30
31 Output
32 ------
33 Default: human-readable summary table showing per-track timing/velocity deltas.
34 ``--json`` Emit a machine-readable JSON payload — use in agentic pipelines.
35
36 Example::
37
38 muse humanize --natural --seed 42
39 muse humanize HEAD~1 --loose --track bass
40 muse humanize --factor 0.6 --timing-only --json
41 """
42 from __future__ import annotations
43
44 import asyncio
45 import json
46 import logging
47 import pathlib
48 import random
49 from typing import Optional
50
51 import typer
52 from sqlalchemy.ext.asyncio import AsyncSession
53 from typing_extensions import Annotated, TypedDict
54
55 from maestro.muse_cli._repo import require_repo
56 from maestro.muse_cli.db import open_session
57 from maestro.muse_cli.errors import ExitCode
58
59 logger = logging.getLogger(__name__)
60
61 app = typer.Typer()
62
63 # ---------------------------------------------------------------------------
64 # Preset constants — timing in ms, velocity in MIDI units (0–127)
65 # ---------------------------------------------------------------------------
66
67 TIGHT_TIMING_MS: int = 5
68 TIGHT_VELOCITY: int = 5
69
70 NATURAL_TIMING_MS: int = 12
71 NATURAL_VELOCITY: int = 10
72
73 LOOSE_TIMING_MS: int = 20
74 LOOSE_VELOCITY: int = 15
75
76 FACTOR_MIN: float = 0.0
77 FACTOR_MAX: float = 1.0
78
79 # Drum channel (General MIDI channel 10, zero-indexed as 9).
80 # Drum tracks are excluded from timing humanization to preserve groove identity.
81 DRUM_CHANNEL: int = 9
82
83
84 # ---------------------------------------------------------------------------
85 # Named result types (stable CLI contract)
86 # ---------------------------------------------------------------------------
87
88
89 class TrackHumanizeResult(TypedDict):
90 """Humanization outcome for a single track."""
91
92 track: str
93 timing_range_ms: int
94 velocity_range: int
95 notes_affected: int
96 drum_channel_excluded: bool
97
98
99 class HumanizeResult(TypedDict):
100 """Full result emitted by ``muse humanize``.
101
102 Consumed by downstream agents that need to know what changed and by
103 how much — e.g. a groove-check agent can compare timing variance
104 before/after to confirm humanization took effect.
105 """
106
107 commit: str
108 branch: str
109 source_commit: str
110 preset: str
111 factor: float
112 seed: Optional[int]
113 timing_only: bool
114 velocity_only: bool
115 track_filter: Optional[str]
116 section_filter: Optional[str]
117 tracks: list[TrackHumanizeResult]
118 new_commit_id: str
119
120
121 # ---------------------------------------------------------------------------
122 # Humanization engine
123 # ---------------------------------------------------------------------------
124
125 #: Stub track names used as placeholders until real MIDI data is queryable.
126 _STUB_TRACKS: tuple[str, ...] = ("drums", "bass", "keys", "lead")
127
128 #: MIDI note count per track (stub placeholder).
129 _STUB_NOTE_COUNT: int = 64
130
131
132 def _resolve_preset(
133 *,
134 tight: bool,
135 natural: bool,
136 loose: bool,
137 factor: Optional[float],
138 ) -> tuple[str, float]:
139 """Resolve flag combination to a (preset_label, factor) pair.
140
141 Priority: ``--factor`` > ``--tight`` > ``--natural`` > ``--loose``.
142 Default when no flag is given: ``natural`` preset.
143
144 Args:
145 tight: ``--tight`` flag.
146 natural: ``--natural`` flag.
147 loose: ``--loose`` flag.
148 factor: ``--factor`` float, if provided.
149
150 Returns:
151 Tuple of (preset_label, normalized_factor).
152
153 Raises:
154 ValueError: If more than one preset flag is set simultaneously.
155 """
156 preset_count = sum([tight, natural, loose, factor is not None])
157 if preset_count > 1 and factor is None:
158 raise ValueError(
159 "Only one of --tight / --natural / --loose may be specified at a time."
160 )
161
162 if factor is not None:
163 return ("custom", factor)
164 if tight:
165 return ("tight", 0.25)
166 if loose:
167 return ("loose", 1.0)
168 # Default: natural
169 return ("natural", 0.6)
170
171
172 def _timing_ms_for_factor(factor: float) -> int:
173 """Return the timing range in ms for a given factor.
174
175 Linearly interpolates between 0 ms (factor=0.0) and
176 ``LOOSE_TIMING_MS`` (factor=1.0).
177
178 Args:
179 factor: Normalized humanization factor [0.0, 1.0].
180
181 Returns:
182 Timing range in milliseconds (integer, ≥ 0).
183 """
184 return round(factor * LOOSE_TIMING_MS)
185
186
187 def _velocity_range_for_factor(factor: float) -> int:
188 """Return the velocity variation range for a given factor.
189
190 Linearly interpolates between 0 (factor=0.0) and
191 ``LOOSE_VELOCITY`` (factor=1.0).
192
193 Args:
194 factor: Normalized humanization factor [0.0, 1.0].
195
196 Returns:
197 Velocity range in MIDI units (integer, ≥ 0).
198 """
199 return round(factor * LOOSE_VELOCITY)
200
201
202 def _apply_humanization(
203 *,
204 track_name: str,
205 timing_ms: int,
206 velocity_range: int,
207 timing_only: bool,
208 velocity_only: bool,
209 rng: random.Random,
210 ) -> TrackHumanizeResult:
211 """Compute humanization deltas for one track.
212
213 Drums (channel 9) are excluded from timing variation to preserve groove
214 identity — only velocity humanization is applied to the drum track.
215
216 This is a stub: real implementation will load MIDI notes from the commit
217 snapshot, apply the offsets, and write back a new object. The returned
218 ``notes_affected`` count is a realistic placeholder.
219
220 Args:
221 track_name: Name of the MIDI track.
222 timing_ms: Maximum timing offset in milliseconds (±).
223 velocity_range: Maximum velocity offset (±).
224 timing_only: If True, skip velocity humanization.
225 velocity_only: If True, skip timing humanization.
226 rng: Seeded random instance for deterministic output.
227
228 Returns:
229 :class:`TrackHumanizeResult` with applied delta metadata.
230 """
231 is_drum = track_name.lower() in {"drums", "percussion", "kit"}
232
233 effective_timing = 0 if (velocity_only or is_drum) else timing_ms
234 effective_velocity = 0 if timing_only else velocity_range
235
236 # Stub: simulate a note count between 32 and 128.
237 notes_affected = rng.randint(32, 128)
238
239 return TrackHumanizeResult(
240 track=track_name,
241 timing_range_ms=effective_timing,
242 velocity_range=effective_velocity,
243 notes_affected=notes_affected,
244 drum_channel_excluded=is_drum and not velocity_only,
245 )
246
247
248 # ---------------------------------------------------------------------------
249 # Testable async core
250 # ---------------------------------------------------------------------------
251
252
253 async def _humanize_async(
254 *,
255 root: pathlib.Path,
256 session: AsyncSession,
257 source_commit: Optional[str],
258 preset: str,
259 factor: float,
260 seed: Optional[int],
261 timing_only: bool,
262 velocity_only: bool,
263 track: Optional[str],
264 section: Optional[str],
265 message: Optional[str],
266 as_json: bool,
267 ) -> HumanizeResult:
268 """Apply humanization to a commit's MIDI and emit the result.
269
270 Stub implementation: computes per-track humanization metadata from the
271 resolved factor and produces a new (fake) commit ID. The full
272 implementation will:
273
274 1. Load MIDI notes from the source commit's snapshot.
275 2. Apply Gaussian-distributed timing offsets (in ticks) and velocity
276 deltas drawn from a uniform distribution within ±range.
277 3. Write the modified notes back as a new snapshot object.
278 4. Commit the snapshot via the Muse VCS commit engine.
279
280 Args:
281 root: Repository root (containing ``.muse/``).
282 session: Open async DB session (reserved for full implementation).
283 source_commit: Source commit ref, or ``None`` for HEAD.
284 preset: Preset label (``tight``, ``natural``, ``loose``, ``custom``).
285 factor: Normalized humanization factor [0.0, 1.0].
286 seed: Random seed for deterministic output; ``None`` = stochastic.
287 timing_only: If True, skip velocity humanization.
288 velocity_only: If True, skip timing humanization.
289 track: Restrict to a single named track; ``None`` = all tracks.
290 section: Restrict to a named section/region (stub: noted in output).
291 message: Commit message override; ``None`` = auto-generated.
292 as_json: Emit JSON instead of ASCII table.
293
294 Returns:
295 :class:`HumanizeResult` with full metadata for agent consumption.
296 """
297 muse_dir = root / ".muse"
298
299 # Resolve branch and HEAD commit.
300 head_ref = (muse_dir / "HEAD").read_text().strip()
301 branch = head_ref.rsplit("/", 1)[-1] if "/" in head_ref else head_ref
302 ref_path = muse_dir / pathlib.Path(head_ref)
303 head_sha = ref_path.read_text().strip() if ref_path.exists() else ""
304 resolved_source = source_commit or (head_sha[:8] if head_sha else "HEAD")
305
306 timing_ms = _timing_ms_for_factor(factor)
307 vel_range = _velocity_range_for_factor(factor)
308
309 # Seed-based RNG for deterministic humanization.
310 rng = random.Random(seed)
311
312 # Determine which tracks to process.
313 all_tracks = list(_STUB_TRACKS)
314 if track:
315 all_tracks = [t for t in all_tracks if t.lower().startswith(track.lower())]
316
317 if section:
318 logger.warning("⚠️ --section %s: region scoping not yet implemented.", section)
319
320 track_results = [
321 _apply_humanization(
322 track_name=t,
323 timing_ms=timing_ms,
324 velocity_range=vel_range,
325 timing_only=timing_only,
326 velocity_only=velocity_only,
327 rng=rng,
328 )
329 for t in all_tracks
330 ]
331
332 # Stub: generate a deterministic fake commit ID.
333 fake_commit_seed = rng.randint(0, 0xFFFFFFFF)
334 new_commit_id = f"{fake_commit_seed:08x}stub"
335
336 result = HumanizeResult(
337 commit=resolved_source,
338 branch=branch,
339 source_commit=resolved_source,
340 preset=preset,
341 factor=factor,
342 seed=seed,
343 timing_only=timing_only,
344 velocity_only=velocity_only,
345 track_filter=track,
346 section_filter=section,
347 tracks=track_results,
348 new_commit_id=new_commit_id,
349 )
350
351 if as_json:
352 _render_json(result)
353 else:
354 _render_table(result)
355
356 logger.info("✅ muse humanize: created commit %s", new_commit_id)
357 return result
358
359
360 # ---------------------------------------------------------------------------
361 # Renderers
362 # ---------------------------------------------------------------------------
363
364 _COL_WIDTHS = (12, 12, 10, 14, 16) # track, timing_ms, vel_range, notes, drum_excluded
365
366
367 def _render_table(result: HumanizeResult) -> None:
368 """Emit a human-readable summary of the humanization pass."""
369 seed_note = f" seed={result['seed']}" if result["seed"] is not None else ""
370 typer.echo(
371 f"Humanize — {result['preset']} (factor={result['factor']:.2f})"
372 f" source={result['source_commit']}{seed_note}"
373 )
374 typer.echo("")
375
376 header = (
377 f"{'Track':<{_COL_WIDTHS[0]}} "
378 f"{'Timing ±ms':>{_COL_WIDTHS[1]}} "
379 f"{'Vel ±':>{_COL_WIDTHS[2]}} "
380 f"{'Notes':>{_COL_WIDTHS[3]}} "
381 f"{'Drum excluded':<{_COL_WIDTHS[4]}}"
382 )
383 sep = " ".join("-" * w for w in _COL_WIDTHS)
384 typer.echo(header)
385 typer.echo(sep)
386
387 for tr in result["tracks"]:
388 drum_flag = "yes" if tr["drum_channel_excluded"] else "no"
389 typer.echo(
390 f"{tr['track']:<{_COL_WIDTHS[0]}} "
391 f"{tr['timing_range_ms']:>{_COL_WIDTHS[1]}} "
392 f"{tr['velocity_range']:>{_COL_WIDTHS[2]}} "
393 f"{tr['notes_affected']:>{_COL_WIDTHS[3]}} "
394 f"{drum_flag:<{_COL_WIDTHS[4]}}"
395 )
396
397 typer.echo("")
398 typer.echo(f"✅ New commit: {result['new_commit_id']}")
399 typer.echo(" (stub — full MIDI rewrite pending Storpheus note-level access)")
400
401
402 def _render_json(result: HumanizeResult) -> None:
403 """Emit the humanization result as a JSON object."""
404 typer.echo(json.dumps(dict(result), indent=2))
405
406
407 # ---------------------------------------------------------------------------
408 # Typer command
409 # ---------------------------------------------------------------------------
410
411
412 @app.callback(invoke_without_command=True)
413 def humanize(
414 ctx: typer.Context,
415 source_commit: Annotated[
416 Optional[str],
417 typer.Argument(
418 help="Source commit ref to humanize (default: HEAD).",
419 show_default=False,
420 metavar="COMMIT",
421 ),
422 ] = None,
423 tight: Annotated[
424 bool,
425 typer.Option(
426 "--tight",
427 help=f"Subtle humanization: timing ±{TIGHT_TIMING_MS} ms, velocity ±{TIGHT_VELOCITY}.",
428 ),
429 ] = False,
430 natural: Annotated[
431 bool,
432 typer.Option(
433 "--natural",
434 help=f"Moderate humanization: timing ±{NATURAL_TIMING_MS} ms, velocity ±{NATURAL_VELOCITY}. (default)",
435 ),
436 ] = False,
437 loose: Annotated[
438 bool,
439 typer.Option(
440 "--loose",
441 help=f"Heavy humanization: timing ±{LOOSE_TIMING_MS} ms, velocity ±{LOOSE_VELOCITY}.",
442 ),
443 ] = False,
444 factor: Annotated[
445 Optional[float],
446 typer.Option(
447 "--factor",
448 help="Custom factor: 0.0 = no change, 1.0 = maximum variation. Overrides preset flags.",
449 min=FACTOR_MIN,
450 max=FACTOR_MAX,
451 show_default=False,
452 ),
453 ] = None,
454 timing_only: Annotated[
455 bool,
456 typer.Option(
457 "--timing-only",
458 help="Apply timing variation only; preserve all velocities.",
459 ),
460 ] = False,
461 velocity_only: Annotated[
462 bool,
463 typer.Option(
464 "--velocity-only",
465 help="Apply velocity variation only; preserve all note positions.",
466 ),
467 ] = False,
468 track: Annotated[
469 Optional[str],
470 typer.Option(
471 "--track",
472 help="Restrict humanization to a specific track (case-insensitive prefix match).",
473 show_default=False,
474 metavar="TEXT",
475 ),
476 ] = None,
477 section: Annotated[
478 Optional[str],
479 typer.Option(
480 "--section",
481 help="Restrict humanization to a specific section/region.",
482 show_default=False,
483 metavar="TEXT",
484 ),
485 ] = None,
486 seed: Annotated[
487 Optional[int],
488 typer.Option(
489 "--seed",
490 help="Random seed for reproducible humanization. Without --seed, output is stochastic.",
491 show_default=False,
492 ),
493 ] = None,
494 message: Annotated[
495 Optional[str],
496 typer.Option(
497 "--message",
498 "-m",
499 help="Commit message for the humanization commit.",
500 show_default=False,
501 metavar="TEXT",
502 ),
503 ] = None,
504 as_json: Annotated[
505 bool,
506 typer.Option(
507 "--json",
508 help="Emit machine-readable JSON output for agent consumption.",
509 ),
510 ] = False,
511 ) -> None:
512 """Apply micro-timing and velocity humanization to quantized MIDI.
513
514 Produces a new Muse commit with realistic human-performance variation.
515 Use ``--seed`` for deterministic output; omit it for a fresh stochastic
516 pass each invocation. Drum tracks are automatically excluded from timing
517 variation to preserve groove identity.
518 """
519 root = require_repo()
520
521 # Validate mutually exclusive flags.
522 if timing_only and velocity_only:
523 typer.echo(
524 "❌ --timing-only and --velocity-only are mutually exclusive. "
525 "Omit one or both to apply full humanization."
526 )
527 raise typer.Exit(code=ExitCode.USER_ERROR)
528
529 preset_flags = sum([tight, natural, loose])
530 if preset_flags > 1:
531 typer.echo("❌ Only one of --tight / --natural / --loose may be specified at a time.")
532 raise typer.Exit(code=ExitCode.USER_ERROR)
533
534 try:
535 preset_label, resolved_factor = _resolve_preset(
536 tight=tight,
537 natural=natural,
538 loose=loose,
539 factor=factor,
540 )
541 except ValueError as exc:
542 typer.echo(f"❌ {exc}")
543 raise typer.Exit(code=ExitCode.USER_ERROR)
544
545 async def _run() -> None:
546 async with open_session() as session:
547 await _humanize_async(
548 root=root,
549 session=session,
550 source_commit=source_commit,
551 preset=preset_label,
552 factor=resolved_factor,
553 seed=seed,
554 timing_only=timing_only,
555 velocity_only=velocity_only,
556 track=track,
557 section=section,
558 message=message,
559 as_json=as_json,
560 )
561
562 try:
563 asyncio.run(_run())
564 except typer.Exit:
565 raise
566 except Exception as exc:
567 typer.echo(f"❌ muse humanize failed: {exc}")
568 logger.error("❌ muse humanize error: %s", exc, exc_info=True)
569 raise typer.Exit(code=ExitCode.INTERNAL_ERROR)