tempo_scale.py
python
| 1 | """muse tempo-scale — stretch or compress the timing of a commit. |
| 2 | |
| 3 | Time-scaling (changing tempo while preserving pitch) is a fundamental |
| 4 | production operation: halving time values creates a double-time feel; |
| 5 | doubling them creates a half-time groove. Because Muse commits track |
| 6 | MIDI note events and tempo metadata, this transformation is applied |
| 7 | deterministically — same factor + same source commit = identical result. |
| 8 | |
| 9 | Pitch is *not* affected; this is pure MIDI timing manipulation, not |
| 10 | audio time-stretching. |
| 11 | |
| 12 | Command forms |
| 13 | ------------- |
| 14 | |
| 15 | Scale by an explicit factor (0.5 = half-time, 2.0 = double-time):: |
| 16 | |
| 17 | muse tempo-scale 0.5 |
| 18 | |
| 19 | Scale to reach an exact BPM (computes factor = target / current BPM):: |
| 20 | |
| 21 | muse tempo-scale --bpm 128 |
| 22 | |
| 23 | Scale a specific commit instead of HEAD:: |
| 24 | |
| 25 | muse tempo-scale 0.5 a1b2c3d4 |
| 26 | |
| 27 | Scale only one MIDI track:: |
| 28 | |
| 29 | muse tempo-scale 2.0 --track bass |
| 30 | |
| 31 | Preserve CC/expression timing proportionally:: |
| 32 | |
| 33 | muse tempo-scale 0.5 --preserve-expressions |
| 34 | |
| 35 | Provide a custom commit message:: |
| 36 | |
| 37 | muse tempo-scale 0.5 --message "half-time remix" |
| 38 | |
| 39 | Machine-readable JSON output:: |
| 40 | |
| 41 | muse tempo-scale 2.0 --json |
| 42 | """ |
| 43 | from __future__ import annotations |
| 44 | |
| 45 | import asyncio |
| 46 | import hashlib |
| 47 | import json |
| 48 | import logging |
| 49 | import pathlib |
| 50 | from typing import Optional |
| 51 | |
| 52 | import typer |
| 53 | from sqlalchemy.ext.asyncio import AsyncSession |
| 54 | from typing_extensions import Annotated, TypedDict |
| 55 | |
| 56 | from maestro.muse_cli._repo import require_repo |
| 57 | from maestro.muse_cli.db import open_session |
| 58 | from maestro.muse_cli.errors import ExitCode |
| 59 | |
| 60 | logger = logging.getLogger(__name__) |
| 61 | |
| 62 | app = typer.Typer() |
| 63 | |
| 64 | # --------------------------------------------------------------------------- |
| 65 | # Constants |
| 66 | # --------------------------------------------------------------------------- |
| 67 | |
| 68 | FACTOR_MIN = 0.01 # below this the result is effectively silence |
| 69 | FACTOR_MAX = 100.0 # above this is unreasonably fast |
| 70 | |
| 71 | |
| 72 | # --------------------------------------------------------------------------- |
| 73 | # Named result types (stable CLI contract) |
| 74 | # --------------------------------------------------------------------------- |
| 75 | |
| 76 | |
| 77 | class TempoScaleResult(TypedDict): |
| 78 | """Result of a tempo-scale operation. |
| 79 | |
| 80 | Returned by ``_tempo_scale_async`` and emitted as JSON when ``--json`` |
| 81 | is given. Agents should treat ``new_commit`` as the SHA that replaces |
| 82 | the source commit in the timeline. |
| 83 | |
| 84 | Fields |
| 85 | ------ |
| 86 | source_commit: |
| 87 | Short SHA of the input commit. |
| 88 | new_commit: |
| 89 | Short SHA of the newly created tempo-scaled commit. |
| 90 | factor: |
| 91 | Scaling factor that was applied. ``< 1`` = slower; ``> 1`` = faster. |
| 92 | source_bpm: |
| 93 | Tempo of the source commit, in beats per minute (stub: placeholder). |
| 94 | new_bpm: |
| 95 | Resulting tempo after scaling. |
| 96 | track: |
| 97 | Name of the MIDI track that was scaled, or ``"all"`` if no filter. |
| 98 | preserve_expressions: |
| 99 | Whether CC/expression events were scaled proportionally. |
| 100 | message: |
| 101 | Commit message for the new scaled commit. |
| 102 | """ |
| 103 | |
| 104 | source_commit: str |
| 105 | new_commit: str |
| 106 | factor: float |
| 107 | source_bpm: float |
| 108 | new_bpm: float |
| 109 | track: str |
| 110 | preserve_expressions: bool |
| 111 | message: str |
| 112 | |
| 113 | |
| 114 | # --------------------------------------------------------------------------- |
| 115 | # Pure helper — factor computation from BPM target |
| 116 | # --------------------------------------------------------------------------- |
| 117 | |
| 118 | |
| 119 | def compute_factor_from_bpm(source_bpm: float, target_bpm: float) -> float: |
| 120 | """Compute the scaling factor needed to reach *target_bpm* from *source_bpm*. |
| 121 | |
| 122 | Uses the relation: factor = target_bpm / source_bpm. A factor > 1 |
| 123 | compresses time (faster); < 1 stretches it (slower). |
| 124 | |
| 125 | Args: |
| 126 | source_bpm: Current tempo (must be > 0). |
| 127 | target_bpm: Desired tempo in BPM (must be > 0). |
| 128 | |
| 129 | Returns: |
| 130 | Scaling factor as a float. |
| 131 | |
| 132 | Raises: |
| 133 | ValueError: If either BPM value is non-positive. |
| 134 | """ |
| 135 | if source_bpm <= 0: |
| 136 | raise ValueError(f"source_bpm must be positive, got {source_bpm}") |
| 137 | if target_bpm <= 0: |
| 138 | raise ValueError(f"target_bpm must be positive, got {target_bpm}") |
| 139 | return target_bpm / source_bpm |
| 140 | |
| 141 | |
| 142 | def apply_factor(bpm: float, factor: float) -> float: |
| 143 | """Return the new BPM after applying *factor*. |
| 144 | |
| 145 | Args: |
| 146 | bpm: Source tempo in BPM. |
| 147 | factor: Scaling factor (> 0). |
| 148 | |
| 149 | Returns: |
| 150 | New tempo = ``bpm * factor``, rounded to four decimal places. |
| 151 | """ |
| 152 | return round(bpm * factor, 4) |
| 153 | |
| 154 | |
| 155 | # --------------------------------------------------------------------------- |
| 156 | # Testable async core |
| 157 | # --------------------------------------------------------------------------- |
| 158 | |
| 159 | |
| 160 | async def _tempo_scale_async( |
| 161 | *, |
| 162 | root: pathlib.Path, |
| 163 | session: AsyncSession, |
| 164 | commit: Optional[str], |
| 165 | factor: Optional[float], |
| 166 | bpm: Optional[float], |
| 167 | track: Optional[str], |
| 168 | preserve_expressions: bool, |
| 169 | message: Optional[str], |
| 170 | ) -> TempoScaleResult: |
| 171 | """Apply tempo scaling to a commit and return the operation result. |
| 172 | |
| 173 | This is a stub implementation that models the correct schema and |
| 174 | deterministic semantics. Full MIDI note manipulation will be wired in |
| 175 | when the Storpheus note-event query route is available. |
| 176 | |
| 177 | The scaling factor is resolved in this order: |
| 178 | 1. If *bpm* is given, compute factor = bpm / source_bpm. |
| 179 | 2. Otherwise use *factor* directly. |
| 180 | |
| 181 | Args: |
| 182 | root: Repository root (directory containing ``.muse/``). |
| 183 | session: Open async DB session (reserved for full impl). |
| 184 | commit: Source commit SHA; defaults to HEAD. |
| 185 | factor: Explicit scaling factor (``None`` when ``--bpm`` used). |
| 186 | bpm: Target BPM (``None`` when factor is used directly). |
| 187 | track: Restrict scaling to a named MIDI track, or ``None``. |
| 188 | preserve_expressions: Scale CC/expression events proportionally. |
| 189 | message: Commit message for the new commit. |
| 190 | |
| 191 | Returns: |
| 192 | A :class:`TempoScaleResult` describing the new tempo-scaled commit. |
| 193 | |
| 194 | Raises: |
| 195 | ValueError: If neither *factor* nor *bpm* is provided, or if the |
| 196 | computed factor is outside ``[FACTOR_MIN, FACTOR_MAX]``. |
| 197 | """ |
| 198 | muse_dir = root / ".muse" |
| 199 | head_ref = (muse_dir / "HEAD").read_text().strip() |
| 200 | branch = head_ref.rsplit("/", 1)[-1] if "/" in head_ref else head_ref |
| 201 | ref_path = muse_dir / pathlib.Path(head_ref) |
| 202 | head_sha = ref_path.read_text().strip() if ref_path.exists() else "0000000" |
| 203 | source_commit = commit or (head_sha[:8] if head_sha else "0000000") |
| 204 | |
| 205 | # Stub source BPM — full implementation queries tempo metadata from the commit. |
| 206 | stub_source_bpm = 120.0 |
| 207 | |
| 208 | # Resolve factor |
| 209 | if bpm is not None: |
| 210 | resolved_factor = compute_factor_from_bpm(stub_source_bpm, bpm) |
| 211 | elif factor is not None: |
| 212 | resolved_factor = factor |
| 213 | else: |
| 214 | raise ValueError("Either factor or --bpm must be provided.") |
| 215 | |
| 216 | if not (FACTOR_MIN <= resolved_factor <= FACTOR_MAX): |
| 217 | raise ValueError( |
| 218 | f"Computed factor {resolved_factor:.4f} is outside the allowed range " |
| 219 | f"[{FACTOR_MIN}, {FACTOR_MAX}]." |
| 220 | ) |
| 221 | |
| 222 | new_bpm = apply_factor(stub_source_bpm, resolved_factor) |
| 223 | |
| 224 | # Deterministic stub commit SHA — same inputs always produce the same hash. |
| 225 | # Full implementation persists note events and tempo metadata to the DB. |
| 226 | raw = f"{source_commit}:{resolved_factor}:{track or 'all'}:{preserve_expressions}" |
| 227 | new_commit = hashlib.sha1(raw.encode()).hexdigest()[:8] # noqa: S324 — not crypto |
| 228 | |
| 229 | resolved_message = message or f"tempo-scale {resolved_factor:.4f}x (stub)" |
| 230 | |
| 231 | logger.info( |
| 232 | "✅ tempo-scale: %s -> %s factor=%.4f %.1f->%.1f BPM track=%s", |
| 233 | source_commit, |
| 234 | new_commit, |
| 235 | resolved_factor, |
| 236 | stub_source_bpm, |
| 237 | new_bpm, |
| 238 | track or "all", |
| 239 | ) |
| 240 | |
| 241 | return TempoScaleResult( |
| 242 | source_commit=source_commit, |
| 243 | new_commit=new_commit, |
| 244 | factor=round(resolved_factor, 6), |
| 245 | source_bpm=stub_source_bpm, |
| 246 | new_bpm=new_bpm, |
| 247 | track=track or "all", |
| 248 | preserve_expressions=preserve_expressions, |
| 249 | message=resolved_message, |
| 250 | ) |
| 251 | |
| 252 | |
| 253 | # --------------------------------------------------------------------------- |
| 254 | # Output formatters |
| 255 | # --------------------------------------------------------------------------- |
| 256 | |
| 257 | |
| 258 | def _format_result(result: TempoScaleResult, *, as_json: bool) -> str: |
| 259 | """Render a TempoScaleResult as human-readable text or compact JSON. |
| 260 | |
| 261 | Args: |
| 262 | result: The tempo-scale operation result to render. |
| 263 | as_json: Emit compact JSON when True; ASCII summary when False. |
| 264 | |
| 265 | Returns: |
| 266 | Formatted string ready for ``typer.echo``. |
| 267 | """ |
| 268 | if as_json: |
| 269 | return json.dumps(dict(result), indent=2) |
| 270 | |
| 271 | factor = result["factor"] |
| 272 | if factor >= 1: |
| 273 | display = f"x{factor:.4f}" |
| 274 | else: |
| 275 | display = f"/{1.0 / factor:.4f}" |
| 276 | lines = [ |
| 277 | f"Tempo scaled: {result['source_commit']} -> {result['new_commit']}", |
| 278 | f" Factor: {factor:.4f} ({display})", |
| 279 | f" Tempo: {result['source_bpm']:.1f} BPM -> {result['new_bpm']:.1f} BPM", |
| 280 | f" Track: {result['track']}", |
| 281 | ] |
| 282 | if result["preserve_expressions"]: |
| 283 | lines.append(" Expressions: scaled proportionally") |
| 284 | lines.append(f" Message: {result['message']}") |
| 285 | lines.append(" (stub -- full MIDI note manipulation pending)") |
| 286 | return "\n".join(lines) |
| 287 | |
| 288 | |
| 289 | # --------------------------------------------------------------------------- |
| 290 | # Typer command |
| 291 | # --------------------------------------------------------------------------- |
| 292 | |
| 293 | |
| 294 | @app.callback(invoke_without_command=True) |
| 295 | def tempo_scale( |
| 296 | ctx: typer.Context, |
| 297 | factor: Annotated[ |
| 298 | Optional[float], |
| 299 | typer.Argument( |
| 300 | help=( |
| 301 | "Scaling factor: 0.5 = half-time (slower), 2.0 = double-time (faster). " |
| 302 | "Omit when using --bpm." |
| 303 | ), |
| 304 | show_default=False, |
| 305 | ), |
| 306 | ] = None, |
| 307 | commit: Annotated[ |
| 308 | Optional[str], |
| 309 | typer.Argument( |
| 310 | help="Source commit SHA to scale. Defaults to HEAD.", |
| 311 | show_default=False, |
| 312 | ), |
| 313 | ] = None, |
| 314 | bpm: Annotated[ |
| 315 | Optional[float], |
| 316 | typer.Option( |
| 317 | "--bpm", |
| 318 | metavar="N", |
| 319 | help=( |
| 320 | "Scale to reach exactly N BPM. " |
| 321 | "Computes factor = N / current_bpm. " |
| 322 | "Mutually exclusive with the <factor> argument." |
| 323 | ), |
| 324 | show_default=False, |
| 325 | ), |
| 326 | ] = None, |
| 327 | track: Annotated[ |
| 328 | Optional[str], |
| 329 | typer.Option( |
| 330 | "--track", |
| 331 | metavar="TEXT", |
| 332 | help="Scale only a specific MIDI track (useful for individual overdubs).", |
| 333 | show_default=False, |
| 334 | ), |
| 335 | ] = None, |
| 336 | preserve_expressions: Annotated[ |
| 337 | bool, |
| 338 | typer.Option( |
| 339 | "--preserve-expressions", |
| 340 | help="Preserve CC/expression event timing proportionally.", |
| 341 | ), |
| 342 | ] = False, |
| 343 | message: Annotated[ |
| 344 | Optional[str], |
| 345 | typer.Option( |
| 346 | "--message", |
| 347 | "-m", |
| 348 | metavar="TEXT", |
| 349 | help="Commit message for the new tempo-scaled commit.", |
| 350 | show_default=False, |
| 351 | ), |
| 352 | ] = None, |
| 353 | as_json: Annotated[ |
| 354 | bool, |
| 355 | typer.Option("--json", help="Emit machine-readable JSON output."), |
| 356 | ] = False, |
| 357 | ) -> None: |
| 358 | """Stretch or compress the timing of a commit. |
| 359 | |
| 360 | Scales all MIDI note onset/offset times by <factor>, updates tempo |
| 361 | metadata, and records a new commit. Pitch is preserved -- this is pure |
| 362 | MIDI timing manipulation, not audio time-stretching. |
| 363 | |
| 364 | Use ``--bpm N`` instead of <factor> to target an exact tempo. |
| 365 | """ |
| 366 | root = require_repo() |
| 367 | |
| 368 | # Validate: at least one of factor or --bpm must be given |
| 369 | if factor is None and bpm is None: |
| 370 | typer.echo("Provide either a <factor> argument or --bpm N.") |
| 371 | raise typer.Exit(code=ExitCode.USER_ERROR) |
| 372 | |
| 373 | # Validate: factor and --bpm are mutually exclusive |
| 374 | if factor is not None and bpm is not None: |
| 375 | typer.echo("<factor> and --bpm are mutually exclusive. Provide one or the other.") |
| 376 | raise typer.Exit(code=ExitCode.USER_ERROR) |
| 377 | |
| 378 | # Validate factor range when provided directly |
| 379 | if factor is not None and not (FACTOR_MIN <= factor <= FACTOR_MAX): |
| 380 | typer.echo( |
| 381 | f"Factor {factor!r} is outside the allowed range " |
| 382 | f"[{FACTOR_MIN}, {FACTOR_MAX}]." |
| 383 | ) |
| 384 | raise typer.Exit(code=ExitCode.USER_ERROR) |
| 385 | |
| 386 | # Validate bpm when provided |
| 387 | if bpm is not None and bpm <= 0: |
| 388 | typer.echo(f"--bpm must be a positive number, got {bpm!r}.") |
| 389 | raise typer.Exit(code=ExitCode.USER_ERROR) |
| 390 | |
| 391 | async def _run() -> None: |
| 392 | async with open_session() as session: |
| 393 | result = await _tempo_scale_async( |
| 394 | root=root, |
| 395 | session=session, |
| 396 | commit=commit, |
| 397 | factor=factor, |
| 398 | bpm=bpm, |
| 399 | track=track, |
| 400 | preserve_expressions=preserve_expressions, |
| 401 | message=message, |
| 402 | ) |
| 403 | typer.echo(_format_result(result, as_json=as_json)) |
| 404 | |
| 405 | try: |
| 406 | asyncio.run(_run()) |
| 407 | except typer.Exit: |
| 408 | raise |
| 409 | except ValueError as exc: |
| 410 | typer.echo(str(exc)) |
| 411 | raise typer.Exit(code=ExitCode.USER_ERROR) |
| 412 | except Exception as exc: |
| 413 | typer.echo(f"muse tempo-scale failed: {exc}") |
| 414 | logger.error("muse tempo-scale error: %s", exc, exc_info=True) |
| 415 | raise typer.Exit(code=ExitCode.INTERNAL_ERROR) |