#!/usr/bin/env python3 """Small Ralph/prd.json progress dashboard for local project tasks. Usage: python scripts/ralph_progress.py python scripts/ralph_progress.py .scratch/distributed-inference-network/prd.json """ from __future__ import annotations import json import shutil import sys from pathlib import Path from textwrap import shorten REPO_ROOT = Path(__file__).resolve().parents[1] DEFAULT_PRD = REPO_ROOT / ".scratch/distributed-inference-network/prd.json" USAGE = "Usage: ralph_progress.py [path/to/prd.json]" def _status_symbol(passes: bool, blocked: bool) -> str: if passes: return "✓" if blocked: return "•" return "→" def _bar(done: int, total: int, width: int = 28) -> str: if total <= 0: return "[" + "-" * width + "]" filled = round(width * done / total) return "[" + "#" * filled + "-" * (width - filled) + "]" def _load_prd(path: Path) -> dict: try: return json.loads(path.read_text(encoding="utf-8")) except FileNotFoundError: raise SystemExit(f"PRD not found: {path}") except json.JSONDecodeError as exc: raise SystemExit(f"Invalid JSON in {path}: {exc}") def _deps_done(story: dict, story_by_id: dict[str, dict]) -> bool: return all(story_by_id.get(dep, {}).get("passes") for dep in story.get("dependsOn", [])) def main(argv: list[str]) -> int: if any(arg in {"-h", "--help"} for arg in argv[1:]): print(USAGE) return 0 if len(argv) > 2: raise SystemExit(USAGE) path = Path(argv[1]) if len(argv) == 2 else DEFAULT_PRD data = _load_prd(path) stories = data.get("userStories", []) story_by_id = {story.get("id", ""): story for story in stories} total = len(stories) done = sum(1 for story in stories if story.get("passes")) ready = [story for story in stories if not story.get("passes") and _deps_done(story, story_by_id)] blocked = [story for story in stories if not story.get("passes") and not _deps_done(story, story_by_id)] columns = shutil.get_terminal_size((100, 24)).columns title = data.get("name", path.stem) percent = int(done * 100 / total) if total else 0 print(f"Ralph progress: {title}") print(f"PRD: {path}") print(f"{_bar(done, total)} {done}/{total} complete ({percent}%)") print(f"Ready: {len(ready)} Blocked: {len(blocked)}") print() for story in stories: story_id = story.get("id", "?") title = story.get("title", "(untitled)") passes = bool(story.get("passes")) deps_ok = _deps_done(story, story_by_id) symbol = _status_symbol(passes, blocked=not deps_ok) dep_text = "" if not passes and story.get("dependsOn"): unmet = [dep for dep in story["dependsOn"] if not story_by_id.get(dep, {}).get("passes")] if unmet: dep_text = f" waits: {', '.join(unmet)}" line = f"{symbol} {story_id:<6} {title}{dep_text}" print(shorten(line, width=columns, placeholder="…")) if ready: next_story = sorted(ready, key=lambda item: item.get("priority", 9999))[0] print() print(f"Next ready task: {next_story.get('id')} — {next_story.get('title')}") notes = next_story.get("notes") if notes: print(notes) return 0 if __name__ == "__main__": raise SystemExit(main(sys.argv))