cgcardona / muse public
form.py python
498 lines 15.0 KB
Raw
12901c5a Initial extraction from tellurstori/maestro cgcardona <gabriel@tellurstori.com> 4d ago
1 """muse form — analyze and display the musical form of a commit.
2
3 Musical form is the large-scale structural blueprint of a composition:
4 the ordering and labelling of sections (intro, verse, chorus, bridge,
5 outro, etc.) that define how a piece unfolds over time.
6
7 Command forms
8 -------------
9
10 Detect form on HEAD (default)::
11
12 muse form
13
14 Detect form at a specific commit::
15
16 muse form a1b2c3d4
17
18 Annotate the current working tree with an explicit form string::
19
20 muse form --set "verse-chorus-verse-chorus-bridge-chorus"
21
22 Show a section timeline (map view)::
23
24 muse form --map
25
26 Show how the form changed across commits::
27
28 muse form --history
29
30 Machine-readable JSON output::
31
32 muse form --json
33
34 Flags
35 -----
36 ``[<commit>]`` Target commit ref (default: HEAD).
37 ``--set TEXT`` Annotate with an explicit form string (e.g. "AABA", "verse-chorus").
38 ``--detect`` Auto-detect form from section repetition patterns (default).
39 ``--map`` Show the section arrangement as a visual timeline.
40 ``--history`` Show how the form changed across commits.
41 ``--json`` Machine-readable output.
42
43 Section vocabulary
44 ------------------
45 intro, verse, pre-chorus, chorus, bridge, breakdown, outro, A, B, C
46
47 Detection heuristic
48 -------------------
49 Sections with identical content fingerprints are assigned the same label
50 (A, B, C...). Named roles (verse, chorus, etc.) are inferred from MIDI
51 metadata stored in ``.muse/sections/`` when available; otherwise uppercase
52 letter labels are used.
53
54 Result type
55 -----------
56 ``FormAnalysisResult`` (TypedDict) -- stable schema for agent consumers.
57 ``FormHistoryEntry`` (TypedDict) -- wraps FormAnalysisResult with commit metadata.
58
59 See ``docs/reference/type_contracts.md S FormAnalysisResult``.
60 """
61 from __future__ import annotations
62
63 import asyncio
64 import json
65 import logging
66 import pathlib
67 from typing import Optional
68
69 import typer
70 from sqlalchemy.ext.asyncio import AsyncSession
71 from typing_extensions import Annotated, TypedDict
72
73 from maestro.muse_cli._repo import require_repo
74 from maestro.muse_cli.db import open_session
75 from maestro.muse_cli.errors import ExitCode
76
77 logger = logging.getLogger(__name__)
78
79 app = typer.Typer()
80
81 # ---------------------------------------------------------------------------
82 # Section label vocabulary
83 # ---------------------------------------------------------------------------
84
85 #: Canonical role labels -- used when MIDI metadata provides named sections.
86 ROLE_LABELS: tuple[str, ...] = (
87 "intro",
88 "verse",
89 "pre-chorus",
90 "chorus",
91 "bridge",
92 "breakdown",
93 "outro",
94 )
95
96 #: Structural letter labels -- used when roles cannot be inferred.
97 LETTER_LABELS: tuple[str, ...] = tuple("ABCDEFGHIJ")
98
99 _VALID_ROLES: frozenset[str] = frozenset(ROLE_LABELS)
100
101
102 # ---------------------------------------------------------------------------
103 # Named result types (stable CLI contract)
104 # ---------------------------------------------------------------------------
105
106
107 class FormSection(TypedDict):
108 """A single structural unit within the detected form."""
109
110 label: str
111 role: str
112 index: int
113
114
115 class FormAnalysisResult(TypedDict):
116 """Full form analysis for one commit."""
117
118 commit: str
119 branch: str
120 form_string: str
121 sections: list[FormSection]
122 source: str
123
124
125 class FormHistoryEntry(TypedDict):
126 """Form analysis result paired with its position in the commit history."""
127
128 position: int
129 result: FormAnalysisResult
130
131
132 # ---------------------------------------------------------------------------
133 # Stub data -- realistic placeholder until section metadata is queryable
134 # ---------------------------------------------------------------------------
135
136 _STUB_SECTIONS: list[tuple[str, str]] = [
137 ("intro", "intro"),
138 ("A", "verse"),
139 ("B", "chorus"),
140 ("A", "verse"),
141 ("B", "chorus"),
142 ("C", "bridge"),
143 ("B", "chorus"),
144 ("outro", "outro"),
145 ]
146
147
148 def _stub_form_sections() -> list[FormSection]:
149 """Return stub FormSection entries (placeholder for real DB/file query).
150
151 The stub models a common verse-chorus-verse-chorus-bridge-chorus structure,
152 which is the most frequent form in contemporary pop/R&B production.
153 """
154 return [
155 FormSection(label=label, role=role, index=i)
156 for i, (label, role) in enumerate(_STUB_SECTIONS)
157 ]
158
159
160 def _sections_to_form_string(sections: list[FormSection]) -> str:
161 """Convert a section list into the canonical pipe-separated form string.
162
163 Example: ``"intro | A | B | A | B | C | B | outro"``
164
165 Args:
166 sections: Ordered list of FormSection entries.
167
168 Returns:
169 Human-readable form string.
170 """
171 return " | ".join(s["label"] for s in sections)
172
173
174 # ---------------------------------------------------------------------------
175 # Output formatters
176 # ---------------------------------------------------------------------------
177
178
179 def _render_form_text(result: FormAnalysisResult) -> str:
180 """Render a form result as human-readable text.
181
182 Args:
183 result: Populated FormAnalysisResult.
184
185 Returns:
186 Multi-line string ready for typer.echo.
187 """
188 head_label = f" (HEAD -> {result['branch']})" if result["branch"] else ""
189 lines = [
190 f"Musical form -- commit {result['commit']}{head_label}",
191 "",
192 f" {result['form_string']}",
193 "",
194 "Sections:",
195 ]
196 for sec in result["sections"]:
197 role_hint = f" [{sec['role']}]" if sec["role"] != sec["label"] else ""
198 lines.append(f" {sec['index'] + 1:>2}. {sec['label']:<12}{role_hint}")
199 if result.get("source") == "stub":
200 lines.append("")
201 lines.append(" (stub -- full section analysis pending)")
202 return "\n".join(lines)
203
204
205 def _render_map_text(result: FormAnalysisResult) -> str:
206 """Render the section arrangement as a visual timeline.
207
208 Produces a compact horizontal timeline where each section occupies a
209 fixed-width cell, making structural repetition immediately visible.
210
211 Args:
212 result: Populated FormAnalysisResult.
213
214 Returns:
215 Multi-line string ready for typer.echo.
216 """
217 head_label = f" (HEAD -> {result['branch']})" if result["branch"] else ""
218 cell_w = 10
219 sections = result["sections"]
220 top = "+" + "+".join("-" * cell_w for _ in sections) + "+"
221 mid = "|" + "|".join(s["label"][:cell_w].center(cell_w) for s in sections) + "|"
222 bot = "+" + "+".join("-" * cell_w for _ in sections) + "+"
223 nums = " " + " ".join(str(s["index"] + 1).center(cell_w) for s in sections)
224 lines = [
225 f"Form map -- commit {result['commit']}{head_label}",
226 "",
227 top,
228 mid,
229 bot,
230 nums,
231 ]
232 if result.get("source") == "stub":
233 lines.append("")
234 lines.append("(stub -- full section analysis pending)")
235 return "\n".join(lines)
236
237
238 def _render_history_text(entries: list[FormHistoryEntry]) -> str:
239 """Render the form history as a chronological list.
240
241 Args:
242 entries: List of FormHistoryEntry from newest to oldest.
243
244 Returns:
245 Multi-line string ready for typer.echo.
246 """
247 if not entries:
248 return "(no form history found)"
249 lines: list[str] = []
250 for entry in entries:
251 r = entry["result"]
252 lines.append(f" #{entry['position']} {r['commit']} {r['form_string']}")
253 return "\n".join(lines)
254
255
256 # ---------------------------------------------------------------------------
257 # Testable async core
258 # ---------------------------------------------------------------------------
259
260
261 async def _form_detect_async(
262 *,
263 root: pathlib.Path,
264 session: AsyncSession,
265 commit: Optional[str],
266 ) -> FormAnalysisResult:
267 """Detect the musical form for a given commit (or HEAD).
268
269 Stub implementation: resolves the branch/commit from ``.muse/HEAD`` and
270 returns a placeholder verse-chorus-bridge structure. Full analysis will
271 read section fingerprints from ``.muse/sections/`` and compare content
272 hashes to assign repeated-section labels automatically.
273
274 Args:
275 root: Repository root (directory containing ``.muse/``).
276 session: Open async DB session (reserved for full implementation).
277 commit: Commit SHA to analyse, or None for HEAD.
278
279 Returns:
280 A FormAnalysisResult with commit, branch, form_string, sections, source.
281 """
282 muse_dir = root / ".muse"
283 head_path = muse_dir / "HEAD"
284 head_ref = head_path.read_text().strip()
285 branch = head_ref.rsplit("/", 1)[-1] if "/" in head_ref else head_ref
286
287 ref_path = muse_dir / pathlib.Path(head_ref)
288 head_sha = ref_path.read_text().strip() if ref_path.exists() else "0000000"
289 resolved_commit = commit or (head_sha[:8] if head_sha else "HEAD")
290
291 sections = _stub_form_sections()
292 form_string = _sections_to_form_string(sections)
293
294 return FormAnalysisResult(
295 commit=resolved_commit,
296 branch=branch,
297 form_string=form_string,
298 sections=sections,
299 source="stub",
300 )
301
302
303 async def _form_set_async(
304 *,
305 root: pathlib.Path,
306 session: AsyncSession,
307 form_value: str,
308 ) -> FormAnalysisResult:
309 """Store an explicit form annotation for the current working tree.
310
311 Parses the user-supplied form string (e.g. "AABA" or
312 "verse-chorus-bridge") into FormSection entries and records the
313 annotation. The stub writes the annotation to
314 ``.muse/form_annotation.json``; the full implementation will attach it
315 to the pending commit object.
316
317 Args:
318 root: Repository root.
319 session: Open async DB session.
320 form_value: Explicit form string supplied via --set.
321
322 Returns:
323 A FormAnalysisResult representing the stored annotation.
324 """
325 muse_dir = root / ".muse"
326 head_path = muse_dir / "HEAD"
327 head_ref = head_path.read_text().strip()
328 branch = head_ref.rsplit("/", 1)[-1] if "/" in head_ref else head_ref
329
330 # Parse pipe-separated, hyphen-separated, or space-separated tokens.
331 if "|" in form_value:
332 tokens = [t.strip() for t in form_value.split("|") if t.strip()]
333 elif "-" in form_value and not any(c in form_value for c in (" ", "|")):
334 tokens = [t.strip() for t in form_value.split("-") if t.strip()]
335 else:
336 tokens = [t.strip() for t in form_value.split() if t.strip()]
337
338 sections: list[FormSection] = [
339 FormSection(
340 label=tok,
341 role=tok.lower() if tok.lower() in _VALID_ROLES else tok,
342 index=i,
343 )
344 for i, tok in enumerate(tokens)
345 ]
346 reconstructed = _sections_to_form_string(sections)
347
348 # Stub: persist to .muse/form_annotation.json
349 annotation_path = muse_dir / "form_annotation.json"
350 annotation_path.write_text(
351 json.dumps(
352 {
353 "form_string": reconstructed,
354 "sections": [dict(s) for s in sections],
355 "source": "annotation",
356 },
357 indent=2,
358 )
359 )
360
361 return FormAnalysisResult(
362 commit="",
363 branch=branch,
364 form_string=reconstructed,
365 sections=sections,
366 source="annotation",
367 )
368
369
370 async def _form_history_async(
371 *,
372 root: pathlib.Path,
373 session: AsyncSession,
374 ) -> list[FormHistoryEntry]:
375 """Return the form history for the current branch.
376
377 Stub implementation returning a single HEAD entry. Full implementation
378 will walk the commit chain and aggregate form annotations stored per-commit
379 in ``.muse/objects/``, surfacing structural restructures as distinct entries.
380
381 Args:
382 root: Repository root.
383 session: Open async DB session.
384
385 Returns:
386 List of FormHistoryEntry entries, newest first.
387 """
388 head_result = await _form_detect_async(root=root, session=session, commit=None)
389 return [FormHistoryEntry(position=1, result=head_result)]
390
391
392 # ---------------------------------------------------------------------------
393 # Typer command
394 # ---------------------------------------------------------------------------
395
396
397 @app.callback(invoke_without_command=True)
398 def form(
399 ctx: typer.Context,
400 commit: Annotated[
401 Optional[str],
402 typer.Argument(
403 help="Commit ref to analyse (default: HEAD).",
404 show_default=False,
405 ),
406 ] = None,
407 set_form: Annotated[
408 Optional[str],
409 typer.Option(
410 "--set",
411 help=(
412 "Annotate with an explicit form string "
413 "(e.g. \"AABA\", \"verse-chorus-bridge\", \"Intro | A | B | A\")."
414 ),
415 show_default=False,
416 ),
417 ] = None,
418 detect: Annotated[
419 bool,
420 typer.Option(
421 "--detect",
422 help="Auto-detect form from section repetition patterns (default).",
423 ),
424 ] = True,
425 map_flag: Annotated[
426 bool,
427 typer.Option(
428 "--map",
429 help="Show the section arrangement as a visual timeline.",
430 ),
431 ] = False,
432 history: Annotated[
433 bool,
434 typer.Option(
435 "--history",
436 help="Show how the form changed across commits.",
437 ),
438 ] = False,
439 as_json: Annotated[
440 bool,
441 typer.Option("--json", help="Emit machine-readable JSON output."),
442 ] = False,
443 ) -> None:
444 """Analyze and display the musical form of a composition.
445
446 With no flags, detects and displays the musical form for the HEAD commit.
447 Use --set to persist an explicit form annotation. Use --map to
448 visualise the section layout as a timeline. Use --history to see how
449 the form evolved across the commit chain.
450 """
451 root = require_repo()
452
453 async def _run() -> None:
454 async with open_session() as session:
455 if set_form is not None:
456 result = await _form_set_async(
457 root=root, session=session, form_value=set_form
458 )
459 if as_json:
460 typer.echo(json.dumps(dict(result), indent=2))
461 else:
462 typer.echo(f"Form annotated: {result['form_string']}")
463 return
464
465 if history:
466 entries = await _form_history_async(root=root, session=session)
467 if as_json:
468 payload = [
469 {"position": e["position"], "result": dict(e["result"])}
470 for e in entries
471 ]
472 typer.echo(json.dumps(payload, indent=2))
473 else:
474 typer.echo("Form history (newest first):")
475 typer.echo("")
476 typer.echo(_render_history_text(entries))
477 return
478
479 # Default or --detect: show the form for the target commit.
480 result = await _form_detect_async(
481 root=root, session=session, commit=commit
482 )
483
484 if as_json:
485 typer.echo(json.dumps(dict(result), indent=2))
486 elif map_flag:
487 typer.echo(_render_map_text(result))
488 else:
489 typer.echo(_render_form_text(result))
490
491 try:
492 asyncio.run(_run())
493 except typer.Exit:
494 raise
495 except Exception as exc:
496 typer.echo(f"muse form failed: {exc}")
497 logger.error("❌ muse form error: %s", exc, exc_info=True)
498 raise typer.Exit(code=ExitCode.INTERNAL_ERROR)