chord_map.py
python
| 1 | """muse chord-map — visualize the chord progression embedded in a commit. |
| 2 | |
| 3 | Extracts and displays the chord timeline of a specific commit, showing |
| 4 | when each chord occurs in the arrangement. This gives an AI agent a |
| 5 | precise picture of the harmonic structure at any commit — not just |
| 6 | *which* chords are present, but *where exactly* each chord falls. |
| 7 | |
| 8 | Output (default text, ``--bar-grid``):: |
| 9 | |
| 10 | Chord map — commit a1b2c3d4 (HEAD -> main) |
| 11 | |
| 12 | Bar 1: Cmaj9 ████████ |
| 13 | Bar 2: Am11 ████████ |
| 14 | Bar 3: Dm7 ████ Gsus4 ████ |
| 15 | Bar 4: G7 ████████ |
| 16 | Bar 5: Cmaj9 ████████ |
| 17 | |
| 18 | With ``--voice-leading``:: |
| 19 | |
| 20 | Chord map — commit a1b2c3d4 (HEAD -> main) |
| 21 | |
| 22 | Bar 1: Cmaj9 → Am11 (E→E, G→G, B→A, D→C) |
| 23 | Bar 2: Am11 → Dm7 (A→D, C→C, E→F, G→A) |
| 24 | ... |
| 25 | |
| 26 | Flags |
| 27 | ----- |
| 28 | ``COMMIT`` Commit ref to analyse (default: HEAD). |
| 29 | ``--section TEXT`` Scope to a named section/region. |
| 30 | ``--track TEXT`` Scope to a specific track (e.g. piano for chord voicings). |
| 31 | ``--bar-grid`` Align chord events to musical bar numbers (default: on). |
| 32 | ``--format FORMAT`` Output format: ``text`` (default), ``json``, or ``mermaid``. |
| 33 | ``--voice-leading`` Show how individual notes move between consecutive chords. |
| 34 | """ |
| 35 | from __future__ import annotations |
| 36 | |
| 37 | import asyncio |
| 38 | import json |
| 39 | import logging |
| 40 | import pathlib |
| 41 | from typing import Optional |
| 42 | |
| 43 | import typer |
| 44 | from sqlalchemy.ext.asyncio import AsyncSession |
| 45 | from typing_extensions import Annotated, TypedDict |
| 46 | |
| 47 | from maestro.muse_cli._repo import require_repo |
| 48 | from maestro.muse_cli.db import open_session |
| 49 | from maestro.muse_cli.errors import ExitCode |
| 50 | |
| 51 | logger = logging.getLogger(__name__) |
| 52 | |
| 53 | app = typer.Typer() |
| 54 | |
| 55 | # --------------------------------------------------------------------------- |
| 56 | # Named result types (registered in docs/reference/type_contracts.md) |
| 57 | # --------------------------------------------------------------------------- |
| 58 | |
| 59 | |
| 60 | class ChordEvent(TypedDict): |
| 61 | """A single chord occurrence in the arrangement timeline. |
| 62 | |
| 63 | Fields: |
| 64 | bar: Musical bar number (1-indexed, or 0 if not bar-aligned). |
| 65 | beat: Beat within the bar (1-indexed). |
| 66 | chord: Chord symbol, e.g. ``"Cmaj9"``, ``"Am11"``, ``"G7"``. |
| 67 | duration: Duration in bars (fractional for chords shorter than one bar). |
| 68 | track: Track/instrument the chord belongs to (or ``"all"``). |
| 69 | """ |
| 70 | |
| 71 | bar: int |
| 72 | beat: int |
| 73 | chord: str |
| 74 | duration: float |
| 75 | track: str |
| 76 | |
| 77 | |
| 78 | class VoiceLeadingStep(TypedDict): |
| 79 | """Voice-leading movement from one chord to the next. |
| 80 | |
| 81 | Fields: |
| 82 | from_chord: Source chord symbol. |
| 83 | to_chord: Target chord symbol. |
| 84 | from_bar: Bar where the source chord begins. |
| 85 | to_bar: Bar where the target chord begins. |
| 86 | movements: List of ``"NoteFrom->NoteTo"`` strings per voice. |
| 87 | """ |
| 88 | |
| 89 | from_chord: str |
| 90 | to_chord: str |
| 91 | from_bar: int |
| 92 | to_bar: int |
| 93 | movements: list[str] |
| 94 | |
| 95 | |
| 96 | class ChordMapResult(TypedDict): |
| 97 | """Full chord-map result for a commit. |
| 98 | |
| 99 | Fields: |
| 100 | commit: Short commit ref (8 chars). |
| 101 | branch: Branch name at HEAD. |
| 102 | track: Track filter applied (``"all"`` if none). |
| 103 | section: Section filter applied (empty string if none). |
| 104 | chords: Ordered list of :class:`ChordEvent` entries. |
| 105 | voice_leading: Ordered list of :class:`VoiceLeadingStep` entries |
| 106 | (empty unless ``--voice-leading`` was requested). |
| 107 | """ |
| 108 | |
| 109 | commit: str |
| 110 | branch: str |
| 111 | track: str |
| 112 | section: str |
| 113 | chords: list[ChordEvent] |
| 114 | voice_leading: list[VoiceLeadingStep] |
| 115 | |
| 116 | |
| 117 | # --------------------------------------------------------------------------- |
| 118 | # Valid output formats |
| 119 | # --------------------------------------------------------------------------- |
| 120 | |
| 121 | _VALID_FORMATS: frozenset[str] = frozenset({"text", "json", "mermaid"}) |
| 122 | |
| 123 | # --------------------------------------------------------------------------- |
| 124 | # Stub chord data — realistic placeholder until MIDI analysis is wired in |
| 125 | # --------------------------------------------------------------------------- |
| 126 | |
| 127 | _STUB_CHORDS: list[tuple[int, int, str, float]] = [ |
| 128 | (1, 1, "Cmaj9", 1.0), |
| 129 | (2, 1, "Am11", 1.0), |
| 130 | (3, 1, "Dm7", 0.5), |
| 131 | (3, 3, "Gsus4", 0.5), |
| 132 | (4, 1, "G7", 1.0), |
| 133 | (5, 1, "Cmaj9", 1.0), |
| 134 | ] |
| 135 | |
| 136 | _STUB_VOICE_LEADING: list[tuple[str, str, int, int, list[str]]] = [ |
| 137 | ("Cmaj9", "Am11", 1, 2, ["E->E", "G->G", "B->A", "D->C"]), |
| 138 | ("Am11", "Dm7", 2, 3, ["A->D", "C->C", "E->F", "G->A"]), |
| 139 | ("Dm7", "Gsus4", 3, 3, ["D->G", "F->G", "A->D", "C->C"]), |
| 140 | ("Gsus4", "G7", 3, 4, ["G->G", "D->D", "C->B", "G->F"]), |
| 141 | ("G7", "Cmaj9", 4, 5, ["G->C", "F->E", "D->G", "B->B"]), |
| 142 | ] |
| 143 | |
| 144 | |
| 145 | def _stub_chord_events(track: str = "keys") -> list[ChordEvent]: |
| 146 | """Return stub ChordEvent entries for a realistic I-vi-ii-V-I progression.""" |
| 147 | return [ |
| 148 | ChordEvent(bar=bar, beat=beat, chord=chord, duration=dur, track=track) |
| 149 | for bar, beat, chord, dur in _STUB_CHORDS |
| 150 | ] |
| 151 | |
| 152 | |
| 153 | def _stub_voice_leading_steps() -> list[VoiceLeadingStep]: |
| 154 | """Return stub VoiceLeadingStep entries connecting the chord timeline.""" |
| 155 | return [ |
| 156 | VoiceLeadingStep( |
| 157 | from_chord=fc, |
| 158 | to_chord=tc, |
| 159 | from_bar=fb, |
| 160 | to_bar=tb, |
| 161 | movements=mv, |
| 162 | ) |
| 163 | for fc, tc, fb, tb, mv in _STUB_VOICE_LEADING |
| 164 | ] |
| 165 | |
| 166 | |
| 167 | # --------------------------------------------------------------------------- |
| 168 | # Renderers |
| 169 | # --------------------------------------------------------------------------- |
| 170 | |
| 171 | _BAR_BLOCK = "########" |
| 172 | |
| 173 | |
| 174 | def _render_text(result: ChordMapResult) -> str: |
| 175 | """Render a human-readable chord-timeline table.""" |
| 176 | branch = result["branch"] |
| 177 | commit = result["commit"] |
| 178 | head_label = f" (HEAD -> {branch})" if branch else "" |
| 179 | lines: list[str] = [f"Chord map -- commit {commit}{head_label}", ""] |
| 180 | |
| 181 | if result["section"]: |
| 182 | lines.append(f"Section: {result['section']}") |
| 183 | if result["track"] != "all": |
| 184 | lines.append(f"Track: {result['track']}") |
| 185 | if result["section"] or result["track"] != "all": |
| 186 | lines.append("") |
| 187 | |
| 188 | chords = result["chords"] |
| 189 | if not chords: |
| 190 | lines.append("(no chord data found for this commit)") |
| 191 | return "\n".join(lines) |
| 192 | |
| 193 | if result["voice_leading"]: |
| 194 | vl_map: dict[int, VoiceLeadingStep] = { |
| 195 | step["from_bar"]: step for step in result["voice_leading"] |
| 196 | } |
| 197 | prev_chord: Optional[str] = None |
| 198 | prev_bar: int = -1 |
| 199 | for event in chords: |
| 200 | bar = event["bar"] |
| 201 | chord = event["chord"] |
| 202 | bar_label = f"Bar {bar:>2}:" |
| 203 | if prev_chord is not None and prev_bar >= 0: |
| 204 | step = vl_map.get(prev_bar) |
| 205 | movements = f" ({', '.join(step['movements'])})" if step else "" |
| 206 | lines.append(f"{bar_label} {prev_chord:<8} -> {chord}{movements}") |
| 207 | else: |
| 208 | lines.append(f"{bar_label} {chord}") |
| 209 | prev_chord = chord |
| 210 | prev_bar = bar |
| 211 | else: |
| 212 | current_bar: Optional[int] = None |
| 213 | bar_chords: list[tuple[str, float]] = [] |
| 214 | |
| 215 | def _flush_bar(bar_num: int, items: list[tuple[str, float]]) -> None: |
| 216 | bar_label = f"Bar {bar_num:>2}:" |
| 217 | chord_parts: list[str] = [] |
| 218 | for ch, dur in items: |
| 219 | blocks = _BAR_BLOCK if dur >= 1.0 else _BAR_BLOCK[:4] |
| 220 | chord_parts.append(f"{ch:<12}{blocks} ") |
| 221 | lines.append(f"{bar_label} {''.join(chord_parts).rstrip()}") |
| 222 | |
| 223 | for event in chords: |
| 224 | if event["bar"] != current_bar: |
| 225 | if current_bar is not None: |
| 226 | _flush_bar(current_bar, bar_chords) |
| 227 | current_bar = event["bar"] |
| 228 | bar_chords = [] |
| 229 | bar_chords.append((event["chord"], event["duration"])) |
| 230 | if current_bar is not None: |
| 231 | _flush_bar(current_bar, bar_chords) |
| 232 | |
| 233 | lines.append("") |
| 234 | lines.append("(stub -- full MIDI chord detection pending)") |
| 235 | return "\n".join(lines) |
| 236 | |
| 237 | |
| 238 | def _render_mermaid(result: ChordMapResult) -> str: |
| 239 | """Render a Mermaid timeline diagram for the chord progression.""" |
| 240 | chords = result["chords"] |
| 241 | lines: list[str] = [ |
| 242 | "timeline", |
| 243 | f" title Chord map -- {result['commit']}", |
| 244 | ] |
| 245 | if not chords: |
| 246 | lines.append(" section (empty)") |
| 247 | return "\n".join(lines) |
| 248 | |
| 249 | current_bar: Optional[int] = None |
| 250 | for event in chords: |
| 251 | bar = event["bar"] |
| 252 | if bar != current_bar: |
| 253 | lines.append(f" section Bar {bar}") |
| 254 | current_bar = bar |
| 255 | lines.append(f" {event['chord']}") |
| 256 | return "\n".join(lines) |
| 257 | |
| 258 | |
| 259 | def _render_json(result: ChordMapResult) -> str: |
| 260 | """Emit the full ChordMapResult as indented JSON.""" |
| 261 | return json.dumps(dict(result), indent=2) |
| 262 | |
| 263 | |
| 264 | # --------------------------------------------------------------------------- |
| 265 | # Testable async core |
| 266 | # --------------------------------------------------------------------------- |
| 267 | |
| 268 | |
| 269 | async def _chord_map_async( |
| 270 | *, |
| 271 | root: pathlib.Path, |
| 272 | session: AsyncSession, |
| 273 | commit: Optional[str], |
| 274 | section: Optional[str], |
| 275 | track: Optional[str], |
| 276 | bar_grid: bool, |
| 277 | fmt: str, |
| 278 | voice_leading: bool, |
| 279 | ) -> ChordMapResult: |
| 280 | """Core chord-map logic — fully injectable for tests. |
| 281 | |
| 282 | Reads branch/commit metadata from ``.muse/``, applies optional |
| 283 | ``--section`` and ``--track`` filters to placeholder chord data, and |
| 284 | returns a :class:`ChordMapResult` ready for rendering. |
| 285 | |
| 286 | Args: |
| 287 | root: Repository root (directory containing ``.muse/``). |
| 288 | session: Open async DB session (reserved for full implementation). |
| 289 | commit: Commit ref to analyse; defaults to HEAD. |
| 290 | section: Named section/region filter (stub: noted in output). |
| 291 | track: Track filter for chord voicings (stub: tag applied). |
| 292 | bar_grid: If True, align events to bar numbers (default on). |
| 293 | fmt: Output format: ``"text"``, ``"json"``, or ``"mermaid"``. |
| 294 | voice_leading: If True, include voice-leading movements between chords. |
| 295 | |
| 296 | Returns: |
| 297 | A :class:`ChordMapResult` with ``commit``, ``branch``, ``track``, |
| 298 | ``section``, ``chords``, and ``voice_leading`` populated. |
| 299 | """ |
| 300 | muse_dir = root / ".muse" |
| 301 | head_path = muse_dir / "HEAD" |
| 302 | head_ref = head_path.read_text().strip() |
| 303 | branch = head_ref.rsplit("/", 1)[-1] if "/" in head_ref else head_ref |
| 304 | |
| 305 | ref_path = muse_dir / pathlib.Path(head_ref) |
| 306 | head_sha = ref_path.read_text().strip() if ref_path.exists() else "" |
| 307 | resolved_commit = commit or (head_sha[:8] if head_sha else "HEAD") |
| 308 | |
| 309 | resolved_track = track or "keys" |
| 310 | |
| 311 | chords = _stub_chord_events(track=resolved_track) |
| 312 | vl_steps: list[VoiceLeadingStep] = ( |
| 313 | _stub_voice_leading_steps() if voice_leading else [] |
| 314 | ) |
| 315 | |
| 316 | return ChordMapResult( |
| 317 | commit=resolved_commit, |
| 318 | branch=branch, |
| 319 | track=resolved_track if track else "all", |
| 320 | section=section or "", |
| 321 | chords=chords, |
| 322 | voice_leading=vl_steps, |
| 323 | ) |
| 324 | |
| 325 | |
| 326 | # --------------------------------------------------------------------------- |
| 327 | # Typer command |
| 328 | # --------------------------------------------------------------------------- |
| 329 | |
| 330 | |
| 331 | @app.callback(invoke_without_command=True) |
| 332 | def chord_map( |
| 333 | ctx: typer.Context, |
| 334 | commit: Annotated[ |
| 335 | Optional[str], |
| 336 | typer.Argument( |
| 337 | help="Commit ref to analyse. Defaults to HEAD.", |
| 338 | show_default=False, |
| 339 | ), |
| 340 | ] = None, |
| 341 | section: Annotated[ |
| 342 | Optional[str], |
| 343 | typer.Option( |
| 344 | "--section", |
| 345 | help="Scope to a named section/region.", |
| 346 | metavar="TEXT", |
| 347 | show_default=False, |
| 348 | ), |
| 349 | ] = None, |
| 350 | track: Annotated[ |
| 351 | Optional[str], |
| 352 | typer.Option( |
| 353 | "--track", |
| 354 | help="Scope to a specific track (e.g. 'piano' for chord voicings).", |
| 355 | metavar="TEXT", |
| 356 | show_default=False, |
| 357 | ), |
| 358 | ] = None, |
| 359 | bar_grid: Annotated[ |
| 360 | bool, |
| 361 | typer.Option( |
| 362 | "--bar-grid/--no-bar-grid", |
| 363 | help="Align chord events to musical bar numbers (default: on).", |
| 364 | ), |
| 365 | ] = True, |
| 366 | fmt: Annotated[ |
| 367 | str, |
| 368 | typer.Option( |
| 369 | "--format", |
| 370 | help="Output format: text (default), json, or mermaid.", |
| 371 | metavar="FORMAT", |
| 372 | ), |
| 373 | ] = "text", |
| 374 | voice_leading: Annotated[ |
| 375 | bool, |
| 376 | typer.Option( |
| 377 | "--voice-leading", |
| 378 | help="Show how individual notes move between consecutive chords.", |
| 379 | ), |
| 380 | ] = False, |
| 381 | ) -> None: |
| 382 | """Visualize the chord progression embedded in a commit. |
| 383 | |
| 384 | Shows a time-aligned chord timeline so AI agents can reason about |
| 385 | harmonic structure at any point in the composition history. |
| 386 | """ |
| 387 | if fmt not in _VALID_FORMATS: |
| 388 | typer.echo( |
| 389 | f"Invalid --format '{fmt}'. " |
| 390 | f"Valid formats: {', '.join(sorted(_VALID_FORMATS))}" |
| 391 | ) |
| 392 | raise typer.Exit(code=ExitCode.USER_ERROR) |
| 393 | |
| 394 | root = require_repo() |
| 395 | |
| 396 | async def _run() -> ChordMapResult: |
| 397 | async with open_session() as session: |
| 398 | return await _chord_map_async( |
| 399 | root=root, |
| 400 | session=session, |
| 401 | commit=commit, |
| 402 | section=section, |
| 403 | track=track, |
| 404 | bar_grid=bar_grid, |
| 405 | fmt=fmt, |
| 406 | voice_leading=voice_leading, |
| 407 | ) |
| 408 | |
| 409 | try: |
| 410 | result = asyncio.run(_run()) |
| 411 | except typer.Exit: |
| 412 | raise |
| 413 | except Exception as exc: |
| 414 | typer.echo(f"muse chord-map failed: {exc}") |
| 415 | logger.error("muse chord-map error: %s", exc, exc_info=True) |
| 416 | raise typer.Exit(code=ExitCode.INTERNAL_ERROR) |
| 417 | |
| 418 | if fmt == "json": |
| 419 | typer.echo(_render_json(result)) |
| 420 | elif fmt == "mermaid": |
| 421 | typer.echo(_render_mermaid(result)) |
| 422 | else: |
| 423 | typer.echo(_render_text(result)) |