mirror of
https://github.com/SikongJueluo/Mini-Nav.git
synced 2026-07-12 20:15:31 +08:00
feat(sim): add CAM correctness runner with scenario matrix and contract tests
Introduce scripts/run_cam_correctness.py — a data-driven CAM correctness checker that runs cocotb and pytest scenarios across multiple noise modes and CAM submodules (cam_top, cam_core_banked, match_engine_pipeline). Outputs structured CSV results and a paper-ready Chinese summary. Add tests/test_run_cam_correctness.py with contract tests for scenario spec validation, command building, subprocess error handling, output file generation, CLI defaults, and environment preservation.
This commit is contained in:
642
scripts/run_cam_correctness.py
Normal file
642
scripts/run_cam_correctness.py
Normal file
@@ -0,0 +1,642 @@
|
||||
#!/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.test_cam_basic",
|
||||
testcase=None,
|
||||
params={},
|
||||
noise_mode="none",
|
||||
expected="match",
|
||||
),
|
||||
Scenario(
|
||||
name="cam_top_write_noise",
|
||||
kind="cocotb",
|
||||
module="tests.test_cam_basic",
|
||||
testcase=None,
|
||||
params={
|
||||
"WRITE_NOISE_EN": "1",
|
||||
"WRITE_NOISE_RATE_NUM": "1",
|
||||
"WRITE_NOISE_RATE_DEN": "100",
|
||||
"READ_NOISE_EN": "0",
|
||||
},
|
||||
noise_mode="write_noise",
|
||||
expected="bit_flip_injected",
|
||||
),
|
||||
Scenario(
|
||||
name="cam_top_read_noise",
|
||||
kind="cocotb",
|
||||
module="tests.test_cam_basic",
|
||||
testcase=None,
|
||||
params={
|
||||
"WRITE_NOISE_EN": "0",
|
||||
"READ_NOISE_EN": "1",
|
||||
"READ_NOISE_RATE_NUM": "1",
|
||||
"READ_NOISE_RATE_DEN": "100",
|
||||
},
|
||||
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
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
# Internal: map unit scenario names to their Verilog TOPLEVEL module.
|
||||
# cam_top_* scenarios use the Makefile default (cam_top) and are omitted.
|
||||
_UNIT_TOPLEVEL: dict[str, str] = {
|
||||
"cam_core_banked": "cam_core_banked",
|
||||
"match_engine_pipeline": "match_engine_pipeline",
|
||||
"cam_write_noise": "cam_write_noise",
|
||||
"cam_read_noise": "cam_read_noise",
|
||||
}
|
||||
|
||||
|
||||
def _infer_toplevel(scenario: Scenario) -> str | None:
|
||||
"""Return the Verilog TOPLEVEL for *scenario* or None to use Makefile default."""
|
||||
return _UNIT_TOPLEVEL.get(scenario.name)
|
||||
|
||||
|
||||
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/test_ref_model_noise.py",
|
||||
"-q",
|
||||
]
|
||||
|
||||
# cocotb
|
||||
cmd = [
|
||||
"make",
|
||||
"-C",
|
||||
"hw/sim",
|
||||
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}")
|
||||
toplevel = _infer_toplevel(scenario)
|
||||
if toplevel is not None:
|
||||
cmd.append(f"TOPLEVEL={toplevel}")
|
||||
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
|
||||
return ["make", "-C", "hw/sim", "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()
|
||||
Reference in New Issue
Block a user