#!/usr/bin/env python3 """CAM correctness runner — data model, scenario matrix, and command builders.""" import csv import os import shutil import subprocess import time from dataclasses import dataclass from datetime import date as date_type from datetime import datetime from pathlib import Path from typing import Annotated, Literal import typer # Timeout for each external command invocation (30 minutes). DEFAULT_COMMAND_TIMEOUT_SEC = 1800 # --------------------------------------------------------------------------- # Types # --------------------------------------------------------------------------- ScenarioKind = Literal["cocotb", "pytest"] Status = Literal["pass", "fail"] # --------------------------------------------------------------------------- # Data model # --------------------------------------------------------------------------- @dataclass(frozen=True) class Scenario: """A single correctness-check scenario.""" name: str kind: ScenarioKind module: str | None testcase: str | None params: dict[str, str] noise_mode: str expected: str @dataclass(frozen=True) class ScenarioRun: """Result of executing a single scenario.""" scenario: Scenario status: Status returncode: int duration_sec: float command: list[str] log_path: Path results_xml: str # --------------------------------------------------------------------------- # Default scenario matrix # --------------------------------------------------------------------------- default_scenarios: list[Scenario] = [ Scenario( name="cam_top_basic", kind="cocotb", module="tests.top.no_noise.test_no_noise", testcase=None, params={}, noise_mode="none", expected="match", ), Scenario( name="cam_top_write_noise", kind="cocotb", module="tests.top.write_noise.test_write_noise", testcase=None, params={}, noise_mode="write_noise", expected="bit_flip_injected", ), Scenario( name="cam_top_read_noise", kind="cocotb", module="tests.top.read_noise.test_read_noise", testcase=None, params={}, noise_mode="read_noise", expected="read_perturbed", ), Scenario( name="cam_core_banked", kind="cocotb", module="tests.test_cam_core_banked", testcase=None, params={}, noise_mode="none", expected="read_write_aligned", ), Scenario( name="match_engine_pipeline", kind="cocotb", module="tests.test_match_engine_pipeline", testcase=None, params={}, noise_mode="none", expected="top1_hamming_correct", ), Scenario( name="cam_write_noise", kind="cocotb", module="tests.test_cam_write_noise", testcase=None, params={}, noise_mode="write_noise", expected="bit_flip_injected", ), Scenario( name="cam_read_noise", kind="cocotb", module="tests.test_cam_read_noise", testcase=None, params={}, noise_mode="read_noise", expected="read_perturbed", ), Scenario( name="ref_model_noise", kind="pytest", module=None, testcase=None, params={}, noise_mode="none", expected="reference_noise_model_pass", ), ] # --------------------------------------------------------------------------- # Command builders # --------------------------------------------------------------------------- # Map scenario names to their Makefile directory under hw/sim/tests/. # Top-level cam_top tests are now split by noise configuration. _MAKE_DIR: dict[str, str] = { "cam_top_basic": "top/no_noise", "cam_top_write_noise": "top/write_noise", "cam_top_read_noise": "top/read_noise", "cam_core_banked": "modules/cam_core_banked", "match_engine_pipeline": "modules/match_engine_pipeline", "cam_write_noise": "modules/cam_write_noise", "cam_read_noise": "modules/cam_read_noise", } def _make_dir_for(scenario: Scenario) -> str: """Return the relative Makefile directory under hw/sim/tests/ for *scenario*.""" if scenario.name in _MAKE_DIR: return _MAKE_DIR[scenario.name] # Fallback: guess from module path (e.g. tests.top.foo → top/foo) parts = scenario.module.split(".") if scenario.module else [] if len(parts) >= 4 and parts[0] == "tests": return "/".join(parts[1:-1]) # tests.top.no_noise → top/no_noise return "." def build_scenario_command( scenario: Scenario, *, num_rows: int = 512, hash_bits: int = 512, lanes: int = 8, ) -> list[str]: """Build the shell command to execute *scenario*.""" if scenario.kind == "pytest": return [ "uv", "run", "pytest", "hw/sim/tests/model/test_ref_model_noise.py", "-q", ] # cocotb — run make in the specific test subdirectory make_dir = f"hw/sim/tests/{_make_dir_for(scenario)}" cmd = [ "make", "-C", make_dir, f"NUM_ROWS={num_rows}", f"HASH_BITS={hash_bits}", f"LANES={lanes}", f"COCOTB_TEST_MODULES={scenario.module}", ] if scenario.testcase is not None: cmd.append(f"COCOTB_TESTCASE={scenario.testcase}") for key, value in scenario.params.items(): cmd.append(f"{key}={value}") return cmd def build_clean_command(scenario: Scenario) -> list[str] | None: """Build a clean command or None if cleanup is not needed.""" if scenario.kind == "pytest": return None make_dir = f"hw/sim/tests/{_make_dir_for(scenario)}" return ["make", "-C", make_dir, "clean"] # --------------------------------------------------------------------------- # Execution helpers # --------------------------------------------------------------------------- def _run_subprocess( cmd: list[str], *, cwd: Path, env: dict[str, str], timeout: int, ) -> tuple[subprocess.CompletedProcess | None, float, str | None]: """Execute *cmd*, capture output, enforce *timeout*. Returns ``(result, duration_sec, error_msg)``. On success *result* is a ``CompletedProcess`` and *error_msg* is ``None``. On failure (timeout / OSError / SubprocessError) *result* is ``None`` and *error_msg* describes what happened. """ start = time.monotonic() try: result = subprocess.run( cmd, cwd=cwd, env=env, text=True, capture_output=True, timeout=timeout, ) except subprocess.TimeoutExpired: end = time.monotonic() return None, round(end - start, 3), f"timed out after {timeout}s" except (OSError, subprocess.SubprocessError) as exc: end = time.monotonic() return None, round(end - start, 3), str(exc) end = time.monotonic() return result, round(end - start, 3), None def run_one_scenario( scenario: Scenario, *, num_rows: int = 512, hash_bits: int = 512, lanes: int = 8, output_dir: Path, repo_root: Path, ) -> ScenarioRun: """Execute a single scenario and return the result.""" logs_dir = output_dir / "logs" logs_dir.mkdir(parents=True, exist_ok=True) log_path = logs_dir / f"{scenario.name}.log" command = build_scenario_command( scenario, num_rows=num_rows, hash_bits=hash_bits, lanes=lanes ) clean_command = build_clean_command(scenario) env = os.environ.copy() if scenario.kind == "pytest": hw_sim = str(repo_root / "hw" / "sim") existing = env.get("PYTHONPATH", "") env["PYTHONPATH"] = ( hw_sim + (os.pathsep + existing) if existing else hw_sim ) log_chunks: list[str] = [] # Run clean step if applicable if clean_command is not None: log_chunks.append(f"# Command: {' '.join(clean_command)}\n") clean_result, clean_duration, clean_err = _run_subprocess( clean_command, cwd=repo_root, env=env, timeout=DEFAULT_COMMAND_TIMEOUT_SEC, ) if clean_result is not None: log_chunks.append(clean_result.stdout) log_chunks.append(clean_result.stderr) else: log_chunks.append(f"# ERROR: {clean_err}\n") if clean_err is not None or ( clean_result is not None and clean_result.returncode != 0 ): returncode = clean_result.returncode if clean_result is not None else -1 log_path.write_text("".join(log_chunks)) return ScenarioRun( scenario=scenario, status="fail", returncode=returncode, duration_sec=clean_duration, command=clean_command, log_path=Path("logs") / f"{scenario.name}.log", results_xml="", ) # Run main command log_chunks.append(f"# Command: {' '.join(command)}\n") main_result, duration, main_err = _run_subprocess( command, cwd=repo_root, env=env, timeout=DEFAULT_COMMAND_TIMEOUT_SEC, ) if main_result is not None: log_chunks.append(main_result.stdout) log_chunks.append(main_result.stderr) else: log_chunks.append(f"# ERROR: {main_err}\n") if main_err is not None: status: Status = "fail" returncode = -1 else: status = "pass" if main_result.returncode == 0 else "fail" returncode = main_result.returncode log_path.write_text("".join(log_chunks)) # Stabilise cocotb results.xml when the file was produced if scenario.kind == "cocotb": src_xml = repo_root / "hw" / "sim" / "results.xml" if src_xml.exists(): dst_xml = logs_dir / f"{scenario.name}.results.xml" shutil.copy2(str(src_xml), str(dst_xml)) results_xml = f"logs/{scenario.name}.results.xml" else: results_xml = "" else: results_xml = "" return ScenarioRun( scenario=scenario, status=status, returncode=returncode, duration_sec=duration, command=command, log_path=Path("logs") / f"{scenario.name}.log", results_xml=results_xml, ) def result_rows( runs: list[ScenarioRun], *, num_rows: int = 512, hash_bits: int = 512, lanes: int = 8, ) -> list[dict[str, str]]: """Convert runs to CSV-ready dict rows, inserting derived rows.""" rows: list[dict[str, str]] = [] for run in runs: actual = run.scenario.expected if run.status == "pass" else "failed" row: dict[str, str] = { "test_name": run.scenario.name, "status": run.status, "num_rows": str(num_rows), "hash_bits": str(hash_bits), "lanes": str(lanes), "noise_mode": run.scenario.noise_mode, "expected": run.scenario.expected, "actual": actual, "log_path": str(run.log_path), } rows.append(row) # Insert derived popcount_pipeline_indirect row after match_engine_pipeline if run.scenario.name == "match_engine_pipeline": rows.append({ "test_name": "popcount_pipeline_indirect", "status": run.status, "num_rows": str(num_rows), "hash_bits": str(hash_bits), "lanes": str(lanes), "noise_mode": "none", "expected": "hamming_score_correct", "actual": "hamming_score_correct" if run.status == "pass" else "failed", "log_path": str(run.log_path), }) return rows def write_results_csv(output_dir: Path, rows: list[dict[str, str]]) -> None: """Write results CSV with fixed header.""" fieldnames = [ "test_name", "status", "num_rows", "hash_bits", "lanes", "noise_mode", "expected", "actual", "log_path", ] path = output_dir / "results.csv" with open(path, "w", newline="") as f: writer = csv.DictWriter(f, fieldnames=fieldnames) writer.writeheader() writer.writerows(rows) def write_raw_results_csv(output_dir: Path, runs: list[ScenarioRun]) -> None: """Write raw results CSV with fixed header.""" fieldnames = [ "scenario", "status", "returncode", "duration_sec", "command", "log_path", "results_xml", ] path = output_dir / "raw_results.csv" with open(path, "w", newline="") as f: writer = csv.DictWriter(f, fieldnames=fieldnames) writer.writeheader() for run in runs: writer.writerow({ "scenario": run.scenario.name, "status": run.status, "returncode": run.returncode, "duration_sec": run.duration_sec, "command": " ".join(run.command), "log_path": str(run.log_path), "results_xml": run.results_xml, }) def write_summary_md( output_dir: Path, rows: list[dict[str, str]], *, run_datetime: str, num_rows: int = 512, hash_bits: int = 512, lanes: int = 8, ) -> None: """Write a Chinese summary markdown file with scenario table, failure list, coverage notes, and a paper-friendly conclusion draft.""" total = len(rows) failed = sum(1 for r in rows if r["status"] == "fail") passed = total - failed failed_rows = [r for r in rows if r["status"] == "fail"] lines: list[str] = [] lines.append("# CAM 功能正确性验证摘要") lines.append("") lines.append(f"**运行日期：** {run_datetime}") lines.append("") lines.append("**参数：**") lines.append(f"- NUM_ROWS={num_rows}") lines.append(f"- HASH_BITS={hash_bits}") lines.append(f"- LANES={lanes}") lines.append("") lines.append(f"**输出目录：** {output_dir}") lines.append("") lines.append(f"**总场景数：** {total}") lines.append(f"**通过数：** {passed}") lines.append(f"**失败数：** {failed}") lines.append("") lines.append("## 场景结果") lines.append("") lines.append("| 场景 | 状态 | 噪声模式 | 预期 | 实际 | 日志 |") lines.append("|------|------|----------|------|------|------|") for r in rows: lines.append( f"| {r['test_name']} | {r['status']} | {r['noise_mode']} " f"| {r['expected']} | {r['actual']} | {r['log_path']} |" ) lines.append("") lines.append("## 失败场景") lines.append("") if failed_rows: for r in failed_rows: lines.append(f"- **{r['test_name']}**: 日志路径 `{r['log_path']}`") else: lines.append("- 无") lines.append("") lines.append("## 覆盖说明") lines.append("") lines.append( "- **cam_top**: 通过 `cam_top_basic`、`cam_top_write_noise`、" "`cam_top_read_noise` 三个场景覆盖顶层 CAM 模块的基本功能、" "写入噪声注入和读取噪声注入路径" ) lines.append( "- **cam_core_banked**: 通过 `cam_core_banked` 场景覆盖" "分块 CAM 核心的读写对齐验证" ) lines.append( "- **match_engine_pipeline**: 通过 `match_engine_pipeline`" " 场景覆盖流水线匹配引擎的 Top-1 汉明距离计算" ) lines.append( "- **popcount_pipeline**: 通过 `match_engine_pipeline`" " 的汉明匹配路径间接覆盖 popcount 流水线" ) lines.append( "- **cam_write_noise**: 通过 `cam_write_noise` 场景独立覆盖" "写入噪声注入路径" ) lines.append( "- **cam_read_noise**: 通过 `cam_read_noise` 场景独立覆盖" "读取噪声注入路径" ) lines.append("") lines.append("## 论文可引用结论草稿") lines.append("") lines.append( f"本测试套件在 {run_datetime} 对 CAM 设计进行了全面的功能正确性验证。" f"测试参数为 NUM_ROWS={num_rows}、HASH_BITS={hash_bits}、" f"LANES={lanes}。共执行 {total} 个测试场景，通过 {passed} 个，" f"失败 {failed} 个。详细结果见 `results.csv` 和 `raw_results.csv`。" f"日志文件位于 `logs/` 目录下，可用于深入调试与复现。" ) lines.append("") path = output_dir / "summary.md" path.write_text("\n".join(lines)) def run_all( *, scenarios: list[Scenario], num_rows: int = 512, hash_bits: int = 512, lanes: int = 8, output_root: str | Path = Path("outputs/cam_correctness"), date: str | None = None, repo_root: Path | None = None, ) -> int: """Run all scenarios and produce all output files. Output is written under *output_root* / *date*. Returns 0 if all scenario *runs* pass, otherwise 1. """ if repo_root is None: repo_root = Path(__file__).resolve().parent.parent output_root = Path(output_root) if date is None: date = datetime.now().strftime("%Y-%m-%d") output_dir = output_root / date output_dir.mkdir(parents=True, exist_ok=True) runs: list[ScenarioRun] = [] for scenario in scenarios: run = run_one_scenario( scenario, num_rows=num_rows, hash_bits=hash_bits, lanes=lanes, output_dir=output_dir, repo_root=repo_root, ) runs.append(run) rows = result_rows( runs, num_rows=num_rows, hash_bits=hash_bits, lanes=lanes, ) write_results_csv(output_dir, rows) write_raw_results_csv(output_dir, runs) run_datetime = datetime.now().strftime("%Y-%m-%d %H:%M:%S") write_summary_md( output_dir, rows, run_datetime=run_datetime, num_rows=num_rows, hash_bits=hash_bits, lanes=lanes, ) all_pass = all(run.status == "pass" for run in runs) return 0 if all_pass else 1 # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def script_repo_root() -> Path: """Return the repository root (two levels up from scripts/run_cam_correctness.py).""" return Path(__file__).resolve().parents[1] # --------------------------------------------------------------------------- # CLI # --------------------------------------------------------------------------- app = typer.Typer( add_completion=False, help=( "Run CAM Cocotb/Verilator correctness checks " "and collect paper-ready outputs." ), ) @app.command() def main( num_rows: Annotated[int, typer.Option('--num-rows', help='CAM row count passed to NUM_ROWS.', min=1)] = 512, hash_bits: Annotated[int, typer.Option('--hash-bits', help='Hash width passed to HASH_BITS.', min=1)] = 512, lanes: Annotated[int, typer.Option('--lanes', help='CAM lane count passed to LANES.', min=1)] = 8, output_root: Annotated[Path, typer.Option('--output-root', help='Root directory for dated correctness outputs.')] = Path('outputs/cam_correctness'), date: Annotated[str | None, typer.Option('--date', help='Output date directory. Defaults to today.')] = None, ) -> None: """Run the full CAM correctness suite.""" run_date = date or date_type.today().isoformat() exit_code = run_all( scenarios=default_scenarios, num_rows=num_rows, hash_bits=hash_bits, lanes=lanes, output_root=output_root, date=run_date, repo_root=script_repo_root(), ) raise typer.Exit(exit_code) if __name__ == "__main__": app()