Files
Mini-Nav/scripts/run_cam_correctness.py
SikongJueluo d6e1b9d8ba refactor(cam): rename read-noise path to read-pass-through and reorganize test module structure
- Rename `read_noise` scenarios and noise_mode to `read_pass_through` across
  run_cam_correctness.py, test_run_cam_correctness.py, and Makefiles
- Update RTL comment in match_engine_pipeline.sv to reflect pass-through behavior
- Move unit-level cocotb tests from `tests.test_*` flat namespace to
  `tests.modules.*` and `tests.top.*` subdirectory layout, matching actual
  Makefile paths (hw/sim/tests/modules/..., hw/sim/tests/top/...)
- Remove redundant dual-noise subtarget from read_noise/Makefile
- Update help text and docs to reflect read-path pass-through semantics
- Add .codegraph to .gitignore
2026-05-27 16:34:34 +08:00

641 lines
20 KiB
Python

#!/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_path",
kind="cocotb",
module="tests.top.read_noise.test_read_noise",
testcase=None,
params={},
noise_mode="read_pass_through",
expected="pure_matching",
),
Scenario(
name="cam_core_banked",
kind="cocotb",
module="tests.modules.cam_core_banked.test_cam_core_banked",
testcase=None,
params={},
noise_mode="none",
expected="read_write_aligned",
),
Scenario(
name="match_engine_pipeline",
kind="cocotb",
module="tests.modules.match_engine_pipeline.test_match_engine_pipeline",
testcase=None,
params={},
noise_mode="none",
expected="top1_hamming_correct",
),
Scenario(
name="cam_write_noise",
kind="cocotb",
module="tests.modules.cam_write_noise.test_cam_write_noise",
testcase=None,
params={},
noise_mode="write_noise",
expected="bit_flip_injected",
),
Scenario(
name="cam_read_path",
kind="cocotb",
module="tests.modules.cam_read_noise.test_cam_read_noise",
testcase=None,
params={},
noise_mode="read_pass_through",
expected="pass_through",
),
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_path": "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_path": "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_path` 三个场景覆盖顶层 CAM 模块的基本功能、"
"写入噪声注入和读取路径 pass-through"
)
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_path` 场景独立覆盖"
"读取路径 pass-through"
)
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()