| |
| """Build the interactive research-roadmap data contract for the public site.""" |
|
|
| from __future__ import annotations |
|
|
| import json |
| from datetime import datetime, timezone |
| from pathlib import Path |
| from typing import Any |
|
|
|
|
| ROOT = Path(__file__).resolve().parents[1] |
| DOCS_DATA = ROOT / "docs" / "data" |
| RESULTS = ROOT / "results" / "episode_task_suite" |
| GITHUB_BLOB = "https://github.com/ChaoYue0307/ropedia-xperience-10m-task-suite/blob/main" |
|
|
|
|
| def repo_link(path: str) -> str: |
| return f"{GITHUB_BLOB}/{path}" |
|
|
|
|
| def load_json(path: Path) -> Any: |
| return json.loads(path.read_text(encoding="utf-8")) |
|
|
|
|
| def rounded_metric(value: Any) -> float | None: |
| if value is None: |
| return None |
| try: |
| return round(float(value), 4) |
| except (TypeError, ValueError): |
| return None |
|
|
|
|
| def metric_summary(metric: dict[str, Any] | None) -> dict[str, Any]: |
| if not metric: |
| return {} |
| return { |
| "key": metric.get("key"), |
| "name": metric.get("name"), |
| "direction": metric.get("direction"), |
| "minimal": rounded_metric(metric.get("minimal")), |
| "neural_mlp": rounded_metric(metric.get("neural_mlp")), |
| "better_baseline": metric.get("better_baseline"), |
| } |
|
|
|
|
| def task_evidence_links(task_id: str) -> list[dict[str, str]]: |
| candidates = [ |
| ("Minimal metrics", f"results/episode_task_suite/{task_id}/metrics.json"), |
| ("Neural metrics", f"results/episode_task_suite/neural_mlp/{task_id}/metrics.json"), |
| ("Minimal predictions", f"results/episode_task_suite/{task_id}/predictions.csv"), |
| ("Neural predictions", f"results/episode_task_suite/neural_mlp/{task_id}/predictions.csv"), |
| ("Confusion matrix", f"results/episode_task_suite/{task_id}/confusion_matrix.csv"), |
| ("Neural confusion matrix", f"results/episode_task_suite/neural_mlp/{task_id}/confusion_matrix.csv"), |
| ] |
| links = [ |
| {"label": "Task walkthrough", "href": "data/task_walkthroughs.json"}, |
| {"label": "Single-episode explorer", "href": "single_episode_explorer.html"}, |
| ] |
| for label, relative_path in candidates: |
| if (ROOT / relative_path).exists(): |
| links.append({"label": label, "href": repo_link(relative_path)}) |
| return links |
|
|
|
|
| def task_payload( |
| task_id: str, |
| direction_task: dict[str, Any], |
| walkthrough: dict[str, Any], |
| ) -> dict[str, Any]: |
| metric = direction_task.get("metric") or walkthrough.get("metric") or {} |
| return { |
| "id": task_id, |
| "display_name": walkthrough.get("display_name") or direction_task.get("name") or task_id, |
| "research_name": walkthrough.get("research_name") or direction_task.get("name") or task_id, |
| "family": direction_task.get("family") or walkthrough.get("task_family"), |
| "architecture_family": walkthrough.get("architecture_family"), |
| "primary_direction": direction_task.get("primary_direction"), |
| "direction_roles": direction_task.get("direction_roles", {}), |
| "modalities": walkthrough.get("modalities", []), |
| "case_study": walkthrough.get("case_study"), |
| "input": walkthrough.get("input"), |
| "input_short": walkthrough.get("input_short"), |
| "process_short": walkthrough.get("process_short"), |
| "output_short": walkthrough.get("output_short"), |
| "module_summary": walkthrough.get("module_summary"), |
| "current_limit": direction_task.get("current_limit") or walkthrough.get("failure_mode"), |
| "why": direction_task.get("why"), |
| "metric": metric_summary(metric), |
| "evidence_links": task_evidence_links(task_id), |
| } |
|
|
|
|
| def phase_payload(phases: list[dict[str, Any]]) -> list[dict[str, Any]]: |
| stage_map = { |
| "implemented": "now", |
| "active": "scale_up", |
| "next": "omni", |
| "planned": "future", |
| } |
| return [ |
| { |
| "id": phase.get("id"), |
| "name": phase.get("name"), |
| "status": phase.get("status"), |
| "stage": stage_map.get(str(phase.get("status", "")).lower(), "future"), |
| "entry_condition": phase.get("entry_condition"), |
| "deliverables": phase.get("deliverables", []), |
| "completion_evidence": phase.get("completion_evidence", []), |
| "reader_takeaway": phase.get("reader_takeaway"), |
| } |
| for phase in phases |
| ] |
|
|
|
|
| def main() -> int: |
| directions_doc = load_json(DOCS_DATA / "research_directions.json") |
| walkthroughs = load_json(DOCS_DATA / "task_walkthroughs.json") |
| roadmap = load_json(DOCS_DATA / "research_roadmap.json") |
| foundation_plan = load_json(DOCS_DATA / "foundation_model_plan.json") |
| summary_metrics = load_json(DOCS_DATA / "summary_metrics.json") |
| episode_summary = load_json(RESULTS / "summary_report.json") |
| feature_manifest = load_json(RESULTS / "feature_manifest.json") |
| extension_doc = load_json(DOCS_DATA / "research_direction_extensions.json") |
|
|
| tasks: dict[str, dict[str, Any]] = {} |
| for task_id, direction_task in directions_doc.get("tasks", {}).items(): |
| tasks[task_id] = task_payload( |
| task_id, |
| direction_task, |
| walkthroughs.get("tasks", {}).get(task_id, {}), |
| ) |
|
|
| directions = [] |
| for code, direction in directions_doc.get("directions", {}).items(): |
| linked_tasks = [tasks[task_id] for task_id in direction.get("tasks", []) if task_id in tasks] |
| extension_tasks = [ |
| { |
| "id": task_id, |
| "name": spec.get("name"), |
| "family": spec.get("family"), |
| "metric_name": spec.get("metric_name"), |
| "current_limit": spec.get("current_limit"), |
| } |
| for task_id, spec in extension_doc.get("task_specs", {}).items() |
| if spec.get("direction") == code |
| ] |
| directions.append( |
| { |
| "code": code, |
| "id": direction.get("id"), |
| "name": direction.get("name"), |
| "focus": direction.get("focus"), |
| "preferred_background": direction.get("preferred_background"), |
| "current_status": direction.get("current_status"), |
| "current_readout": direction.get("current_readout"), |
| "next_steps": direction.get("next_steps", []), |
| "counts": direction.get("counts", {}), |
| "task_ids": direction.get("tasks", []), |
| "tasks": linked_tasks, |
| "extension_tasks": extension_tasks, |
| } |
| ) |
|
|
| omni = summary_metrics.get("omni_relay", {}) |
| payload = { |
| "title": "Interactive Research Roadmap", |
| "generated_at_utc": datetime.now(timezone.utc).isoformat(timespec="seconds"), |
| "source_files": [ |
| "docs/data/research_directions.json", |
| "docs/data/task_walkthroughs.json", |
| "docs/data/research_roadmap.json", |
| "docs/data/foundation_model_plan.json", |
| "docs/data/summary_metrics.json", |
| "docs/data/research_direction_extensions.json", |
| "results/episode_task_suite/summary_report.json", |
| "results/episode_task_suite/feature_manifest.json", |
| ], |
| "scope": { |
| "sample_episode_count": walkthroughs.get("scope", {}).get("episode_count", 1), |
| "num_frames": episode_summary.get("num_frames"), |
| "num_windows": episode_summary.get("num_windows"), |
| "feature_dim": episode_summary.get("feature_dim"), |
| "window_frames": episode_summary.get("window_frames"), |
| "stride_frames": episode_summary.get("stride_frames"), |
| "feature_blocks": len(feature_manifest), |
| "warning": walkthroughs.get("scope", {}).get("warning"), |
| }, |
| "baseline_summary": { |
| "task_count": len(tasks), |
| "baseline_heads": "minimal and neural MLP heads", |
| "split": "chronological single-episode split for public-sample diagnostics", |
| "current_use": "task design, data-contract validation, case studies, and baseline comparison", |
| }, |
| "scale_up": { |
| "target_episodes": omni.get("target_episodes"), |
| "candidate_scan_top_level_sessions": omni.get("candidate_scan_top_level_sessions"), |
| "valid_candidates": omni.get("valid_candidates"), |
| "estimated_bytes": omni.get("estimated_bytes"), |
| "status": omni.get("status"), |
| "access_status": omni.get("access_status"), |
| "exclude": omni.get("exclude", []), |
| "selection_strategy": omni.get("selection_strategy"), |
| }, |
| "omni_plan": { |
| "backbone": "Qwen/Qwen3-Omni-30B-A3B-Instruct", |
| "adapter": "LoRA rank 16, alpha 32, dropout 0.05", |
| "first_pilot": "32 held-out-episode pilot after valid episodes are staged", |
| "training_unit": "episode-level split, window-level supervised examples", |
| "evaluation": [ |
| "JSON validity", |
| "action macro-F1", |
| "subtask accuracy", |
| "transition accuracy", |
| "next-action accuracy", |
| "contact accuracy", |
| "object micro-F1", |
| "held-out episode count", |
| ], |
| }, |
| "foundation_model_plan": { |
| "status": foundation_plan.get("status"), |
| "decision": foundation_plan.get("decision", {}), |
| "model_families": foundation_plan.get("model_families", []), |
| "execution_order": foundation_plan.get("execution_order", []), |
| "evaluation_additions": foundation_plan.get("evaluation_additions", []), |
| "source_links": foundation_plan.get("source_links", []), |
| }, |
| "phases": phase_payload(roadmap.get("phases", [])), |
| "directions": directions, |
| "tasks": list(tasks.values()), |
| } |
|
|
| out_path = DOCS_DATA / "research_roadmap_interactive.json" |
| out_path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8") |
| print(f"Wrote {out_path}") |
| return 0 |
|
|
|
|
| if __name__ == "__main__": |
| raise SystemExit(main()) |
|
|
