cgcardona / muse public
transpose.py python
410 lines 14.8 KB
Raw
12901c5a Initial extraction from tellurstori/maestro cgcardona <gabriel@tellurstori.com> 4d ago
1 """muse transpose — apply MIDI pitch transposition as a Muse commit.
2
3 Transposes all MIDI files in ``muse-work/`` by the given interval and records
4 the result as a new Muse commit. Drum channels (MIDI channel 9) are always
5 excluded from pitch transposition because drums are unpitched.
6
7 Usage
8 -----
9 ::
10
11 muse transpose +3 # up 3 semitones from HEAD
12 muse transpose -5 # down 5 semitones from HEAD
13 muse transpose up-minor3rd # named interval
14 muse transpose down-perfect5th --track melody # single-track scope
15 muse transpose +2 --section chorus # section scope (stub)
16 muse transpose +3 --dry-run # preview without committing
17 muse transpose +3 --json # machine-readable result
18 muse transpose +2 <commit> # transpose from a named commit
19
20 Interval syntax
21 ---------------
22 - Signed integers: ``+3``, ``-5``, ``+12``
23 - Named intervals: ``up-minor3rd``, ``down-perfect5th``, ``up-octave``
24
25 Named interval identifiers
26 --------------------------
27 unison, minor2nd, major2nd, minor3rd, major3rd, perfect4th,
28 perfect5th, minor6th, major6th, minor7th, major7th, octave
29 (prefix with ``up-`` or ``down-``)
30
31 Key metadata
32 ------------
33 If the source commit has a ``key`` field in its ``metadata`` JSON blob
34 (e.g. set via a future ``muse key --set`` command), the new commit's
35 ``metadata.key`` is automatically updated to reflect the transposition
36 (e.g. ``"Eb major"`` + 2 semitones → ``"F major"``).
37 """
38 from __future__ import annotations
39
40 import asyncio
41 import datetime
42 import json
43 import logging
44 import pathlib
45
46 import typer
47 from sqlalchemy.ext.asyncio import AsyncSession
48
49 from maestro.muse_cli._repo import require_repo
50 from maestro.muse_cli.db import (
51 get_head_snapshot_id,
52 insert_commit,
53 open_session,
54 resolve_commit_ref,
55 upsert_object,
56 upsert_snapshot,
57 )
58 from maestro.muse_cli.errors import ExitCode
59 from maestro.muse_cli.models import MuseCliCommit
60 from maestro.muse_cli.snapshot import (
61 build_snapshot_manifest,
62 compute_commit_id,
63 compute_snapshot_id,
64 )
65 from maestro.services.muse_transpose import (
66 TransposeResult,
67 apply_transpose_to_workdir,
68 parse_interval,
69 update_key_metadata,
70 )
71
72 logger = logging.getLogger(__name__)
73
74 app = typer.Typer()
75
76
77 # ---------------------------------------------------------------------------
78 # Repo context helper (shared with tempo.py pattern)
79 # ---------------------------------------------------------------------------
80
81
82 def _read_repo_context(root: pathlib.Path) -> tuple[str, str, str]:
83 """Return ``(repo_id, branch, head_commit_id_or_empty)`` from ``.muse/``."""
84 muse_dir = root / ".muse"
85 repo_data: dict[str, str] = json.loads((muse_dir / "repo.json").read_text())
86 repo_id = repo_data["repo_id"]
87 head_ref = (muse_dir / "HEAD").read_text().strip()
88 branch = head_ref.rsplit("/", 1)[-1]
89 ref_path = muse_dir / pathlib.Path(head_ref)
90 head_commit_id = ref_path.read_text().strip() if ref_path.exists() else ""
91 return repo_id, branch, head_commit_id
92
93
94 # ---------------------------------------------------------------------------
95 # Testable async core
96 # ---------------------------------------------------------------------------
97
98
99 async def _transpose_async(
100 *,
101 root: pathlib.Path,
102 session: AsyncSession,
103 semitones: int,
104 commit_ref: str | None,
105 track_filter: str | None,
106 section_filter: str | None,
107 message: str | None,
108 dry_run: bool,
109 as_json: bool,
110 ) -> TransposeResult:
111 """Apply transposition and optionally commit the result.
112
113 This is the injectable async core used by tests and the Typer callback.
114 All filesystem and DB side-effects are isolated here so tests can inject
115 an in-memory SQLite session and a ``tmp_path`` root.
116
117 Workflow:
118 1. Resolve source commit (default HEAD).
119 2. Extract ``metadata.key`` from the source commit (if any).
120 3. Apply transposition to all MIDI files in ``muse-work/``.
121 4. Unless ``--dry-run``, create a new Muse commit and update HEAD.
122 5. Annotate the new commit with updated key metadata.
123 6. Return a ``TransposeResult`` and print human/JSON output.
124
125 Args:
126 root: Repository root (directory containing ``.muse/``).
127 session: Open async DB session; committed by caller.
128 semitones: Signed semitone offset to apply.
129 commit_ref: Commit SHA or ref to transpose from, or ``None`` for HEAD.
130 track_filter: Case-insensitive track name substring filter, or ``None``.
131 section_filter: Section name filter (stub — logged as not implemented).
132 message: Custom commit message, or ``None`` for auto-generated.
133 dry_run: When True, do not write files or create a commit.
134 as_json: When True, emit JSON output instead of human text.
135
136 Returns:
137 ``TransposeResult`` with all fields populated.
138
139 Raises:
140 ``typer.Exit``: On user errors (missing commit, parse failure) or
141 internal errors (DB failure, I/O error).
142 """
143 repo_id, branch, _ = _read_repo_context(root)
144
145 # ── Resolve source commit ─────────────────────────────────────────────
146 commit = await resolve_commit_ref(session, repo_id, branch, commit_ref)
147 if commit is None:
148 ref_label = commit_ref or "HEAD"
149 typer.echo(f"❌ No commit found for ref '{ref_label}'")
150 raise typer.Exit(code=ExitCode.USER_ERROR)
151
152 source_commit_id = commit.commit_id
153
154 # ── Extract existing key metadata ─────────────────────────────────────
155 meta: dict[str, object] = dict(commit.commit_metadata or {})
156 original_key: str | None = None
157 new_key: str | None = None
158 key_raw = meta.get("key")
159 if isinstance(key_raw, str) and key_raw:
160 original_key = key_raw
161 new_key = update_key_metadata(original_key, semitones)
162 logger.debug("✅ Key: %r → %r", original_key, new_key)
163
164 # ── Apply transposition to muse-work/ ────────────────────────────────
165 workdir = root / "muse-work"
166 files_modified, files_skipped = apply_transpose_to_workdir(
167 workdir=workdir,
168 semitones=semitones,
169 track_filter=track_filter,
170 section_filter=section_filter,
171 dry_run=dry_run,
172 )
173
174 if dry_run:
175 result = TransposeResult(
176 source_commit_id=source_commit_id,
177 semitones=semitones,
178 files_modified=files_modified,
179 files_skipped=files_skipped,
180 new_commit_id=None,
181 original_key=original_key,
182 new_key=new_key,
183 dry_run=True,
184 )
185 _print_result(result, as_json=as_json)
186 return result
187
188 if not files_modified:
189 typer.echo(
190 "⚠️ No MIDI files were modified. "
191 "Check that muse-work/ contains .mid files and the interval is non-zero."
192 )
193 raise typer.Exit(code=ExitCode.USER_ERROR)
194
195 # ── Build snapshot from modified workdir ──────────────────────────────
196 if not workdir.exists():
197 typer.echo("❌ muse-work/ directory not found — cannot create commit")
198 raise typer.Exit(code=ExitCode.USER_ERROR)
199
200 manifest = build_snapshot_manifest(workdir)
201 if not manifest:
202 typer.echo("❌ muse-work/ is empty — nothing to commit")
203 raise typer.Exit(code=ExitCode.USER_ERROR)
204
205 snapshot_id = compute_snapshot_id(manifest)
206
207 # Guard: nothing changed (shouldn't happen, but be safe)
208 last_snapshot_id = await get_head_snapshot_id(session, repo_id, branch)
209 if last_snapshot_id == snapshot_id:
210 typer.echo("Nothing to commit — working tree unchanged after transposition")
211 raise typer.Exit(code=ExitCode.SUCCESS)
212
213 # ── Persist objects ───────────────────────────────────────────────────
214 for rel_path, object_id in manifest.items():
215 file_path = workdir / rel_path
216 size = file_path.stat().st_size
217 await upsert_object(session, object_id=object_id, size_bytes=size)
218
219 # ── Persist snapshot ──────────────────────────────────────────────────
220 await upsert_snapshot(session, manifest=manifest, snapshot_id=snapshot_id)
221 await session.flush()
222
223 # ── Compute and persist commit ────────────────────────────────────────
224 committed_at = datetime.datetime.now(datetime.timezone.utc)
225 interval_label = f"{semitones:+d} semitones" if semitones != 0 else "0 semitones"
226 effective_message = message or f"Transpose {interval_label}"
227 commit_metadata: dict[str, object] = dict(meta)
228 if new_key is not None:
229 commit_metadata["key"] = new_key
230
231 parent_commit_id = source_commit_id
232 new_commit_id = compute_commit_id(
233 parent_ids=[parent_commit_id],
234 snapshot_id=snapshot_id,
235 message=effective_message,
236 committed_at_iso=committed_at.isoformat(),
237 )
238
239 new_commit = MuseCliCommit(
240 commit_id=new_commit_id,
241 repo_id=repo_id,
242 branch=branch,
243 parent_commit_id=parent_commit_id,
244 snapshot_id=snapshot_id,
245 message=effective_message,
246 author="",
247 committed_at=committed_at,
248 commit_metadata=commit_metadata if commit_metadata else None,
249 )
250 await insert_commit(session, new_commit)
251
252 # ── Update branch HEAD pointer ────────────────────────────────────────
253 muse_dir = root / ".muse"
254 head_ref = (muse_dir / "HEAD").read_text().strip()
255 ref_path = muse_dir / pathlib.Path(head_ref)
256 ref_path.parent.mkdir(parents=True, exist_ok=True)
257 ref_path.write_text(new_commit_id)
258
259 result = TransposeResult(
260 source_commit_id=source_commit_id,
261 semitones=semitones,
262 files_modified=files_modified,
263 files_skipped=files_skipped,
264 new_commit_id=new_commit_id,
265 original_key=original_key,
266 new_key=new_key,
267 dry_run=False,
268 )
269 _print_result(result, as_json=as_json)
270 return result
271
272
273 # ---------------------------------------------------------------------------
274 # Renderers
275 # ---------------------------------------------------------------------------
276
277
278 def _print_result(result: TransposeResult, *, as_json: bool) -> None:
279 """Render a TransposeResult as human-readable text or JSON."""
280 if as_json:
281 typer.echo(
282 json.dumps(
283 {
284 "source_commit_id": result.source_commit_id,
285 "semitones": result.semitones,
286 "files_modified": result.files_modified,
287 "files_skipped": result.files_skipped,
288 "new_commit_id": result.new_commit_id,
289 "original_key": result.original_key,
290 "new_key": result.new_key,
291 "dry_run": result.dry_run,
292 },
293 indent=2,
294 )
295 )
296 return
297
298 prefix = "DRY RUN" if result.dry_run else ""
299 if result.new_commit_id:
300 typer.echo(
301 f"✅ {prefix}[{result.new_commit_id[:8]}] Transpose {result.semitones:+d} semitones"
302 )
303 else:
304 typer.echo(f"{prefix}Transpose {result.semitones:+d} semitones")
305
306 if result.original_key and result.new_key:
307 typer.echo(f" Key: {result.original_key} → {result.new_key}")
308
309 typer.echo(f" Modified: {len(result.files_modified)} file(s)")
310 for f in result.files_modified:
311 typer.echo(f" ✅ {f}")
312
313 if result.files_skipped:
314 typer.echo(f" Skipped: {len(result.files_skipped)} file(s) (non-MIDI or no pitched notes)")
315
316 if result.dry_run:
317 typer.echo(" (dry-run: no files written, no commit created)")
318
319
320 # ---------------------------------------------------------------------------
321 # Typer command
322 # ---------------------------------------------------------------------------
323
324
325 @app.callback(invoke_without_command=True)
326 def transpose(
327 ctx: typer.Context,
328 interval: str = typer.Argument(
329 ...,
330 metavar="<interval>",
331 help=(
332 "Interval to transpose by. "
333 "Signed integer (+3, -5) or named interval (up-minor3rd, down-perfect5th)."
334 ),
335 ),
336 commit_ref: str | None = typer.Argument(
337 None,
338 metavar="[<commit>]",
339 help="Source commit SHA or 'HEAD' (default: HEAD).",
340 ),
341 track: str | None = typer.Option(
342 None,
343 "--track",
344 metavar="TEXT",
345 help="Transpose only the MIDI track whose name contains TEXT (case-insensitive).",
346 ),
347 section: str | None = typer.Option(
348 None,
349 "--section",
350 metavar="TEXT",
351 help="Transpose only the named section (stub — full implementation pending).",
352 ),
353 message: str | None = typer.Option(
354 None,
355 "--message",
356 "-m",
357 metavar="TEXT",
358 help="Custom commit message (default: 'Transpose +N semitones').",
359 ),
360 dry_run: bool = typer.Option(
361 False,
362 "--dry-run",
363 help="Show what would change without writing files or creating a commit.",
364 ),
365 as_json: bool = typer.Option(
366 False,
367 "--json",
368 help="Emit machine-readable JSON output.",
369 ),
370 ) -> None:
371 """Apply MIDI pitch transposition and record it as a Muse commit.
372
373 Transposes all MIDI files in ``muse-work/`` by the given interval,
374 then creates a new commit capturing the transposed snapshot. Drum
375 channels (MIDI channel 9) are always excluded.
376
377 Use ``--dry-run`` to preview what would change without committing.
378 Use ``--track`` to restrict transposition to a specific instrument track.
379 """
380 # Parse interval first — fail fast before touching the repo
381 try:
382 semitones = parse_interval(interval)
383 except ValueError as exc:
384 typer.echo(f"❌ {exc}")
385 raise typer.Exit(code=ExitCode.USER_ERROR)
386
387 root = require_repo()
388
389 async def _run() -> None:
390 async with open_session() as session:
391 await _transpose_async(
392 root=root,
393 session=session,
394 semitones=semitones,
395 commit_ref=commit_ref,
396 track_filter=track,
397 section_filter=section,
398 message=message,
399 dry_run=dry_run,
400 as_json=as_json,
401 )
402
403 try:
404 asyncio.run(_run())
405 except typer.Exit:
406 raise
407 except Exception as exc:
408 typer.echo(f"❌ muse transpose failed: {exc}")
409 logger.error("❌ muse transpose error: %s", exc, exc_info=True)
410 raise typer.Exit(code=ExitCode.INTERNAL_ERROR)