timeline.py
python
| 1 | """muse timeline — visualize musical evolution chronologically. |
| 2 | |
| 3 | Renders a commit-by-commit chronological view of a composition's creative |
| 4 | arc with music-semantic metadata: emotion tags, section grouping, and |
| 5 | per-track activity. This is the "album liner notes" view of a project's |
| 6 | evolution — no Git equivalent exists. |
| 7 | |
| 8 | Default output (text):: |
| 9 | |
| 10 | 2026-02-01 abc1234 Initial drum arrangement [drums] [melancholic] ████ |
| 11 | 2026-02-02 def5678 Add bass line [bass] [melancholic] ██████ |
| 12 | 2026-02-03 ghi9012 Chorus melody [keys,vocals] [joyful] █████████ |
| 13 | |
| 14 | Flags |
| 15 | ----- |
| 16 | --emotion Add emotion indicator column (shown by default when tags exist). |
| 17 | --sections Group commits under section headers (e.g. ── chorus ──). |
| 18 | --tracks Show per-track activity column. |
| 19 | --json Machine-readable JSON for UI rendering or agent consumption. |
| 20 | --limit N Cap the commit walk (default: 1000). |
| 21 | [<range>] Optional commit range for future ``HEAD~10..HEAD`` syntax. |
| 22 | Currently accepted but reserved (full history is always shown). |
| 23 | """ |
| 24 | from __future__ import annotations |
| 25 | |
| 26 | import asyncio |
| 27 | import json |
| 28 | import logging |
| 29 | import pathlib |
| 30 | from typing import TypedDict |
| 31 | |
| 32 | import typer |
| 33 | from sqlalchemy.ext.asyncio import AsyncSession |
| 34 | |
| 35 | from maestro.muse_cli._repo import require_repo |
| 36 | from maestro.muse_cli.db import open_session |
| 37 | from maestro.muse_cli.errors import ExitCode |
| 38 | from maestro.services.muse_timeline import ( |
| 39 | MuseTimelineEntry, |
| 40 | MuseTimelineResult, |
| 41 | build_timeline, |
| 42 | ) |
| 43 | |
| 44 | logger = logging.getLogger(__name__) |
| 45 | |
| 46 | app = typer.Typer() |
| 47 | |
| 48 | _DEFAULT_LIMIT = 1000 |
| 49 | |
| 50 | |
| 51 | class _TimelineEntryDict(TypedDict): |
| 52 | """JSON-serializable shape of a single timeline entry.""" |
| 53 | |
| 54 | commit_id: str |
| 55 | short_id: str |
| 56 | committed_at: str |
| 57 | message: str |
| 58 | emotion: str | None |
| 59 | sections: list[str] |
| 60 | tracks: list[str] |
| 61 | activity: int |
| 62 | |
| 63 | |
| 64 | class _TimelineJsonPayload(TypedDict): |
| 65 | """JSON-serializable shape of the full timeline response.""" |
| 66 | |
| 67 | branch: str |
| 68 | total_commits: int |
| 69 | emotion_arc: list[str] |
| 70 | section_order: list[str] |
| 71 | entries: list[_TimelineEntryDict] |
| 72 | |
| 73 | # Unicode block characters for activity density bars. |
| 74 | _BLOCK = "█" |
| 75 | _MAX_BLOCKS = 10 |
| 76 | _MIN_BLOCKS = 1 |
| 77 | |
| 78 | |
| 79 | # --------------------------------------------------------------------------- |
| 80 | # Internal helpers |
| 81 | # --------------------------------------------------------------------------- |
| 82 | |
| 83 | |
| 84 | def _activity_bar(activity: int, max_activity: int) -> str: |
| 85 | """Render a Unicode block bar proportional to *activity*. |
| 86 | |
| 87 | Width is scaled so the most-active commit gets ``_MAX_BLOCKS`` blocks |
| 88 | and the least-active gets at least ``_MIN_BLOCKS``. Returns a blank |
| 89 | string when ``max_activity`` is 0. |
| 90 | """ |
| 91 | if max_activity == 0: |
| 92 | return _BLOCK * _MIN_BLOCKS |
| 93 | scaled = max( |
| 94 | _MIN_BLOCKS, |
| 95 | round(_MAX_BLOCKS * activity / max_activity), |
| 96 | ) |
| 97 | return _BLOCK * scaled |
| 98 | |
| 99 | |
| 100 | def _load_muse_state(root: pathlib.Path) -> tuple[str, str, str]: |
| 101 | """Read branch name, HEAD ref, and head_commit_id from ``.muse/``. |
| 102 | |
| 103 | Returns ``(branch, head_ref, head_commit_id)``. ``head_commit_id`` |
| 104 | is an empty string when the branch has no commits yet. |
| 105 | """ |
| 106 | import json as _json |
| 107 | |
| 108 | muse_dir = root / ".muse" |
| 109 | head_ref = (muse_dir / "HEAD").read_text().strip() |
| 110 | branch = head_ref.rsplit("/", 1)[-1] if "/" in head_ref else head_ref |
| 111 | ref_path = muse_dir / pathlib.Path(head_ref) |
| 112 | head_commit_id = ref_path.read_text().strip() if ref_path.exists() else "" |
| 113 | repo_data: dict[str, str] = _json.loads((muse_dir / "repo.json").read_text()) |
| 114 | repo_id = repo_data["repo_id"] |
| 115 | return repo_id, branch, head_commit_id |
| 116 | |
| 117 | |
| 118 | # --------------------------------------------------------------------------- |
| 119 | # Renderers |
| 120 | # --------------------------------------------------------------------------- |
| 121 | |
| 122 | |
| 123 | def _render_text( |
| 124 | result: MuseTimelineResult, |
| 125 | *, |
| 126 | show_emotion: bool, |
| 127 | show_sections: bool, |
| 128 | show_tracks: bool, |
| 129 | ) -> None: |
| 130 | """Render the timeline as a human-readable terminal table. |
| 131 | |
| 132 | Columns (left to right): |
| 133 | - Date (YYYY-MM-DD) |
| 134 | - Short ID (7 chars) |
| 135 | - Message (truncated to 30 chars) |
| 136 | - Tracks (comma-joined, or — if none) |
| 137 | - Emotion (or — if none) [when show_emotion or show_sections are on] |
| 138 | - Unicode activity bar |
| 139 | |
| 140 | When ``show_sections`` is True, a section header line is printed |
| 141 | whenever the section set changes. |
| 142 | """ |
| 143 | entries = result.entries |
| 144 | if not entries: |
| 145 | typer.echo("No commits in timeline.") |
| 146 | return |
| 147 | |
| 148 | typer.echo(f"Timeline — branch: {result.branch} ({result.total_commits} commit(s))") |
| 149 | typer.echo("") |
| 150 | |
| 151 | max_activity = max(e.activity for e in entries) if entries else 1 |
| 152 | |
| 153 | prev_sections: tuple[str, ...] = () |
| 154 | |
| 155 | for entry in entries: |
| 156 | if show_sections and entry.sections != prev_sections: |
| 157 | sections_label = ", ".join(entry.sections) if entry.sections else "" |
| 158 | typer.echo(f" ── {sections_label} ──") |
| 159 | prev_sections = entry.sections |
| 160 | |
| 161 | date_str = entry.committed_at.strftime("%Y-%m-%d") |
| 162 | short_id = entry.short_id |
| 163 | message = entry.message[:30].ljust(30) |
| 164 | bar = _activity_bar(entry.activity, max_activity) |
| 165 | |
| 166 | tracks_col = "" |
| 167 | if show_tracks: |
| 168 | tracks_label = ",".join(entry.tracks) if entry.tracks else "" |
| 169 | tracks_col = f" [{tracks_label:<20}]" |
| 170 | |
| 171 | emotion_col = "" |
| 172 | if show_emotion: |
| 173 | emotion_label = entry.emotion or "" |
| 174 | emotion_col = f" [{emotion_label:<15}]" |
| 175 | |
| 176 | typer.echo( |
| 177 | f"{date_str} {short_id} {message}{tracks_col}{emotion_col} {bar}" |
| 178 | ) |
| 179 | |
| 180 | typer.echo("") |
| 181 | if result.emotion_arc: |
| 182 | typer.echo(f"Emotion arc: {' → '.join(result.emotion_arc)}") |
| 183 | if result.section_order: |
| 184 | typer.echo(f"Sections: {' → '.join(result.section_order)}") |
| 185 | |
| 186 | |
| 187 | def _entry_to_dict(entry: MuseTimelineEntry) -> _TimelineEntryDict: |
| 188 | """Serialize a :class:`MuseTimelineEntry` to a JSON-safe dict.""" |
| 189 | return { |
| 190 | "commit_id": entry.commit_id, |
| 191 | "short_id": entry.short_id, |
| 192 | "committed_at": entry.committed_at.isoformat(), |
| 193 | "message": entry.message, |
| 194 | "emotion": entry.emotion, |
| 195 | "sections": list(entry.sections), |
| 196 | "tracks": list(entry.tracks), |
| 197 | "activity": entry.activity, |
| 198 | } |
| 199 | |
| 200 | |
| 201 | def _render_json(result: MuseTimelineResult) -> None: |
| 202 | """Emit the timeline as a JSON object for UI rendering or agent consumption.""" |
| 203 | payload: _TimelineJsonPayload = { |
| 204 | "branch": result.branch, |
| 205 | "total_commits": result.total_commits, |
| 206 | "emotion_arc": list(result.emotion_arc), |
| 207 | "section_order": list(result.section_order), |
| 208 | "entries": [_entry_to_dict(e) for e in result.entries], |
| 209 | } |
| 210 | typer.echo(json.dumps(payload, indent=2)) |
| 211 | |
| 212 | |
| 213 | # --------------------------------------------------------------------------- |
| 214 | # Testable async core |
| 215 | # --------------------------------------------------------------------------- |
| 216 | |
| 217 | |
| 218 | async def _timeline_async( |
| 219 | *, |
| 220 | root: pathlib.Path, |
| 221 | session: AsyncSession, |
| 222 | commit_range: str | None, |
| 223 | show_emotion: bool, |
| 224 | show_sections: bool, |
| 225 | show_tracks: bool, |
| 226 | as_json: bool, |
| 227 | limit: int, |
| 228 | ) -> MuseTimelineResult: |
| 229 | """Core timeline logic — fully injectable for tests. |
| 230 | |
| 231 | Reads repo state from ``.muse/``, loads commits + tags from the DB |
| 232 | session, then renders the result. Returns the :class:`MuseTimelineResult` |
| 233 | so callers can inspect it without parsing printed output. |
| 234 | |
| 235 | Args: |
| 236 | root: Repository root (contains ``.muse/``). |
| 237 | session: Open async DB session. |
| 238 | commit_range: Optional range string (reserved for future use). |
| 239 | show_emotion: Include emotion column in text output. |
| 240 | show_sections: Group commits by section in text output. |
| 241 | show_tracks: Include tracks column in text output. |
| 242 | as_json: Emit JSON instead of the text table. |
| 243 | limit: Maximum commits to include. |
| 244 | |
| 245 | Returns: |
| 246 | :class:`MuseTimelineResult` (oldest-first). |
| 247 | """ |
| 248 | repo_id, branch, head_commit_id = _load_muse_state(root) |
| 249 | |
| 250 | if not head_commit_id: |
| 251 | typer.echo(f"No commits yet on branch {branch} — timeline is empty.") |
| 252 | raise typer.Exit(code=ExitCode.SUCCESS) |
| 253 | |
| 254 | if commit_range is not None: |
| 255 | typer.echo( |
| 256 | f"⚠️ Commit range '{commit_range}' is reserved for a future iteration. " |
| 257 | "Showing full history." |
| 258 | ) |
| 259 | |
| 260 | result = await build_timeline( |
| 261 | session, |
| 262 | repo_id=repo_id, |
| 263 | branch=branch, |
| 264 | head_commit_id=head_commit_id, |
| 265 | limit=limit, |
| 266 | ) |
| 267 | |
| 268 | if as_json: |
| 269 | _render_json(result) |
| 270 | else: |
| 271 | _render_text( |
| 272 | result, |
| 273 | show_emotion=show_emotion, |
| 274 | show_sections=show_sections, |
| 275 | show_tracks=show_tracks, |
| 276 | ) |
| 277 | |
| 278 | return result |
| 279 | |
| 280 | |
| 281 | # --------------------------------------------------------------------------- |
| 282 | # Typer command |
| 283 | # --------------------------------------------------------------------------- |
| 284 | |
| 285 | |
| 286 | @app.callback(invoke_without_command=True) |
| 287 | def timeline( |
| 288 | ctx: typer.Context, |
| 289 | commit_range: str | None = typer.Argument( |
| 290 | None, |
| 291 | help="Commit range (reserved — full history is always shown for now).", |
| 292 | metavar="RANGE", |
| 293 | ), |
| 294 | show_emotion: bool = typer.Option( |
| 295 | False, |
| 296 | "--emotion", |
| 297 | help="Add an emotion column (derived from emotion:* tags).", |
| 298 | ), |
| 299 | show_sections: bool = typer.Option( |
| 300 | False, |
| 301 | "--sections", |
| 302 | help="Group commits under section headers (derived from section:* tags).", |
| 303 | ), |
| 304 | show_tracks: bool = typer.Option( |
| 305 | False, |
| 306 | "--tracks", |
| 307 | help="Show per-track activity column (derived from track:* tags).", |
| 308 | ), |
| 309 | as_json: bool = typer.Option( |
| 310 | False, |
| 311 | "--json", |
| 312 | help="Emit structured JSON suitable for UI rendering or agent consumption.", |
| 313 | ), |
| 314 | limit: int = typer.Option( |
| 315 | _DEFAULT_LIMIT, |
| 316 | "--limit", |
| 317 | "-n", |
| 318 | help="Maximum number of commits to walk.", |
| 319 | min=1, |
| 320 | ), |
| 321 | ) -> None: |
| 322 | """Visualize the musical evolution of a composition chronologically.""" |
| 323 | root = require_repo() |
| 324 | |
| 325 | async def _run() -> None: |
| 326 | async with open_session() as session: |
| 327 | await _timeline_async( |
| 328 | root=root, |
| 329 | session=session, |
| 330 | commit_range=commit_range, |
| 331 | show_emotion=show_emotion, |
| 332 | show_sections=show_sections, |
| 333 | show_tracks=show_tracks, |
| 334 | as_json=as_json, |
| 335 | limit=limit, |
| 336 | ) |
| 337 | |
| 338 | try: |
| 339 | asyncio.run(_run()) |
| 340 | except typer.Exit: |
| 341 | raise |
| 342 | except Exception as exc: |
| 343 | typer.echo(f"❌ muse timeline failed: {exc}") |
| 344 | logger.error("❌ muse timeline error: %s", exc, exc_info=True) |
| 345 | raise typer.Exit(code=ExitCode.INTERNAL_ERROR) |