cgcardona / muse public
harmony.py python
561 lines 17.5 KB
Raw
12901c5a Initial extraction from tellurstori/maestro cgcardona <gabriel@tellurstori.com> 4d ago
1 """muse harmony — analyze and query harmonic content across commits.
2
3 Examines the harmonic profile (key, mode, chord progression, harmonic
4 rhythm, and tension) of a given commit (default: HEAD) or a range of
5 commits. Harmonic analysis is one of the most musically significant
6 dimensions exposed by Muse VCS — information that Git has no concept of.
7
8 An AI agent calling ``muse harmony --json`` receives a structured snapshot
9 of the harmonic landscape it can use to make musically coherent generation
10 decisions: stay in the same key, continue the same chord progression, or
11 intentionally create harmonic contrast.
12
13 Command forms
14 -------------
15
16 Analyze HEAD (default)::
17
18 muse harmony
19
20 Analyze a specific commit::
21
22 muse harmony a1b2c3d4
23
24 Analyze a commit range::
25
26 muse harmony HEAD~10..HEAD
27
28 Compare two commits::
29
30 muse harmony --compare HEAD~5
31
32 Extract only the chord progression::
33
34 muse harmony --progression
35
36 Show key center::
37
38 muse harmony --key
39
40 Show mode::
41
42 muse harmony --mode
43
44 Show harmonic tension profile::
45
46 muse harmony --tension
47
48 Restrict to a single instrument track::
49
50 muse harmony --track keys
51
52 Machine-readable JSON output::
53
54 muse harmony --json
55
56 Stub note
57 ---------
58 Full chord detection requires MIDI note extraction from committed snapshot
59 objects. This implementation provides a realistic placeholder in the
60 correct schema. The result type and CLI contract are stable.
61 """
62 from __future__ import annotations
63
64 import asyncio
65 import json
66 import logging
67 import pathlib
68 from typing import Optional
69
70 import typer
71 from sqlalchemy.ext.asyncio import AsyncSession
72 from typing_extensions import Annotated, TypedDict
73
74 from maestro.muse_cli._repo import require_repo
75 from maestro.muse_cli.db import open_session
76 from maestro.muse_cli.errors import ExitCode
77
78 logger = logging.getLogger(__name__)
79
80 app = typer.Typer()
81
82 # ---------------------------------------------------------------------------
83 # Constants — mode vocabulary
84 # ---------------------------------------------------------------------------
85
86 KNOWN_MODES: tuple[str, ...] = (
87 "major",
88 "minor",
89 "dorian",
90 "phrygian",
91 "lydian",
92 "mixolydian",
93 "aeolian",
94 "locrian",
95 )
96
97 KNOWN_MODES_SET: frozenset[str] = frozenset(KNOWN_MODES)
98
99 # ---------------------------------------------------------------------------
100 # Named result types (stable CLI contract)
101 # ---------------------------------------------------------------------------
102
103
104 class HarmonyResult(TypedDict):
105 """Harmonic analysis result for a single commit.
106
107 This is the primary result type for ``muse harmony``. Every field is
108 populated by stub logic today and will be backed by MIDI analysis once
109 the Storpheus inference endpoint exposes a chord detection route.
110
111 Fields
112 ------
113 commit_id : str
114 Short or full commit SHA that was analyzed.
115 branch : str
116 Name of the current branch.
117 key : str | None
118 Detected key center (e.g. ``"Eb"``), or ``None`` for drum-only
119 snapshots with no pitched content.
120 mode : str | None
121 Detected mode (e.g. ``"major"``, ``"dorian"``), or ``None``.
122 confidence : float
123 Key/mode detection confidence in [0.0, 1.0].
124 chord_progression : list[str]
125 Ordered list of chord symbol strings (e.g. ``["Ebmaj7", "Fm7"]``).
126 harmonic_rhythm_avg : float
127 Average number of chord changes per bar.
128 tension_profile : list[float]
129 Per-section tension scores in [0.0, 1.0], where 0.0 = fully
130 consonant and 1.0 = maximally dissonant.
131 track : str
132 Instrument track scope (``"all"`` unless ``--track`` is specified).
133 source : str
134 ``"stub"`` until backed by real MIDI analysis.
135 """
136
137 commit_id: str
138 branch: str
139 key: Optional[str]
140 mode: Optional[str]
141 confidence: float
142 chord_progression: list[str]
143 harmonic_rhythm_avg: float
144 tension_profile: list[float]
145 track: str
146 source: str
147
148
149 class HarmonyCompareResult(TypedDict):
150 """Comparison of harmonic content between two commits.
151
152 Fields
153 ------
154 head : HarmonyResult
155 Harmonic analysis for the HEAD (or specified) commit.
156 compare : HarmonyResult
157 Harmonic analysis for the reference commit.
158 key_changed : bool
159 ``True`` if the key center differs between the two commits.
160 mode_changed : bool
161 ``True`` if the mode differs between the two commits.
162 chord_progression_delta : list[str]
163 Chords present in HEAD but absent in compare (new chords).
164 """
165
166 head: HarmonyResult
167 compare: HarmonyResult
168 key_changed: bool
169 mode_changed: bool
170 chord_progression_delta: list[str]
171
172
173 # ---------------------------------------------------------------------------
174 # Stub data — realistic placeholder until MIDI note data is queryable
175 # ---------------------------------------------------------------------------
176
177 _STUB_KEY = "Eb"
178 _STUB_MODE = "major"
179 _STUB_CONFIDENCE = 0.92
180 _STUB_CHORD_PROGRESSION = ["Ebmaj7", "Fm7", "Bb7sus4", "Bb7", "Ebmaj7", "Abmaj7", "Gm7", "Cm7"]
181 _STUB_HARMONIC_RHYTHM_AVG = 2.1
182 _STUB_TENSION_PROFILE = [0.2, 0.4, 0.8, 0.3]
183
184
185 def _stub_harmony(commit_id: str, branch: str, track: str = "all") -> HarmonyResult:
186 """Return a realistic placeholder HarmonyResult.
187
188 Produces a II-V-I flavored progression in Eb major — one of the most
189 common key centers in jazz and soul productions. Confidence and tension
190 values reflect a textbook tension-release arc.
191 """
192 return HarmonyResult(
193 commit_id=commit_id,
194 branch=branch,
195 key=_STUB_KEY,
196 mode=_STUB_MODE,
197 confidence=_STUB_CONFIDENCE,
198 chord_progression=list(_STUB_CHORD_PROGRESSION),
199 harmonic_rhythm_avg=_STUB_HARMONIC_RHYTHM_AVG,
200 tension_profile=list(_STUB_TENSION_PROFILE),
201 track=track,
202 source="stub",
203 )
204
205
206 # ---------------------------------------------------------------------------
207 # Testable async core
208 # ---------------------------------------------------------------------------
209
210
211 async def _harmony_analyze_async(
212 *,
213 root: pathlib.Path,
214 session: AsyncSession,
215 commit: Optional[str],
216 track: Optional[str],
217 section: Optional[str],
218 compare: Optional[str],
219 commit_range: Optional[str],
220 show_progression: bool,
221 show_key: bool,
222 show_mode: bool,
223 show_tension: bool,
224 as_json: bool,
225 ) -> HarmonyResult:
226 """Core harmonic analysis logic — fully injectable for tests.
227
228 Resolves the target commit from the ``.muse/`` layout, produces a
229 ``HarmonyResult`` (stub today, full MIDI analysis in future), and
230 renders it to stdout according to the active flags.
231
232 Returns the ``HarmonyResult`` so callers (tests) can assert on values
233 without parsing stdout.
234
235 Args:
236 root: Repository root (directory containing ``.muse/``).
237 session: Open async DB session (reserved for full implementation).
238 commit: Commit ref to analyse; defaults to HEAD.
239 track: Restrict to a named MIDI track, or ``None`` for all.
240 section: Restrict to a named region (stub: noted in output).
241 compare: Second commit ref for side-by-side comparison.
242 commit_range: ``from..to`` range string (stub: noted in output).
243 show_progression: If ``True``, show only the chord progression sequence.
244 show_key: If ``True``, show only the detected key center.
245 show_mode: If ``True``, show only the detected mode.
246 show_tension: If ``True``, show only the tension profile.
247 as_json: Emit JSON instead of human-readable text.
248 """
249 muse_dir = root / ".muse"
250
251 # -- Resolve branch / commit ref --
252 head_ref = (muse_dir / "HEAD").read_text().strip()
253 branch = head_ref.rsplit("/", 1)[-1] if "/" in head_ref else head_ref
254 ref_path = muse_dir / pathlib.Path(head_ref)
255
256 head_sha = ref_path.read_text().strip() if ref_path.exists() else ""
257
258 if not head_sha and not commit:
259 typer.echo(f"No commits yet on branch {branch} — nothing to analyse.")
260 raise typer.Exit(code=ExitCode.SUCCESS)
261
262 resolved_commit = commit or (head_sha[:8] if head_sha else "HEAD")
263 effective_track = track or "all"
264
265 # -- Stub: produce placeholder result --
266 result = _stub_harmony(
267 commit_id=resolved_commit,
268 branch=branch,
269 track=effective_track,
270 )
271
272 # -- Stub boundary notes for unimplemented flags --
273 if commit_range:
274 typer.echo(
275 f"⚠️ --range {commit_range!r}: range analysis not yet implemented. "
276 f"Showing HEAD ({resolved_commit}) only."
277 )
278 if section:
279 typer.echo(f"⚠️ --section {section!r}: region filtering not yet implemented.")
280
281 # -- Render --
282 if compare is not None:
283 compare_result = _stub_harmony(
284 commit_id=compare,
285 branch=branch,
286 track=effective_track,
287 )
288 cmp = HarmonyCompareResult(
289 head=result,
290 compare=compare_result,
291 key_changed=result["key"] != compare_result["key"],
292 mode_changed=result["mode"] != compare_result["mode"],
293 chord_progression_delta=[
294 c
295 for c in result["chord_progression"]
296 if c not in compare_result["chord_progression"]
297 ],
298 )
299 if as_json:
300 _render_compare_json(cmp)
301 else:
302 _render_compare_human(cmp)
303 return result
304
305 # Single-commit render with optional field scoping
306 if as_json:
307 _render_result_json(result, show_progression, show_key, show_mode, show_tension)
308 else:
309 _render_result_human(result, show_progression, show_key, show_mode, show_tension)
310
311 return result
312
313
314 # ---------------------------------------------------------------------------
315 # Output formatters
316 # ---------------------------------------------------------------------------
317
318
319 def _tension_label(profile: list[float]) -> str:
320 """Classify a tension profile into a human-readable arc description.
321
322 Uses the shape of the profile (monotone rise/fall, arch, valley) to
323 produce vocabulary familiar to producers and music directors.
324 """
325 if not profile:
326 return "unknown"
327 if len(profile) == 1:
328 v = profile[0]
329 if v < 0.3:
330 return "Low"
331 if v < 0.6:
332 return "Medium"
333 return "High"
334
335 rising = all(profile[i] <= profile[i + 1] for i in range(len(profile) - 1))
336 falling = all(profile[i] >= profile[i + 1] for i in range(len(profile) - 1))
337 peak_idx = profile.index(max(profile))
338 valley_idx = profile.index(min(profile))
339
340 if rising:
341 return "Rising (tension build)"
342 if falling:
343 return "Falling (tension release)"
344 if 0 < peak_idx < len(profile) - 1:
345 return "Low → Medium → High → Resolution (textbook tension-release arc)"
346 if 0 < valley_idx < len(profile) - 1:
347 return "High → Resolution → High (bracketed release)"
348 return "Variable"
349
350
351 def _render_result_human(
352 result: HarmonyResult,
353 show_progression: bool,
354 show_key: bool,
355 show_mode: bool,
356 show_tension: bool,
357 ) -> None:
358 """Render a HarmonyResult as human-readable text."""
359 full = not any([show_progression, show_key, show_mode, show_tension])
360
361 if full:
362 typer.echo(f"Commit {result['commit_id']} — Harmonic Analysis")
363 if result["source"] == "stub":
364 typer.echo("(stub — full MIDI analysis pending)")
365 typer.echo("")
366
367 if full or show_key:
368 key_display = result["key"] or "— (no pitched content)"
369 typer.echo(
370 f"Key: {key_display}"
371 + (f" (confidence: {result['confidence']:.2f})" if result["key"] else "")
372 )
373
374 if full or show_mode:
375 mode_display = result["mode"] or ""
376 typer.echo(f"Mode: {mode_display}")
377
378 if full or show_progression:
379 if result["chord_progression"]:
380 progression_str = " | ".join(result["chord_progression"])
381 else:
382 progression_str = "(no pitched content — drums only)"
383 typer.echo(f"Chord progression: {progression_str}")
384
385 if full:
386 typer.echo(f"Harmonic rhythm: {result['harmonic_rhythm_avg']:.1f} chords/bar avg")
387
388 if full or show_tension:
389 label = _tension_label(result["tension_profile"])
390 profile_str = " → ".join(f"{v:.1f}" for v in result["tension_profile"])
391 typer.echo(f"Tension profile: {label} [{profile_str}]")
392
393
394 def _render_result_json(
395 result: HarmonyResult,
396 show_progression: bool,
397 show_key: bool,
398 show_mode: bool,
399 show_tension: bool,
400 ) -> None:
401 """Render a HarmonyResult as JSON, optionally scoped to requested fields."""
402 full = not any([show_progression, show_key, show_mode, show_tension])
403
404 if full:
405 payload: dict[str, object] = dict(result)
406 else:
407 payload = {"commit_id": result["commit_id"], "branch": result["branch"]}
408 if show_key:
409 payload["key"] = result["key"]
410 payload["confidence"] = result["confidence"]
411 if show_mode:
412 payload["mode"] = result["mode"]
413 if show_progression:
414 payload["chord_progression"] = result["chord_progression"]
415 if show_tension:
416 payload["tension_profile"] = result["tension_profile"]
417
418 typer.echo(json.dumps(payload, indent=2))
419
420
421 def _render_compare_human(cmp: HarmonyCompareResult) -> None:
422 """Render a HarmonyCompareResult as human-readable text."""
423 head = cmp["head"]
424 ref = cmp["compare"]
425
426 typer.echo(f"Harmonic Comparison — HEAD ({head['commit_id']}) vs {ref['commit_id']}")
427 typer.echo("")
428 typer.echo(f" Key HEAD: {head['key'] or ''} Compare: {ref['key'] or ''}")
429 typer.echo(f" Mode HEAD: {head['mode'] or ''} Compare: {ref['mode'] or ''}")
430 typer.echo(f" Key changed: {'yes' if cmp['key_changed'] else 'no'}")
431 typer.echo(f" Mode changed: {'yes' if cmp['mode_changed'] else 'no'}")
432 if cmp["chord_progression_delta"]:
433 typer.echo(f" New chords in HEAD: {' '.join(cmp['chord_progression_delta'])}")
434 else:
435 typer.echo(" Chord progression: unchanged")
436
437
438 def _render_compare_json(cmp: HarmonyCompareResult) -> None:
439 """Render a HarmonyCompareResult as JSON."""
440 typer.echo(json.dumps(dict(cmp), indent=2))
441
442
443 # ---------------------------------------------------------------------------
444 # Typer command
445 # ---------------------------------------------------------------------------
446
447
448 @app.callback(invoke_without_command=True)
449 def harmony(
450 ctx: typer.Context,
451 commit: Annotated[
452 Optional[str],
453 typer.Argument(
454 help="Commit SHA to analyze. Defaults to HEAD.",
455 show_default=False,
456 ),
457 ] = None,
458 track: Annotated[
459 Optional[str],
460 typer.Option(
461 "--track",
462 help="Restrict analysis to a named MIDI track (e.g. 'keys', 'bass').",
463 show_default=False,
464 ),
465 ] = None,
466 section: Annotated[
467 Optional[str],
468 typer.Option(
469 "--section",
470 help="Restrict analysis to a named musical section or region.",
471 show_default=False,
472 ),
473 ] = None,
474 compare: Annotated[
475 Optional[str],
476 typer.Option(
477 "--compare",
478 metavar="COMMIT",
479 help="Compare harmonic content of HEAD against another commit.",
480 show_default=False,
481 ),
482 ] = None,
483 commit_range: Annotated[
484 Optional[str],
485 typer.Option(
486 "--range",
487 metavar="FROM..TO",
488 help="Analyze harmonic content across a commit range (e.g. HEAD~10..HEAD).",
489 show_default=False,
490 ),
491 ] = None,
492 show_progression: Annotated[
493 bool,
494 typer.Option(
495 "--progression",
496 help="Show only the chord progression sequence.",
497 ),
498 ] = False,
499 show_key: Annotated[
500 bool,
501 typer.Option(
502 "--key",
503 help="Show only the detected key center.",
504 ),
505 ] = False,
506 show_mode: Annotated[
507 bool,
508 typer.Option(
509 "--mode",
510 help="Show only the detected mode (major, minor, dorian, etc.).",
511 ),
512 ] = False,
513 show_tension: Annotated[
514 bool,
515 typer.Option(
516 "--tension",
517 help="Show only the harmonic tension profile.",
518 ),
519 ] = False,
520 as_json: Annotated[
521 bool,
522 typer.Option(
523 "--json",
524 help="Emit machine-readable JSON output.",
525 ),
526 ] = False,
527 ) -> None:
528 """Analyze harmonic content (key, mode, chords, tension) of a commit.
529
530 Without flags, prints a full harmonic summary for the target commit.
531 Use ``--key``, ``--mode``, ``--progression``, or ``--tension`` to
532 scope the output to a single dimension. Use ``--json`` for structured
533 output suitable for AI agent consumption.
534 """
535 root = require_repo()
536
537 async def _run() -> None:
538 async with open_session() as session:
539 await _harmony_analyze_async(
540 root=root,
541 session=session,
542 commit=commit,
543 track=track,
544 section=section,
545 compare=compare,
546 commit_range=commit_range,
547 show_progression=show_progression,
548 show_key=show_key,
549 show_mode=show_mode,
550 show_tension=show_tension,
551 as_json=as_json,
552 )
553
554 try:
555 asyncio.run(_run())
556 except typer.Exit:
557 raise
558 except Exception as exc:
559 typer.echo(f"❌ muse harmony failed: {exc}")
560 logger.error("❌ muse harmony error: %s", exc, exc_info=True)
561 raise typer.Exit(code=ExitCode.INTERNAL_ERROR)