tempo.py
python
| 1 | """muse tempo — read or set the tempo (BPM) of a commit. |
| 2 | |
| 3 | Usage |
| 4 | ----- |
| 5 | :: |
| 6 | |
| 7 | muse tempo [<commit>] # read tempo from HEAD or named commit |
| 8 | muse tempo --set 128 # annotate HEAD with explicit BPM |
| 9 | muse tempo --set 128 <commit> # annotate a named commit |
| 10 | muse tempo --history # show BPM across all commits |
| 11 | muse tempo --json # machine-readable JSON output |
| 12 | |
| 13 | Tempo resolution order (read path) |
| 14 | ----------------------------------- |
| 15 | 1. Explicit annotation stored via ``muse tempo --set`` (``metadata.tempo_bpm``). |
| 16 | 2. Auto-detection from MIDI Set Tempo events in the commit's snapshot. |
| 17 | 3. ``None`` (displayed as ``--`` in table output) when neither is available. |
| 18 | |
| 19 | Tempo storage (write path) |
| 20 | --------------------------- |
| 21 | ``--set`` writes ``{"tempo_bpm": <float>}`` into the ``metadata`` JSON column |
| 22 | of the target commit row. Other metadata keys are preserved. No new DB rows |
| 23 | are created — only the existing commit is annotated. |
| 24 | |
| 25 | History traversal |
| 26 | ----------------- |
| 27 | ``--history`` walks the full parent chain from HEAD (or the named commit), |
| 28 | using only explicitly annotated values (``metadata.tempo_bpm``). Auto-detected |
| 29 | BPM is shown on the single-commit read path but is not persisted, so it cannot |
| 30 | appear in history. |
| 31 | """ |
| 32 | from __future__ import annotations |
| 33 | |
| 34 | import asyncio |
| 35 | import json |
| 36 | import logging |
| 37 | import pathlib |
| 38 | from typing import Optional |
| 39 | |
| 40 | import typer |
| 41 | from sqlalchemy.ext.asyncio import AsyncSession |
| 42 | |
| 43 | from maestro.muse_cli._repo import require_repo |
| 44 | from maestro.muse_cli.db import ( |
| 45 | open_session, |
| 46 | resolve_commit_ref, |
| 47 | set_commit_tempo_bpm, |
| 48 | ) |
| 49 | from maestro.muse_cli.errors import ExitCode |
| 50 | from maestro.muse_cli.models import MuseCliCommit |
| 51 | from maestro.services.muse_tempo import ( |
| 52 | MuseTempoHistoryEntry, |
| 53 | MuseTempoResult, |
| 54 | build_tempo_history, |
| 55 | detect_tempo_from_snapshot, |
| 56 | ) |
| 57 | |
| 58 | logger = logging.getLogger(__name__) |
| 59 | |
| 60 | app = typer.Typer() |
| 61 | |
| 62 | _BPM_MIN = 20.0 |
| 63 | _BPM_MAX = 400.0 |
| 64 | |
| 65 | |
| 66 | # --------------------------------------------------------------------------- |
| 67 | # Repo context helpers |
| 68 | # --------------------------------------------------------------------------- |
| 69 | |
| 70 | |
| 71 | def _read_repo_context(root: pathlib.Path) -> tuple[str, str, str]: |
| 72 | """Return (repo_id, branch, head_commit_id_or_empty) from .muse/.""" |
| 73 | import json as _json |
| 74 | |
| 75 | muse_dir = root / ".muse" |
| 76 | repo_data: dict[str, str] = _json.loads((muse_dir / "repo.json").read_text()) |
| 77 | repo_id = repo_data["repo_id"] |
| 78 | head_ref = (muse_dir / "HEAD").read_text().strip() |
| 79 | branch = head_ref.rsplit("/", 1)[-1] |
| 80 | ref_path = muse_dir / pathlib.Path(head_ref) |
| 81 | head_commit_id = ref_path.read_text().strip() if ref_path.exists() else "" |
| 82 | return repo_id, branch, head_commit_id |
| 83 | |
| 84 | |
| 85 | # --------------------------------------------------------------------------- |
| 86 | # Testable async core |
| 87 | # --------------------------------------------------------------------------- |
| 88 | |
| 89 | |
| 90 | async def _load_commit_chain( |
| 91 | session: AsyncSession, |
| 92 | head_commit_id: str, |
| 93 | limit: int = 1000, |
| 94 | ) -> list[MuseCliCommit]: |
| 95 | """Walk the parent chain from *head_commit_id*, returning newest-first.""" |
| 96 | commits: list[MuseCliCommit] = [] |
| 97 | current_id: str | None = head_commit_id |
| 98 | while current_id and len(commits) < limit: |
| 99 | commit = await session.get(MuseCliCommit, current_id) |
| 100 | if commit is None: |
| 101 | logger.warning("⚠️ Commit %s not found — chain broken", current_id[:8]) |
| 102 | break |
| 103 | commits.append(commit) |
| 104 | current_id = commit.parent_commit_id |
| 105 | return commits |
| 106 | |
| 107 | |
| 108 | async def _tempo_read_async( |
| 109 | *, |
| 110 | root: pathlib.Path, |
| 111 | session: AsyncSession, |
| 112 | commit_ref: str | None, |
| 113 | as_json: bool, |
| 114 | ) -> MuseTempoResult: |
| 115 | """Load a commit and return its tempo result. |
| 116 | |
| 117 | Reads the annotated BPM from ``metadata.tempo_bpm``. If absent, scans |
| 118 | MIDI files in the commit's snapshot for a Set Tempo event. |
| 119 | """ |
| 120 | from maestro.muse_cli.db import get_commit_snapshot_manifest |
| 121 | |
| 122 | repo_id, branch, _ = _read_repo_context(root) |
| 123 | commit = await resolve_commit_ref(session, repo_id, branch, commit_ref) |
| 124 | if commit is None: |
| 125 | ref_label = commit_ref or "HEAD" |
| 126 | typer.echo(f"❌ No commit found for ref '{ref_label}'") |
| 127 | raise typer.Exit(code=ExitCode.USER_ERROR) |
| 128 | |
| 129 | meta: dict[str, object] = commit.commit_metadata or {} |
| 130 | bpm_raw = meta.get("tempo_bpm") |
| 131 | annotated_bpm: float | None = float(bpm_raw) if isinstance(bpm_raw, (int, float)) else None |
| 132 | |
| 133 | # Auto-detect from MIDI files in snapshot |
| 134 | detected_bpm: float | None = None |
| 135 | manifest = await get_commit_snapshot_manifest(session, commit.commit_id) |
| 136 | if manifest: |
| 137 | workdir = root / "muse-work" |
| 138 | detected_bpm = detect_tempo_from_snapshot(manifest, workdir) |
| 139 | |
| 140 | result = MuseTempoResult( |
| 141 | commit_id=commit.commit_id, |
| 142 | branch=branch, |
| 143 | message=commit.message, |
| 144 | tempo_bpm=annotated_bpm, |
| 145 | detected_bpm=detected_bpm, |
| 146 | ) |
| 147 | |
| 148 | if as_json: |
| 149 | _print_result_json(result) |
| 150 | else: |
| 151 | _print_result_human(result) |
| 152 | |
| 153 | return result |
| 154 | |
| 155 | |
| 156 | async def _tempo_set_async( |
| 157 | *, |
| 158 | root: pathlib.Path, |
| 159 | session: AsyncSession, |
| 160 | commit_ref: str | None, |
| 161 | bpm: float, |
| 162 | ) -> None: |
| 163 | """Annotate a commit with an explicit BPM.""" |
| 164 | repo_id, branch, _ = _read_repo_context(root) |
| 165 | commit = await resolve_commit_ref(session, repo_id, branch, commit_ref) |
| 166 | if commit is None: |
| 167 | ref_label = commit_ref or "HEAD" |
| 168 | typer.echo(f"❌ No commit found for ref '{ref_label}'") |
| 169 | raise typer.Exit(code=ExitCode.USER_ERROR) |
| 170 | |
| 171 | updated = await set_commit_tempo_bpm(session, commit.commit_id, bpm) |
| 172 | if updated is None: |
| 173 | typer.echo(f"❌ Could not update commit {commit.commit_id[:8]}") |
| 174 | raise typer.Exit(code=ExitCode.INTERNAL_ERROR) |
| 175 | |
| 176 | typer.echo(f"✅ Set tempo {bpm:.1f} BPM on commit {commit.commit_id[:8]} ({commit.message})") |
| 177 | |
| 178 | |
| 179 | async def _tempo_history_async( |
| 180 | *, |
| 181 | root: pathlib.Path, |
| 182 | session: AsyncSession, |
| 183 | commit_ref: str | None, |
| 184 | as_json: bool, |
| 185 | ) -> list[MuseTempoHistoryEntry]: |
| 186 | """Walk parent chain and return a tempo history list.""" |
| 187 | repo_id, branch, _ = _read_repo_context(root) |
| 188 | commit = await resolve_commit_ref(session, repo_id, branch, commit_ref) |
| 189 | if commit is None: |
| 190 | ref_label = commit_ref or "HEAD" |
| 191 | typer.echo(f"❌ No commit found for ref '{ref_label}'") |
| 192 | raise typer.Exit(code=ExitCode.USER_ERROR) |
| 193 | |
| 194 | chain = await _load_commit_chain(session, commit.commit_id) |
| 195 | history = build_tempo_history(chain) |
| 196 | |
| 197 | if as_json: |
| 198 | _print_history_json(history) |
| 199 | else: |
| 200 | _print_history_human(history) |
| 201 | |
| 202 | return history |
| 203 | |
| 204 | |
| 205 | # --------------------------------------------------------------------------- |
| 206 | # Renderers |
| 207 | # --------------------------------------------------------------------------- |
| 208 | |
| 209 | |
| 210 | def _bpm_str(bpm: float | None) -> str: |
| 211 | return f"{bpm:.1f}" if bpm is not None else "--" |
| 212 | |
| 213 | |
| 214 | def _print_result_human(result: MuseTempoResult) -> None: |
| 215 | typer.echo(f"commit {result.commit_id}") |
| 216 | typer.echo(f"branch {result.branch}") |
| 217 | typer.echo(f"message {result.message}") |
| 218 | typer.echo("") |
| 219 | if result.tempo_bpm is not None: |
| 220 | typer.echo(f"tempo {result.tempo_bpm:.1f} BPM (annotated)") |
| 221 | elif result.detected_bpm is not None: |
| 222 | typer.echo(f"tempo {result.detected_bpm:.1f} BPM (detected from MIDI)") |
| 223 | else: |
| 224 | typer.echo("tempo -- (no annotation; no MIDI tempo event found)") |
| 225 | |
| 226 | |
| 227 | def _print_result_json(result: MuseTempoResult) -> None: |
| 228 | typer.echo( |
| 229 | json.dumps( |
| 230 | { |
| 231 | "commit_id": result.commit_id, |
| 232 | "branch": result.branch, |
| 233 | "message": result.message, |
| 234 | "tempo_bpm": result.tempo_bpm, |
| 235 | "detected_bpm": result.detected_bpm, |
| 236 | "effective_bpm": result.effective_bpm, |
| 237 | }, |
| 238 | indent=2, |
| 239 | ) |
| 240 | ) |
| 241 | |
| 242 | |
| 243 | def _print_history_human(history: list[MuseTempoHistoryEntry]) -> None: |
| 244 | if not history: |
| 245 | typer.echo("No commits in history.") |
| 246 | return |
| 247 | header = f"{'COMMIT':<10} {'BPM':>7} {'DELTA':>7} MESSAGE" |
| 248 | typer.echo(header) |
| 249 | typer.echo("-" * len(header)) |
| 250 | for entry in history: |
| 251 | short_id = entry.commit_id[:8] |
| 252 | bpm_col = _bpm_str(entry.effective_bpm) |
| 253 | if entry.delta_bpm is None: |
| 254 | delta_col = " --" |
| 255 | elif entry.delta_bpm > 0: |
| 256 | delta_col = f"+{entry.delta_bpm:.1f}" |
| 257 | else: |
| 258 | delta_col = f"{entry.delta_bpm:.1f}" |
| 259 | typer.echo(f"{short_id:<10} {bpm_col:>7} {delta_col:>7} {entry.message}") |
| 260 | |
| 261 | |
| 262 | def _print_history_json(history: list[MuseTempoHistoryEntry]) -> None: |
| 263 | rows = [ |
| 264 | { |
| 265 | "commit_id": e.commit_id, |
| 266 | "message": e.message, |
| 267 | "effective_bpm": e.effective_bpm, |
| 268 | "delta_bpm": e.delta_bpm, |
| 269 | } |
| 270 | for e in history |
| 271 | ] |
| 272 | typer.echo(json.dumps(rows, indent=2)) |
| 273 | |
| 274 | |
| 275 | # --------------------------------------------------------------------------- |
| 276 | # Typer command |
| 277 | # --------------------------------------------------------------------------- |
| 278 | |
| 279 | |
| 280 | @app.callback(invoke_without_command=True) |
| 281 | def tempo( |
| 282 | ctx: typer.Context, |
| 283 | commit_ref: Optional[str] = typer.Argument( |
| 284 | None, |
| 285 | metavar="<commit>", |
| 286 | help="Commit SHA (full or abbreviated) or 'HEAD' (default).", |
| 287 | ), |
| 288 | set_bpm: Optional[float] = typer.Option( |
| 289 | None, |
| 290 | "--set", |
| 291 | metavar="<bpm>", |
| 292 | help=f"Annotate the commit with this BPM ({_BPM_MIN}–{_BPM_MAX}).", |
| 293 | ), |
| 294 | history: bool = typer.Option( |
| 295 | False, |
| 296 | "--history", |
| 297 | help="Show tempo changes across all commits (newest first).", |
| 298 | ), |
| 299 | as_json: bool = typer.Option( |
| 300 | False, |
| 301 | "--json", |
| 302 | help="Emit machine-readable JSON instead of human-readable text.", |
| 303 | ), |
| 304 | ) -> None: |
| 305 | """Read or set the tempo (BPM) of a commit. |
| 306 | |
| 307 | Without flags, prints the BPM for the target commit. Use ``--set`` |
| 308 | to annotate a commit with an explicit BPM. Use ``--history`` to show |
| 309 | the BPM timeline across the full parent chain. |
| 310 | """ |
| 311 | root = require_repo() |
| 312 | |
| 313 | if set_bpm is not None: |
| 314 | if not (_BPM_MIN <= set_bpm <= _BPM_MAX): |
| 315 | typer.echo(f"❌ BPM must be between {_BPM_MIN} and {_BPM_MAX} (got {set_bpm})") |
| 316 | raise typer.Exit(code=ExitCode.USER_ERROR) |
| 317 | |
| 318 | async def _run_set() -> None: |
| 319 | async with open_session() as session: |
| 320 | await _tempo_set_async( |
| 321 | root=root, |
| 322 | session=session, |
| 323 | commit_ref=commit_ref, |
| 324 | bpm=set_bpm, |
| 325 | ) |
| 326 | |
| 327 | try: |
| 328 | asyncio.run(_run_set()) |
| 329 | except typer.Exit: |
| 330 | raise |
| 331 | except Exception as exc: |
| 332 | typer.echo(f"❌ muse tempo --set failed: {exc}") |
| 333 | logger.error("❌ muse tempo --set error: %s", exc, exc_info=True) |
| 334 | raise typer.Exit(code=ExitCode.INTERNAL_ERROR) |
| 335 | return |
| 336 | |
| 337 | if history: |
| 338 | async def _run_history() -> None: |
| 339 | async with open_session() as session: |
| 340 | await _tempo_history_async( |
| 341 | root=root, |
| 342 | session=session, |
| 343 | commit_ref=commit_ref, |
| 344 | as_json=as_json, |
| 345 | ) |
| 346 | |
| 347 | try: |
| 348 | asyncio.run(_run_history()) |
| 349 | except typer.Exit: |
| 350 | raise |
| 351 | except Exception as exc: |
| 352 | typer.echo(f"❌ muse tempo --history failed: {exc}") |
| 353 | logger.error("❌ muse tempo --history error: %s", exc, exc_info=True) |
| 354 | raise typer.Exit(code=ExitCode.INTERNAL_ERROR) |
| 355 | return |
| 356 | |
| 357 | # Default: read path |
| 358 | async def _run_read() -> None: |
| 359 | async with open_session() as session: |
| 360 | await _tempo_read_async( |
| 361 | root=root, |
| 362 | session=session, |
| 363 | commit_ref=commit_ref, |
| 364 | as_json=as_json, |
| 365 | ) |
| 366 | |
| 367 | try: |
| 368 | asyncio.run(_run_read()) |
| 369 | except typer.Exit: |
| 370 | raise |
| 371 | except Exception as exc: |
| 372 | typer.echo(f"❌ muse tempo failed: {exc}") |
| 373 | logger.error("❌ muse tempo error: %s", exc, exc_info=True) |
| 374 | raise typer.Exit(code=ExitCode.INTERNAL_ERROR) |