velocity_normalize.py
python
| 1 | """muse normalize — normalize MIDI velocities to a target range. |
| 2 | |
| 3 | Rescales all note velocities so the softest note maps to --min and the loudest |
| 4 | to --max. Preserves the relative dynamics while adjusting the overall level. |
| 5 | Essential when merging tracks from multiple agents that were recorded at |
| 6 | different volumes. |
| 7 | |
| 8 | Usage:: |
| 9 | |
| 10 | muse normalize tracks/melody.mid |
| 11 | muse normalize tracks/drums.mid --min 50 --max 110 |
| 12 | muse normalize tracks/piano.mid --target-mean 80 |
| 13 | muse normalize tracks/bass.mid --dry-run |
| 14 | |
| 15 | Output:: |
| 16 | |
| 17 | ✅ Normalised tracks/melody.mid |
| 18 | 48 notes rescaled · range: 32–78 → 40–100 |
| 19 | Mean velocity: 61.3 → 72.0 |
| 20 | Run `muse status` to review, then `muse commit` |
| 21 | """ |
| 22 | |
| 23 | from __future__ import annotations |
| 24 | |
| 25 | import logging |
| 26 | import pathlib |
| 27 | |
| 28 | import typer |
| 29 | |
| 30 | from muse.core.errors import ExitCode |
| 31 | from muse.core.repo import require_repo |
| 32 | from muse.plugins.midi._query import NoteInfo, load_track_from_workdir, notes_to_midi_bytes |
| 33 | |
| 34 | logger = logging.getLogger(__name__) |
| 35 | app = typer.Typer() |
| 36 | |
| 37 | _MIDI_MIN = 1 |
| 38 | _MIDI_MAX = 127 |
| 39 | |
| 40 | |
| 41 | def _rescale(velocity: int, src_lo: int, src_hi: int, dst_lo: int, dst_hi: int) -> int: |
| 42 | if src_hi == src_lo: |
| 43 | return (dst_lo + dst_hi) // 2 |
| 44 | ratio = (velocity - src_lo) / (src_hi - src_lo) |
| 45 | return round(dst_lo + ratio * (dst_hi - dst_lo)) |
| 46 | |
| 47 | |
| 48 | @app.callback(invoke_without_command=True) |
| 49 | def normalize( |
| 50 | ctx: typer.Context, |
| 51 | track: str = typer.Argument(..., metavar="TRACK", help="Workspace-relative path to a .mid file."), |
| 52 | min_vel: int = typer.Option( |
| 53 | 40, "--min", metavar="VEL", |
| 54 | help="Target minimum velocity (default 40).", |
| 55 | ), |
| 56 | max_vel: int = typer.Option( |
| 57 | 110, "--max", metavar="VEL", |
| 58 | help="Target maximum velocity (default 110).", |
| 59 | ), |
| 60 | dry_run: bool = typer.Option(False, "--dry-run", "-n", help="Preview without writing."), |
| 61 | ) -> None: |
| 62 | """Rescale note velocities to a target dynamic range. |
| 63 | |
| 64 | ``muse normalize`` linearly maps the existing velocity range to |
| 65 | [--min, --max], preserving the relative dynamic contour while adjusting |
| 66 | the absolute level. This is the standard first step when integrating |
| 67 | tracks from multiple agents that were written at different volume levels. |
| 68 | |
| 69 | Use ``--min 64 --max 96`` for a narrow, compressed dynamic range; |
| 70 | use ``--min 20 --max 127`` for the full MIDI spectrum. |
| 71 | """ |
| 72 | if not _MIDI_MIN <= min_vel <= _MIDI_MAX: |
| 73 | typer.echo(f"❌ --min must be between {_MIDI_MIN} and {_MIDI_MAX}.", err=True) |
| 74 | raise typer.Exit(code=ExitCode.USER_ERROR) |
| 75 | if not _MIDI_MIN <= max_vel <= _MIDI_MAX: |
| 76 | typer.echo(f"❌ --max must be between {_MIDI_MIN} and {_MIDI_MAX}.", err=True) |
| 77 | raise typer.Exit(code=ExitCode.USER_ERROR) |
| 78 | if min_vel >= max_vel: |
| 79 | typer.echo("❌ --min must be less than --max.", err=True) |
| 80 | raise typer.Exit(code=ExitCode.USER_ERROR) |
| 81 | |
| 82 | root = require_repo() |
| 83 | result = load_track_from_workdir(root, track) |
| 84 | if result is None: |
| 85 | typer.echo(f"❌ Track '{track}' not found or not a valid MIDI file.", err=True) |
| 86 | raise typer.Exit(code=ExitCode.USER_ERROR) |
| 87 | |
| 88 | notes, tpb = result |
| 89 | if not notes: |
| 90 | typer.echo(f" (track '{track}' contains no notes — nothing to normalise)") |
| 91 | return |
| 92 | |
| 93 | vels = [n.velocity for n in notes] |
| 94 | src_lo, src_hi = min(vels), max(vels) |
| 95 | old_mean = sum(vels) / len(vels) |
| 96 | |
| 97 | normalised: list[NoteInfo] = [ |
| 98 | NoteInfo( |
| 99 | pitch=n.pitch, |
| 100 | velocity=_rescale(n.velocity, src_lo, src_hi, min_vel, max_vel), |
| 101 | start_tick=n.start_tick, |
| 102 | duration_ticks=n.duration_ticks, |
| 103 | channel=n.channel, |
| 104 | ticks_per_beat=n.ticks_per_beat, |
| 105 | ) |
| 106 | for n in notes |
| 107 | ] |
| 108 | new_mean = sum(n.velocity for n in normalised) / len(normalised) |
| 109 | |
| 110 | if dry_run: |
| 111 | typer.echo(f"\n[dry-run] Would normalise {track}") |
| 112 | typer.echo(f" Notes: {len(notes)}") |
| 113 | typer.echo(f" Range: {src_lo}–{src_hi} → {min_vel}–{max_vel}") |
| 114 | typer.echo(f" Mean velocity: {old_mean:.1f} → {new_mean:.1f}") |
| 115 | typer.echo(" No changes written (--dry-run).") |
| 116 | return |
| 117 | |
| 118 | midi_bytes = notes_to_midi_bytes(normalised, tpb) |
| 119 | work_path = root / "muse-work" / track |
| 120 | if not work_path.parent.exists(): |
| 121 | work_path = root / track |
| 122 | work_path.write_bytes(midi_bytes) |
| 123 | |
| 124 | typer.echo(f"\n✅ Normalised {track}") |
| 125 | typer.echo(f" {len(normalised)} notes rescaled · range: {src_lo}–{src_hi} → {min_vel}–{max_vel}") |
| 126 | typer.echo(f" Mean velocity: {old_mean:.1f} → {new_mean:.1f}") |
| 127 | typer.echo(" Run `muse status` to review, then `muse commit`") |