Function run_level

kernelbench/scorer.py:182–243

Run all (or selected) problems in a level.

(
    level: int,
    problem_ids: Optional[List[int]] = None,
    quick: bool = False,
    backend: str = "cuda",
)

def run_level(
    level: int,
    problem_ids: Optional[List[int]] = None,
    quick: bool = False,
    backend: str = "cuda",
) -> List[Dict[str, Any]]:
    """Run all (or selected) problems in a level."""

    # Discover available problems
    level_dir = KB_CACHE_DIR / f"level{level}"
    available_ids = []
    if level_dir.exists():
        for f in sorted(level_dir.glob("*.json")):
            try:
                meta = json.loads(f.read_text(encoding="utf-8"))
                available_ids.append(meta["problem_id"])
            except (json.JSONDecodeError, KeyError):
                continue

    if not available_ids:
        print(f"No cached problems for Level {level}.")
        print(f"  Fetch first: uv run kernelbench/bridge.py fetch --source hf --level {level}")
        return []

    # Filter if specific IDs requested
    if problem_ids:
        run_ids = [pid for pid in problem_ids if pid in available_ids]
        if not run_ids:
            print(f"None of the requested problem IDs found in cache for Level {level}.")
            return []
    else:
        run_ids = available_ids

    print(f"=== KernelBench Scorer: Level {level} ({len(run_ids)} problems) ===\n")

    results = []
    scores = load_scores()

    for i, pid in enumerate(run_ids):
        print(f"[{i + 1}/{len(run_ids)}] Problem L{level}_P{pid:03d}...", end=" ", flush=True)
        t0 = time.time()

        result = run_single_problem(level, pid, quick=quick, backend=backend)
        elapsed = time.time() - t0

        result["elapsed_s"] = elapsed
        results.append(result)

        # Save incrementally
        key = f"L{level}_P{pid:03d}"
        scores["problems"][key] = result
        save_scores(scores)

        # Print one-line result
        status = result["correctness"]
        speedup = result["speedup"]
        if result.get("error"):
            print(f"ERROR ({result['error'][:60]})")