scripts/build_interactive_research_roadmap.py

#!/usr/bin/env python3
"""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")
    additional_directions = load_json(DOCS_DATA / "additional_development_directions.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/additional_development_directions.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 prepared",
            "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", []),
        },
        "additional_development_directions": {
            "title": additional_directions.get("title"),
            "summary": additional_directions.get("summary"),
            "status": additional_directions.get("status"),
            "public_boundary": additional_directions.get("public_boundary"),
            "directions": additional_directions.get("directions", []),
            "practical_order": additional_directions.get("practical_order", []),
            "source_document": additional_directions.get("source_document"),
        },
        "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())