| """ |
| Dashboard - ASCII-formatted system status display for the Codette training lab. |
| |
| Shows: |
| - Latest training run stats |
| - Best adapter scores |
| - Dataset sizes and quality |
| - Failure rates |
| - Improvement trends |
| |
| No web framework required; pure terminal output. |
| """ |
|
|
| from __future__ import annotations |
|
|
| import argparse |
| import json |
| import os |
| import sys |
| from datetime import datetime |
| from pathlib import Path |
| from typing import Any, Dict, List, Optional |
|
|
| _THIS_DIR = Path(__file__).resolve().parent |
| _PROJECT_ROOT = _THIS_DIR.parent |
| if str(_PROJECT_ROOT) not in sys.path: |
| sys.path.insert(0, str(_PROJECT_ROOT)) |
|
|
| from observatory.metrics_logger import MetricsLogger |
| from observatory.performance_tracker import PerformanceTracker |
| from observatory.dataset_quality_monitor import DatasetQualityMonitor |
|
|
|
|
| class Dashboard: |
| """ASCII dashboard for the Codette training lab.""" |
|
|
| WIDTH = 76 |
|
|
| def __init__( |
| self, |
| metrics_log: Optional[str] = None, |
| quality_log: Optional[str] = None, |
| eval_results: Optional[str] = None, |
| ): |
| self.logger = MetricsLogger(log_file=metrics_log) |
| self.tracker = PerformanceTracker(logger=self.logger) |
| self.quality_monitor = DatasetQualityMonitor(quality_file=quality_log) |
| self.eval_results_path = eval_results |
|
|
| |
|
|
| def _header(self) -> List[str]: |
| lines = [] |
| lines.append("") |
| lines.append("+" + "=" * (self.WIDTH - 2) + "+") |
| lines.append("|" + " CODETTE TRAINING LAB OBSERVATORY ".center(self.WIDTH - 2) + "|") |
| lines.append("|" + f" {datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S UTC')} ".center(self.WIDTH - 2) + "|") |
| lines.append("+" + "=" * (self.WIDTH - 2) + "+") |
| return lines |
|
|
| def _section(self, title: str) -> List[str]: |
| lines = [] |
| lines.append("") |
| lines.append("+" + "-" * (self.WIDTH - 2) + "+") |
| lines.append("|" + f" {title} ".ljust(self.WIDTH - 2) + "|") |
| lines.append("+" + "-" * (self.WIDTH - 2) + "+") |
| return lines |
|
|
| def _row(self, label: str, value: str) -> str: |
| """Single label: value row.""" |
| content = f" {label:<30s} {value}" |
| return "|" + content.ljust(self.WIDTH - 2) + "|" |
|
|
| def _bar_row(self, label: str, value: float, max_width: int = 30) -> str: |
| """Row with ASCII progress bar.""" |
| filled = int(value * max_width) |
| bar = "[" + "#" * filled + "." * (max_width - filled) + "]" |
| content = f" {label:<22s} {value:>6.3f} {bar}" |
| return "|" + content.ljust(self.WIDTH - 2) + "|" |
|
|
| def _empty_row(self) -> str: |
| return "|" + " " * (self.WIDTH - 2) + "|" |
|
|
| def _footer(self) -> List[str]: |
| return ["+" + "=" * (self.WIDTH - 2) + "+", ""] |
|
|
| |
|
|
| def _latest_training_section(self) -> List[str]: |
| lines = self._section("LATEST TRAINING RUN") |
|
|
| latest = self.logger.get_latest() |
| if not latest: |
| lines.append(self._row("Status", "No training runs logged yet")) |
| return lines |
|
|
| lines.append(self._row("Adapter", latest.get("adapter", "N/A"))) |
| lines.append(self._row("Timestamp", latest.get("timestamp", "N/A"))) |
| lines.append(self._row("Dataset Version", latest.get("dataset_version", "N/A"))) |
| lines.append(self._row("Dataset Size", str(latest.get("dataset_size", 0)))) |
| lines.append(self._row("Epoch", str(latest.get("epoch", 0)))) |
| lines.append(self._bar_row("Reasoning Score", latest.get("reasoning_score", 0))) |
| lines.append(self._row("Loss", f"{latest.get('loss', 0):.6f}")) |
|
|
| params = latest.get("training_params", {}) |
| if params: |
| lines.append(self._empty_row()) |
| lines.append(self._row("Training Parameters", "")) |
| for k, v in list(params.items())[:6]: |
| lines.append(self._row(f" {k}", str(v))) |
|
|
| return lines |
|
|
| def _best_adapters_section(self) -> List[str]: |
| lines = self._section("TOP ADAPTERS") |
|
|
| best = self.tracker.best_adapters(top_n=5) |
| if not best: |
| lines.append(self._row("Status", "No adapter data available")) |
| return lines |
|
|
| |
| hdr = f" {'#':<3} {'Adapter':<26} {'Score':>7} {'Loss':>7} {'Epoch':>5}" |
| lines.append("|" + hdr.ljust(self.WIDTH - 2) + "|") |
| sep = f" {'--':<3} {'------':<26} {'-----':>7} {'----':>7} {'-----':>5}" |
| lines.append("|" + sep.ljust(self.WIDTH - 2) + "|") |
|
|
| for i, entry in enumerate(best, 1): |
| name = entry.get("adapter", "?")[:25] |
| score = entry.get("reasoning_score", 0) |
| loss = entry.get("loss", 0) |
| epoch = entry.get("epoch", 0) |
| row = f" {i:<3} {name:<26} {score:>7.4f} {loss:>7.4f} {epoch:>5}" |
| lines.append("|" + row.ljust(self.WIDTH - 2) + "|") |
|
|
| return lines |
|
|
| def _dataset_quality_section(self) -> List[str]: |
| lines = self._section("DATASET QUALITY") |
|
|
| latest = self.quality_monitor.get_latest() |
| if not latest: |
| lines.append(self._row("Status", "No quality data recorded")) |
| return lines |
|
|
| lines.append(self._row("Dataset Version", latest.get("dataset_version", "N/A"))) |
| lines.append(self._row("Total Examples", str(latest.get("total_examples", 0)))) |
| lines.append(self._row("Valid Examples", str(latest.get("valid_examples", 0)))) |
| lines.append(self._bar_row("Validity Rate", latest.get("validity_rate", 0))) |
| lines.append(self._row("Avg Response Length", f"{latest.get('avg_response_length', 0):.1f} words")) |
| lines.append(self._row("Duplicate Rate", f"{latest.get('duplicate_rate', 0):.2%}")) |
| lines.append(self._row("Near-Duplicate Rate", f"{latest.get('near_duplicate_rate', 0):.2%}")) |
| lines.append(self._bar_row("Topic Diversity", min(latest.get("topic_diversity", 0) * 10, 1.0))) |
| lines.append(self._row("Topic Concentration", f"{latest.get('topic_concentration', 0):.2%}")) |
|
|
| |
| regressions = self.quality_monitor.check_latest_regressions() |
| if regressions: |
| lines.append(self._empty_row()) |
| for r in regressions: |
| sev = r["severity"].upper() |
| msg = f" [{sev}] {r['metric']}: {r['percent_change']:+.1f}%" |
| lines.append("|" + msg.ljust(self.WIDTH - 2) + "|") |
|
|
| return lines |
|
|
| def _improvement_trends_section(self) -> List[str]: |
| lines = self._section("IMPROVEMENT TRENDS") |
|
|
| trends = self.tracker.improvement_trends() |
| if not trends: |
| lines.append(self._row("Status", "Insufficient data for trends")) |
| return lines |
|
|
| for t in trends[:5]: |
| name = t["adapter"][:22] |
| delta = t["delta"] |
| pct = t["percent_change"] |
| runs = t["num_runs"] |
| sign = "+" if delta >= 0 else "" |
| indicator = "^" if delta > 0.01 else ("v" if delta < -0.01 else "=") |
|
|
| row = (f" {indicator} {name:<22} " |
| f"delta: {sign}{delta:.4f} " |
| f"({sign}{pct:.1f}%) " |
| f"[{runs} runs]") |
| lines.append("|" + row.ljust(self.WIDTH - 2) + "|") |
|
|
| return lines |
|
|
| def _failure_rates_section(self) -> List[str]: |
| lines = self._section("EVALUATION FAILURE RATES") |
|
|
| if not self.eval_results_path or not os.path.exists(self.eval_results_path): |
| lines.append(self._row("Status", "No evaluation results file specified")) |
| return lines |
|
|
| try: |
| with open(self.eval_results_path, "r", encoding="utf-8") as f: |
| results = json.load(f) |
| except (json.JSONDecodeError, OSError): |
| lines.append(self._row("Status", "Could not load evaluation results")) |
| return lines |
|
|
| |
| overall = results.get("overall", {}) |
| if overall: |
| overall_score = overall.get("overall", 0) |
| lines.append(self._bar_row("Overall Score", overall_score)) |
| lines.append(self._empty_row()) |
|
|
| |
| categories = results.get("categories", {}) |
| if categories: |
| hdr = f" {'Category':<20} {'Score':>7} {'Prompts':>8}" |
| lines.append("|" + hdr.ljust(self.WIDTH - 2) + "|") |
| sep = f" {'--------':<20} {'-----':>7} {'-------':>8}" |
| lines.append("|" + sep.ljust(self.WIDTH - 2) + "|") |
|
|
| for cat, data in sorted(categories.items()): |
| avg = data.get("average_scores", {}).get("overall", 0) |
| n = data.get("prompts_scored", 0) |
| status = "*" if avg < 0.4 else ("~" if avg < 0.55 else " ") |
| row = f" {status}{cat:<19} {avg:>7.4f} {n:>8}" |
| lines.append("|" + row.ljust(self.WIDTH - 2) + "|") |
|
|
| lines.append(self._empty_row()) |
| lines.append("|" + " * = failing, ~ = weak".ljust(self.WIDTH - 2) + "|") |
|
|
| return lines |
|
|
| def _sparkline_section(self) -> List[str]: |
| lines = self._section("SCORE HISTORY") |
|
|
| adapters = self.logger.get_unique_adapters() |
| if not adapters: |
| lines.append(self._row("Status", "No history data")) |
| return lines |
|
|
| for adapter in adapters[:6]: |
| progression = self.tracker.score_progression(adapter) |
| if not progression: |
| continue |
| scores = [p["reasoning_score"] for p in progression] |
| spark = PerformanceTracker._sparkline(scores, width=30) |
| name = adapter[:20] |
| row = f" {name:<21} {spark} [{scores[0]:.3f}->{scores[-1]:.3f}]" |
| lines.append("|" + row.ljust(self.WIDTH - 2) + "|") |
|
|
| return lines |
|
|
| |
|
|
| def render(self) -> str: |
| """Render the complete dashboard.""" |
| all_lines: List[str] = [] |
| all_lines.extend(self._header()) |
| all_lines.extend(self._latest_training_section()) |
| all_lines.extend(self._best_adapters_section()) |
| all_lines.extend(self._dataset_quality_section()) |
| all_lines.extend(self._improvement_trends_section()) |
| all_lines.extend(self._failure_rates_section()) |
| all_lines.extend(self._sparkline_section()) |
| all_lines.extend(self._footer()) |
| return "\n".join(all_lines) |
|
|
|
|
| |
| |
| |
|
|
| def main() -> None: |
| parser = argparse.ArgumentParser( |
| description="Codette Observatory Dashboard - ASCII system status display" |
| ) |
| parser.add_argument( |
| "--metrics-log", "-m", |
| default=None, |
| help="Path to observatory_metrics.json", |
| ) |
| parser.add_argument( |
| "--quality-log", "-q", |
| default=None, |
| help="Path to dataset_quality_log.json", |
| ) |
| parser.add_argument( |
| "--eval-results", "-e", |
| default=None, |
| help="Path to benchmark evaluation results JSON", |
| ) |
| parser.add_argument( |
| "--section", "-s", |
| choices=["training", "adapters", "quality", "trends", "failures", "history", "all"], |
| default="all", |
| help="Show only a specific section (default: all)", |
| ) |
|
|
| args = parser.parse_args() |
|
|
| dashboard = Dashboard( |
| metrics_log=args.metrics_log, |
| quality_log=args.quality_log, |
| eval_results=args.eval_results, |
| ) |
|
|
| if args.section == "all": |
| print(dashboard.render()) |
| else: |
| section_map = { |
| "training": dashboard._latest_training_section, |
| "adapters": dashboard._best_adapters_section, |
| "quality": dashboard._dataset_quality_section, |
| "trends": dashboard._improvement_trends_section, |
| "failures": dashboard._failure_rates_section, |
| "history": dashboard._sparkline_section, |
| } |
| func = section_map.get(args.section) |
| if func: |
| lines = dashboard._header() |
| lines.extend(func()) |
| lines.extend(dashboard._footer()) |
| print("\n".join(lines)) |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
