From a81eda26fe76a3e4f2381b428072f77b7e43be64 Mon Sep 17 00:00:00 2001 From: Rob Taylor Date: Mon, 2 Mar 2026 17:46:39 +0000 Subject: [PATCH] Add Python scripting API for batch automation and regression management Subprocess wrapper around `jacquard map` and `jacquard sim` CLI with: - JacquardConfig dataclass mirroring the Rust jacquard.toml structure - Structured result parsing for Design Statistics and Simulation Summary - Batch regression runner with parallel execution via ProcessPoolExecutor - Config override support for parameter sweeps without modifying config objects - 47 unit tests covering config round-trip, CLI arg building, output parsing, and regression runner logic Co-developed-by: Claude Code v2.1.50 (claude-opus-4-6) --- python/jacquard/pyproject.toml | 29 ++ python/jacquard/src/jacquard/__init__.py | 59 ++++ python/jacquard/src/jacquard/config.py | 266 ++++++++++++++++ python/jacquard/src/jacquard/errors.py | 31 ++ python/jacquard/src/jacquard/regression.py | 197 ++++++++++++ python/jacquard/src/jacquard/result.py | 178 +++++++++++ python/jacquard/src/jacquard/runner.py | 341 +++++++++++++++++++++ python/jacquard/tests/test_config.py | 215 +++++++++++++ python/jacquard/tests/test_regression.py | 158 ++++++++++ python/jacquard/tests/test_runner.py | 297 ++++++++++++++++++ 10 files changed, 1771 insertions(+) create mode 100644 python/jacquard/pyproject.toml create mode 100644 python/jacquard/src/jacquard/__init__.py create mode 100644 python/jacquard/src/jacquard/config.py create mode 100644 python/jacquard/src/jacquard/errors.py create mode 100644 python/jacquard/src/jacquard/regression.py create mode 100644 python/jacquard/src/jacquard/result.py create mode 100644 python/jacquard/src/jacquard/runner.py create mode 100644 python/jacquard/tests/test_config.py create mode 100644 python/jacquard/tests/test_regression.py create mode 100644 python/jacquard/tests/test_runner.py diff --git a/python/jacquard/pyproject.toml b/python/jacquard/pyproject.toml new file mode 100644 index 0000000..ed4b327 --- /dev/null +++ b/python/jacquard/pyproject.toml @@ -0,0 +1,29 @@ +[project] +name = "jacquard" +version = "0.1.0" +description = "Python API for Jacquard GPU-accelerated RTL simulator" +requires-python = ">=3.11" +dependencies = [] + +[dependency-groups] +dev = ["pytest>=7.0", "ruff"] + +[build-system] +requires = ["hatchling"] +build-backend = "hatchling.build" + +[tool.hatch.build.targets.wheel] +packages = ["src/jacquard"] + +[tool.ruff] +line-length = 100 +target-version = "py311" + +[tool.ruff.lint] +select = ["E", "F", "I", "N", "W"] + +[tool.pytest.ini_options] +testpaths = ["tests"] +filterwarnings = [ + "ignore::pytest.PytestCollectionWarning", +] diff --git a/python/jacquard/src/jacquard/__init__.py b/python/jacquard/src/jacquard/__init__.py new file mode 100644 index 0000000..3c5c585 --- /dev/null +++ b/python/jacquard/src/jacquard/__init__.py @@ -0,0 +1,59 @@ +"""Jacquard — Python API for GPU-accelerated RTL simulation. + +Subprocess wrapper around the `jacquard` CLI for batch automation, +regression testing, and result analysis. + +Example: + >>> from jacquard import JacquardConfig, sim + >>> config = JacquardConfig.from_toml(Path("jacquard.toml")) + >>> result = sim(config, max_cycles=100) + >>> assert result.success +""" + +from .config import ( + DesignConfig, + JacquardConfig, + MapConfig, + SdfConfig, + SimConfig, + TimingConfig, +) +from .errors import ( + BinaryNotFoundError, + ConfigError, + JacquardError, + MappingError, + SimulationError, +) +from .regression import RegressionReport, TestCase, TestResult, run_regression +from .result import DesignStats, MapResult, SimResult +from .runner import find_jacquard_binary, map, sim + +__all__ = [ + # Config + "JacquardConfig", + "DesignConfig", + "MapConfig", + "SimConfig", + "SdfConfig", + "TimingConfig", + # Runner + "map", + "sim", + "find_jacquard_binary", + # Results + "MapResult", + "SimResult", + "DesignStats", + # Regression + "TestCase", + "TestResult", + "RegressionReport", + "run_regression", + # Errors + "JacquardError", + "ConfigError", + "MappingError", + "SimulationError", + "BinaryNotFoundError", +] diff --git a/python/jacquard/src/jacquard/config.py b/python/jacquard/src/jacquard/config.py new file mode 100644 index 0000000..3d57c6f --- /dev/null +++ b/python/jacquard/src/jacquard/config.py @@ -0,0 +1,266 @@ +"""Jacquard configuration management — mirrors the Rust JacquardConfig / jacquard.toml.""" + +from __future__ import annotations + +import tomllib +from dataclasses import dataclass, field +from pathlib import Path + +from .errors import ConfigError + + +@dataclass +class SdfConfig: + """SDF timing back-annotation configuration.""" + + file: Path | None = None + corner: str = "typ" + debug: bool = False + + +@dataclass +class TimingConfig: + """Post-simulation timing analysis configuration.""" + + enabled: bool = False + clock_period: int | None = None # picoseconds + report_violations: bool = False + + +@dataclass +class DesignConfig: + """Shared design parameters used across subcommands.""" + + netlist: Path | None = None + top_module: str | None = None + liberty: Path | None = None + + +@dataclass +class MapConfig: + """Partition mapping settings for `jacquard map`.""" + + output: Path | None = None + level_split: list[int] = field(default_factory=list) + max_stage_degrad: int = 0 + xprop: bool = False + + +@dataclass +class SimConfig: + """Simulation settings for `jacquard sim`.""" + + gemparts: Path | None = None + input_vcd: Path | None = None + output_vcd: Path | None = None + num_blocks: int | None = None + input_vcd_scope: str | None = None + output_vcd_scope: str | None = None + max_cycles: int | None = None + check_with_cpu: bool = False + xprop: bool = False + dump_signals: list[str] = field(default_factory=list) + dump_scope: str | None = None + dump_depth: int | None = None + plusargs: dict[str, str] = field(default_factory=dict) + sdf: SdfConfig | None = None + timing: TimingConfig | None = None + + +@dataclass +class JacquardConfig: + """Top-level project configuration, mirrors `jacquard.toml`.""" + + design: DesignConfig = field(default_factory=DesignConfig) + map: MapConfig = field(default_factory=MapConfig) + sim: SimConfig = field(default_factory=SimConfig) + + @classmethod + def from_toml(cls, path: Path) -> JacquardConfig: + """Load configuration from a TOML file. + + Relative paths are resolved against the config file's parent directory. + """ + path = Path(path) + if not path.exists(): + raise ConfigError(f"Config file not found: {path}") + try: + with open(path, "rb") as f: + raw = tomllib.load(f) + except tomllib.TOMLDecodeError as e: + raise ConfigError(f"Failed to parse {path}: {e}") from e + + config_dir = path.parent + config = cls._from_dict(raw) + config._resolve_paths(config_dir) + return config + + @classmethod + def _from_dict(cls, d: dict) -> JacquardConfig: + """Build config from a parsed TOML dict.""" + design_d = d.get("design", {}) + map_d = d.get("map", {}) + sim_d = d.get("sim", {}) + + design = DesignConfig( + netlist=Path(design_d["netlist"]) if "netlist" in design_d else None, + top_module=design_d.get("top_module"), + liberty=Path(design_d["liberty"]) if "liberty" in design_d else None, + ) + + map_cfg = MapConfig( + output=Path(map_d["output"]) if "output" in map_d else None, + level_split=map_d.get("level_split", []), + max_stage_degrad=map_d.get("max_stage_degrad", 0), + xprop=map_d.get("xprop", False), + ) + + # Parse SDF sub-config + sdf_cfg = None + if "sdf" in sim_d: + sdf_d = sim_d["sdf"] + sdf_cfg = SdfConfig( + file=Path(sdf_d["file"]) if "file" in sdf_d else None, + corner=sdf_d.get("corner", "typ"), + debug=sdf_d.get("debug", False), + ) + + # Parse timing sub-config + timing_cfg = None + if "timing" in sim_d: + timing_d = sim_d["timing"] + timing_cfg = TimingConfig( + enabled=timing_d.get("enabled", False), + clock_period=timing_d.get("clock_period"), + report_violations=timing_d.get("report_violations", False), + ) + + sim_cfg = SimConfig( + gemparts=Path(sim_d["gemparts"]) if "gemparts" in sim_d else None, + input_vcd=Path(sim_d["input_vcd"]) if "input_vcd" in sim_d else None, + output_vcd=Path(sim_d["output_vcd"]) if "output_vcd" in sim_d else None, + num_blocks=sim_d.get("num_blocks"), + input_vcd_scope=sim_d.get("input_vcd_scope"), + output_vcd_scope=sim_d.get("output_vcd_scope"), + max_cycles=sim_d.get("max_cycles"), + check_with_cpu=sim_d.get("check_with_cpu", False), + xprop=sim_d.get("xprop", False), + dump_signals=sim_d.get("dump_signals", []), + dump_scope=sim_d.get("dump_scope"), + dump_depth=sim_d.get("dump_depth"), + plusargs=sim_d.get("plusargs", {}), + sdf=sdf_cfg, + timing=timing_cfg, + ) + + return cls(design=design, map=map_cfg, sim=sim_cfg) + + def to_toml(self, path: Path) -> None: + """Serialize configuration to a TOML file.""" + lines: list[str] = [] + + # [design] + lines.append("[design]") + if self.design.netlist is not None: + lines.append(f'netlist = "{self.design.netlist}"') + if self.design.top_module is not None: + lines.append(f'top_module = "{self.design.top_module}"') + if self.design.liberty is not None: + lines.append(f'liberty = "{self.design.liberty}"') + + # [map] + lines.append("") + lines.append("[map]") + if self.map.output is not None: + lines.append(f'output = "{self.map.output}"') + if self.map.level_split: + lines.append(f"level_split = {self.map.level_split}") + if self.map.max_stage_degrad != 0: + lines.append(f"max_stage_degrad = {self.map.max_stage_degrad}") + if self.map.xprop: + lines.append("xprop = true") + + # [sim] + lines.append("") + lines.append("[sim]") + if self.sim.gemparts is not None: + lines.append(f'gemparts = "{self.sim.gemparts}"') + if self.sim.input_vcd is not None: + lines.append(f'input_vcd = "{self.sim.input_vcd}"') + if self.sim.output_vcd is not None: + lines.append(f'output_vcd = "{self.sim.output_vcd}"') + if self.sim.num_blocks is not None: + lines.append(f"num_blocks = {self.sim.num_blocks}") + if self.sim.input_vcd_scope is not None: + lines.append(f'input_vcd_scope = "{self.sim.input_vcd_scope}"') + if self.sim.output_vcd_scope is not None: + lines.append(f'output_vcd_scope = "{self.sim.output_vcd_scope}"') + if self.sim.max_cycles is not None: + lines.append(f"max_cycles = {self.sim.max_cycles}") + if self.sim.check_with_cpu: + lines.append("check_with_cpu = true") + if self.sim.xprop: + lines.append("xprop = true") + if self.sim.dump_signals: + items = ", ".join(f'"{s}"' for s in self.sim.dump_signals) + lines.append(f"dump_signals = [{items}]") + if self.sim.dump_scope is not None: + lines.append(f'dump_scope = "{self.sim.dump_scope}"') + if self.sim.dump_depth is not None: + lines.append(f"dump_depth = {self.sim.dump_depth}") + if self.sim.plusargs: + lines.append("") + lines.append("[sim.plusargs]") + for k, v in sorted(self.sim.plusargs.items()): + # Quote keys that contain dots (TOML requirement) + key = f'"{k}"' if "." in k else k + lines.append(f'{key} = "{v}"') + + # [sim.sdf] + if self.sim.sdf is not None: + lines.append("") + lines.append("[sim.sdf]") + if self.sim.sdf.file is not None: + lines.append(f'file = "{self.sim.sdf.file}"') + if self.sim.sdf.corner != "typ": + lines.append(f'corner = "{self.sim.sdf.corner}"') + if self.sim.sdf.debug: + lines.append("debug = true") + + # [sim.timing] + if self.sim.timing is not None: + lines.append("") + lines.append("[sim.timing]") + if self.sim.timing.enabled: + lines.append("enabled = true") + if self.sim.timing.clock_period is not None: + lines.append(f"clock_period = {self.sim.timing.clock_period}") + if self.sim.timing.report_violations: + lines.append("report_violations = true") + + lines.append("") # trailing newline + Path(path).write_text("\n".join(lines)) + + def effective_gemparts(self) -> Path | None: + """Get effective gemparts path, falling back to map.output.""" + return self.sim.gemparts or self.map.output + + def _resolve_paths(self, config_dir: Path) -> None: + """Resolve relative paths against the config file's directory.""" + self.design.netlist = _resolve(self.design.netlist, config_dir) + self.design.liberty = _resolve(self.design.liberty, config_dir) + self.map.output = _resolve(self.map.output, config_dir) + self.sim.gemparts = _resolve(self.sim.gemparts, config_dir) + self.sim.input_vcd = _resolve(self.sim.input_vcd, config_dir) + self.sim.output_vcd = _resolve(self.sim.output_vcd, config_dir) + if self.sim.sdf is not None: + self.sim.sdf.file = _resolve(self.sim.sdf.file, config_dir) + + +def _resolve(path: Path | None, base: Path) -> Path | None: + """Resolve a relative path against a base directory. Absolute paths unchanged.""" + if path is None: + return None + if path.is_absolute(): + return path + return base / path diff --git a/python/jacquard/src/jacquard/errors.py b/python/jacquard/src/jacquard/errors.py new file mode 100644 index 0000000..8ffde69 --- /dev/null +++ b/python/jacquard/src/jacquard/errors.py @@ -0,0 +1,31 @@ +"""Jacquard error hierarchy for Python API.""" + + +class JacquardError(Exception): + """Base error for all Jacquard operations.""" + + +class ConfigError(JacquardError): + """Error in configuration parsing or validation.""" + + +class MappingError(JacquardError): + """Error during partition mapping (jacquard map).""" + + def __init__(self, message: str, returncode: int = 1, stderr: str = ""): + super().__init__(message) + self.returncode = returncode + self.stderr = stderr + + +class SimulationError(JacquardError): + """Error during GPU simulation (jacquard sim).""" + + def __init__(self, message: str, returncode: int = 1, stderr: str = ""): + super().__init__(message) + self.returncode = returncode + self.stderr = stderr + + +class BinaryNotFoundError(JacquardError): + """Jacquard binary not found in PATH or cargo target directory.""" diff --git a/python/jacquard/src/jacquard/regression.py b/python/jacquard/src/jacquard/regression.py new file mode 100644 index 0000000..55db4d4 --- /dev/null +++ b/python/jacquard/src/jacquard/regression.py @@ -0,0 +1,197 @@ +"""Batch regression runner for multi-design testing.""" + +from __future__ import annotations + +import logging +import time +from concurrent.futures import ProcessPoolExecutor, as_completed +from dataclasses import dataclass, field +from pathlib import Path + +from .config import JacquardConfig +from .result import SimResult +from .runner import sim + +log = logging.getLogger(__name__) + + +@dataclass +class TestCase: + """A single regression test case.""" + + name: str + config: JacquardConfig + expected_exit: int = 0 + max_cycles: int | None = None + golden_vcd: Path | None = None + + +@dataclass +class TestResult: + """Result of a single test case within a regression run.""" + + name: str + sim_result: SimResult + passed: bool + error_message: str = "" + + +@dataclass +class RegressionReport: + """Aggregated results of a regression run.""" + + results: dict[str, TestResult] = field(default_factory=dict) + total_wall_clock: float = 0.0 + + @property + def passed(self) -> int: + return sum(1 for r in self.results.values() if r.passed) + + @property + def failed(self) -> int: + return sum( + 1 for r in self.results.values() + if not r.passed and r.sim_result.returncode != -1 + ) + + @property + def errors(self) -> int: + return sum(1 for r in self.results.values() if r.sim_result.returncode == -1) + + @property + def total(self) -> int: + return len(self.results) + + def summary(self) -> str: + """Human-readable summary of regression results.""" + lines = [ + f"Regression: {self.passed}/{self.total} passed" + f" ({self.failed} failed, {self.errors} errors)", + f"Total wall-clock: {self.total_wall_clock:.1f}s", + "", + ] + for name, result in sorted(self.results.items()): + status = "PASS" if result.passed else "FAIL" + detail = "" + if result.sim_result.num_cycles > 0: + cycles = result.sim_result.num_cycles + elapsed = result.sim_result.elapsed + detail = f" ({cycles} cycles, {elapsed:.1f}s)" + if result.error_message: + detail += f" [{result.error_message}]" + lines.append(f" {status}: {name}{detail}") + + return "\n".join(lines) + + +def _run_single_test( + test: TestCase, + jacquard_bin: Path | None, + cargo_target_dir: Path | None, +) -> TestResult: + """Execute a single test case and evaluate pass/fail.""" + overrides: dict = {} + if test.max_cycles is not None: + overrides["max_cycles"] = test.max_cycles + + try: + result = sim( + test.config, + jacquard_bin=jacquard_bin, + cargo_target_dir=cargo_target_dir, + **overrides, + ) + except Exception as e: + return TestResult( + name=test.name, + sim_result=SimResult( + success=False, + returncode=-1, + stdout="", + stderr=str(e), + elapsed=0.0, + ), + passed=False, + error_message=str(e), + ) + + passed = result.returncode == test.expected_exit + error_msg = "" + if not passed: + error_msg = f"Expected exit {test.expected_exit}, got {result.returncode}" + if result.stderr: + # Grab last line of stderr for context + last_line = result.stderr.strip().splitlines()[-1] if result.stderr.strip() else "" + if last_line: + error_msg += f": {last_line[:200]}" + + return TestResult( + name=test.name, + sim_result=result, + passed=passed, + error_message=error_msg, + ) + + +def run_regression( + tests: list[TestCase], + *, + jacquard_bin: Path | None = None, + cargo_target_dir: Path | None = None, + parallel: int = 1, +) -> RegressionReport: + """Run a batch of test cases, optionally in parallel. + + Args: + tests: List of test cases to execute. + jacquard_bin: Explicit path to jacquard binary. + cargo_target_dir: Cargo target directory for binary lookup. + parallel: Maximum number of concurrent tests. Each sim gets its own + GPU context, so parallelism is limited by GPU memory. + + Returns: + RegressionReport with pass/fail results for each test. + """ + report = RegressionReport() + start = time.monotonic() + + if parallel <= 1: + # Sequential execution + for test in tests: + log.info("Running test: %s", test.name) + result = _run_single_test(test, jacquard_bin, cargo_target_dir) + report.results[test.name] = result + status = "PASS" if result.passed else "FAIL" + log.info(" %s: %s (%.1fs)", test.name, status, result.sim_result.elapsed) + else: + # Parallel execution + with ProcessPoolExecutor(max_workers=parallel) as executor: + futures = { + executor.submit( + _run_single_test, test, jacquard_bin, cargo_target_dir + ): test.name + for test in tests + } + for future in as_completed(futures): + name = futures[future] + try: + result = future.result() + except Exception as e: + result = TestResult( + name=name, + sim_result=SimResult( + success=False, + returncode=-1, + stdout="", + stderr=str(e), + elapsed=0.0, + ), + passed=False, + error_message=f"Executor error: {e}", + ) + report.results[name] = result + status = "PASS" if result.passed else "FAIL" + log.info(" %s: %s (%.1fs)", name, status, result.sim_result.elapsed) + + report.total_wall_clock = time.monotonic() - start + return report diff --git a/python/jacquard/src/jacquard/result.py b/python/jacquard/src/jacquard/result.py new file mode 100644 index 0000000..0112ef2 --- /dev/null +++ b/python/jacquard/src/jacquard/result.py @@ -0,0 +1,178 @@ +"""Structured result types for Jacquard CLI output parsing.""" + +from __future__ import annotations + +import re +from dataclasses import dataclass +from pathlib import Path + + +@dataclass +class DesignStats: + """Design-level statistics parsed from `=== Design Statistics ===` block.""" + + num_aigpins: int = 0 + num_and_gates: int = 0 + num_dffs: int = 0 + num_srams: int = 0 + num_inputs: int = 0 + num_outputs: int = 0 + total_partitions: int = 0 + num_major_stages: int = 0 + num_blocks: int = 0 + reg_io_state_words: int = 0 + sram_storage_words: int = 0 + script_data_words: int = 0 + min_partitions_per_block: int = 0 + max_partitions_per_block: int = 0 + min_script_size: int = 0 + max_script_size: int = 0 + load_balance_ratio: float = 1.0 + xprop_enabled: bool = False + num_x_capable_partitions: int = 0 + + +@dataclass +class MapResult: + """Result of a `jacquard map` invocation.""" + + success: bool + returncode: int + stdout: str + stderr: str + elapsed: float # wall-clock seconds + gemparts_path: Path | None = None + + +@dataclass +class SimResult: + """Result of a `jacquard sim` invocation.""" + + success: bool + returncode: int + stdout: str + stderr: str + elapsed: float # wall-clock seconds (total process time) + design_stats: DesignStats | None = None + gpu_backend: str = "" + num_cycles: int = 0 + wall_clock_secs: float = 0.0 # GPU sim time reported by Jacquard + throughput: float = 0.0 # cycles/sec + state_memory_kb: float = 0.0 + output_signals: int = 0 + output_vcd: Path | None = None + + +# --- Parsing functions --- + +# Regex patterns matching the Rust Display impl in src/sim/report.rs + +_RE_AIG = re.compile( + r"AIG:\s+(\d+)\s+pins,\s+(\d+)\s+AND gates,\s+(\d+)\s+DFFs,\s+(\d+)\s+SRAMs" +) +_RE_IO = re.compile(r"I/O:\s+(\d+)\s+inputs,\s+(\d+)\s+outputs") +_RE_STATE = re.compile(r"State:\s+(\d+)\s+words\s+.*?\+\s+(\d+)\s+words") +_RE_SCRIPT = re.compile(r"Script:\s+(\d+)\s+words") +_RE_PARTITIONS = re.compile( + r"Partitions:\s+(\d+)\s+total\s+across\s+(\d+)\s+stage\(s\)\s+.*?(\d+)\s+block\(s\)" +) +_RE_PER_BLOCK = re.compile( + r"Per block:\s+(\d+)-(\d+)\s+partitions,\s+script\s+(\d+)-(\d+)\s+words\s+" + r"\(balance ratio:\s+([\d.]+)x\)" +) +_RE_XPROP = re.compile(r"X-propagation:\s+(\d+)/(\d+)\s+partitions\s+X-capable") + +_RE_GPU_BACKEND = re.compile(r"GPU backend:\s+(\S+)") +_RE_CYCLES = re.compile(r"Cycles simulated:\s+(\d+)") +_RE_WALL_CLOCK = re.compile(r"Wall-clock time:\s+([\d.]+)s") +_RE_THROUGHPUT = re.compile(r"Throughput:\s+([\d.eE+]+)\s+cycles/sec") +_RE_STATE_MEM = re.compile(r"State memory:\s+([\d.]+)\s+KB/cycle") +_RE_OUTPUT = re.compile(r"Output:\s+(\d+)\s+signals\s+.*?→\s+(.+)") + + +def parse_design_stats(text: str) -> DesignStats | None: + """Parse a `=== Design Statistics ===` block from CLI output. + + Returns None if the block is not found. + """ + if "=== Design Statistics ===" not in text: + return None + + stats = DesignStats() + + m = _RE_AIG.search(text) + if m: + stats.num_aigpins = int(m.group(1)) + stats.num_and_gates = int(m.group(2)) + stats.num_dffs = int(m.group(3)) + stats.num_srams = int(m.group(4)) + + m = _RE_IO.search(text) + if m: + stats.num_inputs = int(m.group(1)) + stats.num_outputs = int(m.group(2)) + + m = _RE_STATE.search(text) + if m: + stats.reg_io_state_words = int(m.group(1)) + stats.sram_storage_words = int(m.group(2)) + + m = _RE_SCRIPT.search(text) + if m: + stats.script_data_words = int(m.group(1)) + + m = _RE_PARTITIONS.search(text) + if m: + stats.total_partitions = int(m.group(1)) + stats.num_major_stages = int(m.group(2)) + stats.num_blocks = int(m.group(3)) + + m = _RE_PER_BLOCK.search(text) + if m: + stats.min_partitions_per_block = int(m.group(1)) + stats.max_partitions_per_block = int(m.group(2)) + stats.min_script_size = int(m.group(3)) + stats.max_script_size = int(m.group(4)) + stats.load_balance_ratio = float(m.group(5)) + + m = _RE_XPROP.search(text) + if m: + stats.xprop_enabled = True + stats.num_x_capable_partitions = int(m.group(1)) + + return stats + + +def parse_sim_summary(text: str) -> dict: + """Parse a `=== Simulation Summary ===` block from CLI output. + + Returns a dict with parsed fields. Missing fields have default values. + """ + result: dict = {} + + m = _RE_GPU_BACKEND.search(text) + if m: + result["gpu_backend"] = m.group(1) + + m = _RE_CYCLES.search(text) + if m: + result["num_cycles"] = int(m.group(1)) + + m = _RE_WALL_CLOCK.search(text) + if m: + result["wall_clock_secs"] = float(m.group(1)) + + m = _RE_THROUGHPUT.search(text) + if m: + result["throughput"] = float(m.group(1)) + + m = _RE_STATE_MEM.search(text) + if m: + result["state_memory_kb"] = float(m.group(1)) + + m = _RE_OUTPUT.search(text) + if m: + result["output_signals"] = int(m.group(1)) + result["output_vcd"] = Path(m.group(2).strip()) + + return result diff --git a/python/jacquard/src/jacquard/runner.py b/python/jacquard/src/jacquard/runner.py new file mode 100644 index 0000000..daf5b65 --- /dev/null +++ b/python/jacquard/src/jacquard/runner.py @@ -0,0 +1,341 @@ +"""Subprocess wrappers for `jacquard map` and `jacquard sim`.""" + +from __future__ import annotations + +import logging +import shutil +import subprocess +import time +from dataclasses import replace +from pathlib import Path + +from .config import JacquardConfig +from .errors import BinaryNotFoundError +from .result import MapResult, SimResult, parse_design_stats, parse_sim_summary + +log = logging.getLogger(__name__) + + +def find_jacquard_binary( + explicit_path: Path | None = None, + cargo_target_dir: Path | None = None, +) -> Path: + """Locate the jacquard binary. + + Search order: + 1. Explicit path (if provided) + 2. cargo target directory (release then debug) + 3. PATH lookup via shutil.which + """ + if explicit_path is not None: + p = Path(explicit_path) + if p.is_file(): + return p + raise BinaryNotFoundError(f"Explicit jacquard binary not found: {p}") + + if cargo_target_dir is not None: + for profile in ("release", "debug"): + candidate = Path(cargo_target_dir) / profile / "jacquard" + if candidate.is_file(): + return candidate + + found = shutil.which("jacquard") + if found is not None: + return Path(found) + + raise BinaryNotFoundError( + "jacquard binary not found. Install it or pass jacquard_bin explicitly." + ) + + +def _build_map_args(config: JacquardConfig) -> list[str]: + """Build CLI arguments for `jacquard map`.""" + args: list[str] = ["map"] + + assert config.design.netlist is not None, "design.netlist is required for map" + args.append(str(config.design.netlist)) + + output = config.map.output + assert output is not None, "map.output is required for map" + args.append(str(output)) + + if config.design.top_module: + args.extend(["--top-module", config.design.top_module]) + + if config.map.level_split: + args.extend(["--level-split", ",".join(str(x) for x in config.map.level_split)]) + + if config.map.max_stage_degrad != 0: + args.extend(["--max-stage-degrad", str(config.map.max_stage_degrad)]) + + if config.map.xprop: + args.append("--xprop") + + return args + + +def _build_sim_args(config: JacquardConfig) -> list[str]: + """Build CLI arguments for `jacquard sim`.""" + args: list[str] = ["sim"] + + assert config.design.netlist is not None, "design.netlist is required for sim" + args.append(str(config.design.netlist)) + + gemparts = config.effective_gemparts() + assert gemparts is not None, "gemparts (or map.output) is required for sim" + args.append(str(gemparts)) + + assert config.sim.input_vcd is not None, "sim.input_vcd is required for sim" + args.append(str(config.sim.input_vcd)) + + assert config.sim.output_vcd is not None, "sim.output_vcd is required for sim" + args.append(str(config.sim.output_vcd)) + + if config.sim.num_blocks is not None: + args.extend(["--num-blocks", str(config.sim.num_blocks)]) + + if config.design.top_module: + args.extend(["--top-module", config.design.top_module]) + + if config.map.level_split: + args.extend(["--level-split", ",".join(str(x) for x in config.map.level_split)]) + + if config.sim.input_vcd_scope: + args.extend(["--input-vcd-scope", config.sim.input_vcd_scope]) + + if config.sim.output_vcd_scope: + args.extend(["--output-vcd-scope", config.sim.output_vcd_scope]) + + if config.sim.check_with_cpu: + args.append("--check-with-cpu") + + if config.sim.max_cycles is not None: + args.extend(["--max-cycles", str(config.sim.max_cycles)]) + + if config.sim.xprop: + args.append("--xprop") + + for pattern in config.sim.dump_signals: + args.extend(["--dump-signals", pattern]) + + if config.sim.dump_scope: + args.extend(["--dump-scope", config.sim.dump_scope]) + + if config.sim.dump_depth is not None: + args.extend(["--dump-depth", str(config.sim.dump_depth)]) + + for k, v in config.sim.plusargs.items(): + args.extend(["--plusarg", f"{k}={v}"]) + + if config.sim.sdf is not None and config.sim.sdf.file is not None: + args.extend(["--sdf", str(config.sim.sdf.file)]) + if config.sim.sdf.corner != "typ": + args.extend(["--sdf-corner", config.sim.sdf.corner]) + if config.sim.sdf.debug: + args.append("--sdf-debug") + + if config.sim.timing is not None and config.sim.timing.enabled: + args.append("--enable-timing") + if config.sim.timing.clock_period is not None: + args.extend(["--timing-clock-period", str(config.sim.timing.clock_period)]) + if config.sim.timing.report_violations: + args.append("--timing-report-violations") + + if config.design.liberty: + args.extend(["--liberty", str(config.design.liberty)]) + + return args + + +def _apply_overrides(config: JacquardConfig, **overrides: object) -> JacquardConfig: + """Apply keyword overrides to a config, returning a new copy. + + Supports flat keys that map to nested fields: + - netlist, top_module, liberty -> design.* + - output, level_split, max_stage_degrad -> map.* + - gemparts, input_vcd, output_vcd, num_blocks, max_cycles, xprop, + dump_signals, dump_scope, dump_depth, plusargs, check_with_cpu -> sim.* + """ + if not overrides: + return config + + # Deep copy via replace + design = replace(config.design) + map_cfg = replace(config.map) + sim_cfg = replace(config.sim) + + design_keys = {"netlist", "top_module", "liberty"} + map_keys = {"output", "level_split", "max_stage_degrad"} + sim_keys = { + "gemparts", "input_vcd", "output_vcd", "num_blocks", "max_cycles", + "xprop", "dump_signals", "dump_scope", "dump_depth", "plusargs", + "check_with_cpu", "input_vcd_scope", "output_vcd_scope", + } + + for key, val in overrides.items(): + if key in design_keys: + if key == "netlist" and val is not None: + val = Path(val) # type: ignore[arg-type] + elif key == "liberty" and val is not None: + val = Path(val) # type: ignore[arg-type] + setattr(design, key, val) + elif key in map_keys: + if key == "output" and val is not None: + val = Path(val) # type: ignore[arg-type] + setattr(map_cfg, key, val) + elif key in sim_keys: + if key in ("gemparts", "input_vcd", "output_vcd") and val is not None: + val = Path(val) # type: ignore[arg-type] + setattr(sim_cfg, key, val) + else: + log.warning("Unknown override key: %s", key) + + return JacquardConfig(design=design, map=map_cfg, sim=sim_cfg) + + +def map( + config: JacquardConfig, + *, + jacquard_bin: Path | None = None, + cargo_target_dir: Path | None = None, + verbose: int = 0, + quiet: int = 0, + timeout: float | None = None, + **overrides: object, +) -> MapResult: + """Run `jacquard map` as a subprocess. + + Args: + config: Project configuration. + jacquard_bin: Explicit path to jacquard binary. + cargo_target_dir: Cargo target directory for binary lookup. + verbose: Verbosity level (number of -v flags). + quiet: Quiet level (number of -q flags). + timeout: Subprocess timeout in seconds. + **overrides: Override any config field for this run. + + Returns: + MapResult with success status and parsed output. + + Raises: + BinaryNotFoundError: If jacquard binary cannot be found. + MappingError: If map fails and caller wants exception on failure. + """ + binary = find_jacquard_binary(jacquard_bin, cargo_target_dir) + effective = _apply_overrides(config, **overrides) + + cmd = [str(binary)] + cmd.extend(["-v"] * verbose) + cmd.extend(["-q"] * quiet) + cmd.extend(_build_map_args(effective)) + + log.info("Running: %s", " ".join(cmd)) + + start = time.monotonic() + try: + proc = subprocess.run( + cmd, + capture_output=True, + text=True, + timeout=timeout, + ) + except subprocess.TimeoutExpired as e: + elapsed = time.monotonic() - start + return MapResult( + success=False, + returncode=-1, + stdout=e.stdout or "", + stderr=f"Timeout after {elapsed:.1f}s", + elapsed=elapsed, + ) + + elapsed = time.monotonic() - start + + return MapResult( + success=proc.returncode == 0, + returncode=proc.returncode, + stdout=proc.stdout, + stderr=proc.stderr, + elapsed=elapsed, + gemparts_path=effective.map.output if proc.returncode == 0 else None, + ) + + +def sim( + config: JacquardConfig, + *, + jacquard_bin: Path | None = None, + cargo_target_dir: Path | None = None, + verbose: int = 0, + quiet: int = 0, + timeout: float | None = None, + **overrides: object, +) -> SimResult: + """Run `jacquard sim` as a subprocess. + + Args: + config: Project configuration. + jacquard_bin: Explicit path to jacquard binary. + cargo_target_dir: Cargo target directory for binary lookup. + verbose: Verbosity level (number of -v flags). + quiet: Quiet level (number of -q flags). + timeout: Subprocess timeout in seconds. + **overrides: Override any config field for this run. + + Returns: + SimResult with success status and parsed output. + + Raises: + BinaryNotFoundError: If jacquard binary cannot be found. + """ + binary = find_jacquard_binary(jacquard_bin, cargo_target_dir) + effective = _apply_overrides(config, **overrides) + + cmd = [str(binary)] + cmd.extend(["-v"] * verbose) + cmd.extend(["-q"] * quiet) + cmd.extend(_build_sim_args(effective)) + + log.info("Running: %s", " ".join(cmd)) + + start = time.monotonic() + try: + proc = subprocess.run( + cmd, + capture_output=True, + text=True, + timeout=timeout, + ) + except subprocess.TimeoutExpired as e: + elapsed = time.monotonic() - start + return SimResult( + success=False, + returncode=-1, + stdout=e.stdout or "", + stderr=f"Timeout after {elapsed:.1f}s", + elapsed=elapsed, + ) + + elapsed = time.monotonic() - start + + # Combine stdout+stderr for parsing (Rust may log to either) + combined = proc.stdout + "\n" + proc.stderr + + design_stats = parse_design_stats(combined) + summary = parse_sim_summary(combined) + + return SimResult( + success=proc.returncode == 0, + returncode=proc.returncode, + stdout=proc.stdout, + stderr=proc.stderr, + elapsed=elapsed, + design_stats=design_stats, + gpu_backend=summary.get("gpu_backend", ""), + num_cycles=summary.get("num_cycles", 0), + wall_clock_secs=summary.get("wall_clock_secs", 0.0), + throughput=summary.get("throughput", 0.0), + state_memory_kb=summary.get("state_memory_kb", 0.0), + output_signals=summary.get("output_signals", 0), + output_vcd=summary.get("output_vcd", effective.sim.output_vcd), + ) diff --git a/python/jacquard/tests/test_config.py b/python/jacquard/tests/test_config.py new file mode 100644 index 0000000..e944218 --- /dev/null +++ b/python/jacquard/tests/test_config.py @@ -0,0 +1,215 @@ +"""Tests for JacquardConfig TOML I/O and field handling.""" + +from pathlib import Path + +import pytest + +from jacquard.config import JacquardConfig, SdfConfig, TimingConfig +from jacquard.errors import ConfigError + + +@pytest.fixture +def tmp_toml(tmp_path): + """Helper to write TOML content and return path.""" + def _write(content: str) -> Path: + p = tmp_path / "jacquard.toml" + p.write_text(content) + return p + return _write + + +class TestFromToml: + def test_empty_config(self, tmp_toml): + config = JacquardConfig.from_toml(tmp_toml("")) + assert config.design.netlist is None + assert config.map.level_split == [] + assert config.sim.num_blocks is None + + def test_minimal_config(self, tmp_toml): + config = JacquardConfig.from_toml(tmp_toml(""" +[design] +netlist = "gatelevel.gv" + +[map] +output = "result.gemparts" + +[sim] +input_vcd = "input.vcd" +output_vcd = "output.vcd" +""")) + # Paths are resolved against tmp_path + assert config.design.netlist is not None + assert config.design.netlist.name == "gatelevel.gv" + assert config.map.output is not None + assert config.map.output.name == "result.gemparts" + + def test_full_config(self, tmp_toml): + config = JacquardConfig.from_toml(tmp_toml(""" +[design] +netlist = "build/gatelevel.gv" +top_module = "my_top" +liberty = "/opt/pdk/lib.lib" + +[map] +output = "build/result.gemparts" +level_split = [20, 40] +max_stage_degrad = 1 +xprop = true + +[sim] +input_vcd = "test/input.vcd" +output_vcd = "test/output.vcd" +num_blocks = 128 +max_cycles = 1000 +check_with_cpu = true +xprop = true +dump_signals = ["*clk*", "data_out*"] +dump_scope = "cpu/core0" +dump_depth = 2 + +[sim.plusargs] +run_id = "nightly_042" +seed = "12345" +"force.reset" = "0" + +[sim.sdf] +file = "build/design.sdf" +corner = "typ" +debug = false + +[sim.timing] +enabled = true +clock_period = 1000 +report_violations = true +""")) + assert config.map.level_split == [20, 40] + assert config.sim.num_blocks == 128 + assert config.sim.dump_signals == ["*clk*", "data_out*"] + assert config.sim.dump_scope == "cpu/core0" + assert config.sim.dump_depth == 2 + assert config.sim.plusargs["run_id"] == "nightly_042" + assert config.sim.plusargs["force.reset"] == "0" + assert config.sim.sdf is not None + assert config.sim.sdf.corner == "typ" + assert config.sim.timing is not None + assert config.sim.timing.clock_period == 1000 + + def test_path_resolution_relative(self, tmp_toml): + config = JacquardConfig.from_toml(tmp_toml(""" +[design] +netlist = "build/gatelevel.gv" + +[map] +output = "build/result.gemparts" +""")) + # Relative paths resolved against config dir (tmp_path) + assert config.design.netlist is not None + assert config.design.netlist.is_absolute() + assert "build" in str(config.design.netlist) + + def test_path_resolution_absolute(self, tmp_toml): + config = JacquardConfig.from_toml(tmp_toml(""" +[design] +netlist = "/absolute/path/gatelevel.gv" +liberty = "/opt/pdk/lib.lib" +""")) + assert config.design.netlist == Path("/absolute/path/gatelevel.gv") + assert config.design.liberty == Path("/opt/pdk/lib.lib") + + def test_missing_file(self, tmp_path): + with pytest.raises(ConfigError, match="not found"): + JacquardConfig.from_toml(tmp_path / "nonexistent.toml") + + def test_invalid_toml(self, tmp_toml): + with pytest.raises(ConfigError, match="Failed to parse"): + JacquardConfig.from_toml(tmp_toml("this is not [valid toml")) + + def test_effective_gemparts_direct(self, tmp_toml): + config = JacquardConfig.from_toml(tmp_toml(""" +[sim] +gemparts = "sim.gemparts" +""")) + assert config.effective_gemparts() is not None + assert config.effective_gemparts().name == "sim.gemparts" # type: ignore[union-attr] + + def test_effective_gemparts_fallback(self, tmp_toml): + config = JacquardConfig.from_toml(tmp_toml(""" +[map] +output = "map.gemparts" +""")) + assert config.effective_gemparts() is not None + assert config.effective_gemparts().name == "map.gemparts" # type: ignore[union-attr] + + def test_plusargs_default_empty(self, tmp_toml): + config = JacquardConfig.from_toml(tmp_toml("")) + assert config.sim.plusargs == {} + + def test_dump_signals_default_empty(self, tmp_toml): + config = JacquardConfig.from_toml(tmp_toml("")) + assert config.sim.dump_signals == [] + assert config.sim.dump_scope is None + assert config.sim.dump_depth is None + + +class TestRoundTrip: + def test_minimal_round_trip(self, tmp_path): + original = JacquardConfig() + original.design.netlist = Path("gatelevel.gv") + original.map.output = Path("result.gemparts") + original.sim.input_vcd = Path("input.vcd") + original.sim.output_vcd = Path("output.vcd") + + toml_path = tmp_path / "jacquard.toml" + original.to_toml(toml_path) + + loaded = JacquardConfig.from_toml(toml_path) + assert loaded.design.netlist is not None + assert loaded.design.netlist.name == "gatelevel.gv" + assert loaded.map.output is not None + assert loaded.map.output.name == "result.gemparts" + + def test_full_round_trip(self, tmp_path): + original = JacquardConfig() + original.design.netlist = Path("build/design.gv") + original.design.top_module = "top" + original.map.output = Path("build/result.gemparts") + original.map.level_split = [20, 40] + original.map.max_stage_degrad = 1 + original.map.xprop = True + original.sim.input_vcd = Path("test/input.vcd") + original.sim.output_vcd = Path("test/output.vcd") + original.sim.num_blocks = 64 + original.sim.max_cycles = 5000 + original.sim.xprop = True + original.sim.check_with_cpu = True + original.sim.dump_signals = ["*clk*", "data*"] + original.sim.dump_scope = "cpu" + original.sim.dump_depth = 3 + original.sim.plusargs = {"run_id": "test1", "force.reset": "0"} + original.sim.sdf = SdfConfig(file=Path("design.sdf"), corner="max", debug=True) + original.sim.timing = TimingConfig(enabled=True, clock_period=2000, report_violations=True) + + toml_path = tmp_path / "jacquard.toml" + original.to_toml(toml_path) + + loaded = JacquardConfig.from_toml(toml_path) + + assert loaded.design.top_module == "top" + assert loaded.map.level_split == [20, 40] + assert loaded.map.max_stage_degrad == 1 + assert loaded.map.xprop is True + assert loaded.sim.num_blocks == 64 + assert loaded.sim.max_cycles == 5000 + assert loaded.sim.xprop is True + assert loaded.sim.check_with_cpu is True + assert loaded.sim.dump_signals == ["*clk*", "data*"] + assert loaded.sim.dump_scope == "cpu" + assert loaded.sim.dump_depth == 3 + assert loaded.sim.plusargs == {"run_id": "test1", "force.reset": "0"} + assert loaded.sim.sdf is not None + assert loaded.sim.sdf.corner == "max" + assert loaded.sim.sdf.debug is True + assert loaded.sim.timing is not None + assert loaded.sim.timing.enabled is True + assert loaded.sim.timing.clock_period == 2000 + assert loaded.sim.timing.report_violations is True diff --git a/python/jacquard/tests/test_regression.py b/python/jacquard/tests/test_regression.py new file mode 100644 index 0000000..63a039b --- /dev/null +++ b/python/jacquard/tests/test_regression.py @@ -0,0 +1,158 @@ +"""Tests for batch regression runner.""" + +from pathlib import Path +from unittest.mock import patch + +from jacquard.config import DesignConfig, JacquardConfig, MapConfig, SimConfig +from jacquard.regression import ( + RegressionReport, + TestCase, + TestResult, + _run_single_test, + run_regression, +) +from jacquard.result import SimResult + + +def _make_config() -> JacquardConfig: + """Build a minimal config for testing.""" + return JacquardConfig( + design=DesignConfig(netlist=Path("design.gv")), + map=MapConfig(output=Path("result.gemparts")), + sim=SimConfig( + input_vcd=Path("input.vcd"), + output_vcd=Path("output.vcd"), + ), + ) + + +def _mock_sim_result(returncode: int = 0, cycles: int = 100) -> SimResult: + return SimResult( + success=returncode == 0, + returncode=returncode, + stdout="", + stderr="" if returncode == 0 else "Error: something failed", + elapsed=1.5, + num_cycles=cycles, + ) + + +class TestRegressionReport: + def test_empty_report(self): + report = RegressionReport() + assert report.passed == 0 + assert report.failed == 0 + assert report.errors == 0 + assert report.total == 0 + + def test_all_passed(self): + report = RegressionReport(results={ + "test1": TestResult("test1", _mock_sim_result(), passed=True), + "test2": TestResult("test2", _mock_sim_result(), passed=True), + }) + assert report.passed == 2 + assert report.failed == 0 + assert report.total == 2 + + def test_mixed_results(self): + report = RegressionReport(results={ + "pass": TestResult("pass", _mock_sim_result(), passed=True), + "fail": TestResult("fail", _mock_sim_result(returncode=1), passed=False), + "error": TestResult("error", _mock_sim_result(returncode=-1), passed=False), + }) + assert report.passed == 1 + assert report.failed == 1 + assert report.errors == 1 + + def test_summary_output(self): + report = RegressionReport( + results={ + "nvdla": TestResult("nvdla", _mock_sim_result(cycles=10000), passed=True), + "rocket": TestResult( + "rocket", + _mock_sim_result(returncode=1), + passed=False, + error_message="Expected exit 0, got 1", + ), + }, + total_wall_clock=5.0, + ) + summary = report.summary() + assert "1/2 passed" in summary + assert "PASS: nvdla" in summary + assert "FAIL: rocket" in summary + assert "5.0s" in summary + + +class TestRunSingleTest: + @patch("jacquard.regression.sim") + def test_passing_test(self, mock_sim): + mock_sim.return_value = _mock_sim_result() + test = TestCase(name="basic", config=_make_config()) + result = _run_single_test(test, None, None) + assert result.passed is True + assert result.error_message == "" + + @patch("jacquard.regression.sim") + def test_failing_test(self, mock_sim): + mock_sim.return_value = _mock_sim_result(returncode=1) + test = TestCase(name="failing", config=_make_config()) + result = _run_single_test(test, None, None) + assert result.passed is False + assert "Expected exit 0, got 1" in result.error_message + + @patch("jacquard.regression.sim") + def test_expected_nonzero_exit(self, mock_sim): + mock_sim.return_value = _mock_sim_result(returncode=1) + test = TestCase(name="expected_fail", config=_make_config(), expected_exit=1) + result = _run_single_test(test, None, None) + assert result.passed is True + + @patch("jacquard.regression.sim") + def test_max_cycles_override(self, mock_sim): + mock_sim.return_value = _mock_sim_result() + test = TestCase(name="limited", config=_make_config(), max_cycles=50) + _run_single_test(test, None, None) + # Verify max_cycles was passed as override + _, kwargs = mock_sim.call_args + assert kwargs.get("max_cycles") == 50 + + @patch("jacquard.regression.sim") + def test_exception_handling(self, mock_sim): + mock_sim.side_effect = RuntimeError("GPU exploded") + test = TestCase(name="crash", config=_make_config()) + result = _run_single_test(test, None, None) + assert result.passed is False + assert result.sim_result.returncode == -1 + assert "GPU exploded" in result.error_message + + +class TestRunRegression: + @patch("jacquard.regression.sim") + def test_sequential_execution(self, mock_sim): + mock_sim.return_value = _mock_sim_result() + tests = [ + TestCase(name="test1", config=_make_config()), + TestCase(name="test2", config=_make_config()), + ] + report = run_regression(tests, parallel=1) + assert report.total == 2 + assert report.passed == 2 + assert report.total_wall_clock > 0 + + @patch("jacquard.regression.sim") + def test_mixed_pass_fail(self, mock_sim): + def side_effect(config, **kwargs): + # Second call fails + if mock_sim.call_count == 2: + return _mock_sim_result(returncode=1) + return _mock_sim_result() + + mock_sim.side_effect = side_effect + tests = [ + TestCase(name="pass", config=_make_config()), + TestCase(name="fail", config=_make_config()), + ] + report = run_regression(tests, parallel=1) + assert report.passed == 1 + assert report.failed == 1 diff --git a/python/jacquard/tests/test_runner.py b/python/jacquard/tests/test_runner.py new file mode 100644 index 0000000..8d170c9 --- /dev/null +++ b/python/jacquard/tests/test_runner.py @@ -0,0 +1,297 @@ +"""Tests for CLI argument building and result parsing.""" + +from pathlib import Path +from unittest.mock import patch + +import pytest + +from jacquard.config import ( + DesignConfig, + JacquardConfig, + MapConfig, + SdfConfig, + SimConfig, + TimingConfig, +) +from jacquard.errors import BinaryNotFoundError +from jacquard.result import parse_design_stats, parse_sim_summary +from jacquard.runner import ( + _apply_overrides, + _build_map_args, + _build_sim_args, + find_jacquard_binary, +) + + +class TestFindBinary: + def test_explicit_path(self, tmp_path): + binary = tmp_path / "jacquard" + binary.touch() + result = find_jacquard_binary(explicit_path=binary) + assert result == binary + + def test_explicit_path_missing(self, tmp_path): + with pytest.raises(BinaryNotFoundError): + find_jacquard_binary(explicit_path=tmp_path / "nonexistent") + + def test_cargo_target_release(self, tmp_path): + release_dir = tmp_path / "release" + release_dir.mkdir() + binary = release_dir / "jacquard" + binary.touch() + result = find_jacquard_binary(cargo_target_dir=tmp_path) + assert result == binary + + def test_cargo_target_debug_fallback(self, tmp_path): + debug_dir = tmp_path / "debug" + debug_dir.mkdir() + binary = debug_dir / "jacquard" + binary.touch() + result = find_jacquard_binary(cargo_target_dir=tmp_path) + assert result == binary + + def test_path_lookup(self): + with patch("jacquard.runner.shutil.which", return_value="/usr/local/bin/jacquard"): + result = find_jacquard_binary() + assert result == Path("/usr/local/bin/jacquard") + + def test_not_found(self): + with patch("jacquard.runner.shutil.which", return_value=None): + with pytest.raises(BinaryNotFoundError): + find_jacquard_binary() + + +class TestBuildMapArgs: + def test_minimal(self): + config = JacquardConfig( + design=DesignConfig(netlist=Path("design.gv")), + map=MapConfig(output=Path("result.gemparts")), + ) + args = _build_map_args(config) + assert args == ["map", "design.gv", "result.gemparts"] + + def test_with_options(self): + config = JacquardConfig( + design=DesignConfig(netlist=Path("design.gv"), top_module="top"), + map=MapConfig( + output=Path("result.gemparts"), + level_split=[20, 40], + max_stage_degrad=1, + xprop=True, + ), + ) + args = _build_map_args(config) + assert "map" in args + assert "--top-module" in args + assert args[args.index("--top-module") + 1] == "top" + assert "--level-split" in args + assert args[args.index("--level-split") + 1] == "20,40" + assert "--max-stage-degrad" in args + assert args[args.index("--max-stage-degrad") + 1] == "1" + assert "--xprop" in args + + def test_missing_netlist_asserts(self): + config = JacquardConfig(map=MapConfig(output=Path("out.gemparts"))) + with pytest.raises(AssertionError, match="netlist"): + _build_map_args(config) + + def test_missing_output_asserts(self): + config = JacquardConfig(design=DesignConfig(netlist=Path("design.gv"))) + with pytest.raises(AssertionError, match="output"): + _build_map_args(config) + + +class TestBuildSimArgs: + def test_minimal(self): + config = JacquardConfig( + design=DesignConfig(netlist=Path("design.gv")), + map=MapConfig(output=Path("result.gemparts")), + sim=SimConfig( + input_vcd=Path("input.vcd"), + output_vcd=Path("output.vcd"), + ), + ) + args = _build_sim_args(config) + assert args[:5] == ["sim", "design.gv", "result.gemparts", "input.vcd", "output.vcd"] + + def test_with_all_options(self): + config = JacquardConfig( + design=DesignConfig( + netlist=Path("design.gv"), + top_module="top", + liberty=Path("timing.lib"), + ), + map=MapConfig(level_split=[30]), + sim=SimConfig( + gemparts=Path("parts.gemparts"), + input_vcd=Path("in.vcd"), + output_vcd=Path("out.vcd"), + num_blocks=128, + input_vcd_scope="tb.dut", + output_vcd_scope="dut", + check_with_cpu=True, + max_cycles=1000, + xprop=True, + dump_signals=["*clk*", "data*"], + dump_scope="cpu", + dump_depth=2, + plusargs={"seed": "42", "force.reset": "0"}, + sdf=SdfConfig(file=Path("design.sdf"), corner="max", debug=True), + timing=TimingConfig(enabled=True, clock_period=2000, report_violations=True), + ), + ) + args = _build_sim_args(config) + + assert "--num-blocks" in args + assert args[args.index("--num-blocks") + 1] == "128" + assert "--top-module" in args + assert "--level-split" in args + assert "--input-vcd-scope" in args + assert "--output-vcd-scope" in args + assert "--check-with-cpu" in args + assert "--max-cycles" in args + assert "--xprop" in args + assert "--dump-scope" in args + assert args[args.index("--dump-scope") + 1] == "cpu" + assert "--dump-depth" in args + assert args[args.index("--dump-depth") + 1] == "2" + assert "--sdf" in args + assert "--sdf-corner" in args + assert args[args.index("--sdf-corner") + 1] == "max" + assert "--sdf-debug" in args + assert "--enable-timing" in args + assert "--timing-clock-period" in args + assert "--timing-report-violations" in args + assert "--liberty" in args + + # Check dump_signals appear twice (two patterns) + dump_idxs = [i for i, a in enumerate(args) if a == "--dump-signals"] + assert len(dump_idxs) == 2 + + # Check plusargs appear + plusarg_idxs = [i for i, a in enumerate(args) if a == "--plusarg"] + assert len(plusarg_idxs) == 2 + plusarg_vals = {args[i + 1] for i in plusarg_idxs} + assert "seed=42" in plusarg_vals + assert "force.reset=0" in plusarg_vals + + +class TestApplyOverrides: + def test_no_overrides(self): + config = JacquardConfig() + result = _apply_overrides(config) + assert result is config # returns same object when no overrides + + def test_design_overrides(self): + config = JacquardConfig(design=DesignConfig(netlist=Path("old.gv"))) + result = _apply_overrides(config, netlist="new.gv", top_module="top") + assert result.design.netlist == Path("new.gv") + assert result.design.top_module == "top" + # Original unchanged + assert config.design.netlist == Path("old.gv") + + def test_sim_overrides(self): + config = JacquardConfig(sim=SimConfig(max_cycles=1000)) + result = _apply_overrides(config, max_cycles=500, num_blocks=64, xprop=True) + assert result.sim.max_cycles == 500 + assert result.sim.num_blocks == 64 + assert result.sim.xprop is True + + def test_map_overrides(self): + config = JacquardConfig() + result = _apply_overrides(config, level_split=[20, 40]) + assert result.map.level_split == [20, 40] + + +class TestParseDesignStats: + SAMPLE_OUTPUT = """\ +=== Design Statistics === +AIG: 5000 pins, 2000 AND gates, 500 DFFs, 4 SRAMs +I/O: 64 inputs, 128 outputs +State: 256 words (1.0 KB reg/io) + 1024 words (4.0 KB SRAM) +Script: 65536 words (256.0 KB) +Partitions: 120 total across 2 stage(s) → 32 block(s) + Per stage: [60, 60] + Per block: 3-5 partitions, script 1800-2200 words (balance ratio: 1.15x) +""" + + SAMPLE_XPROP_OUTPUT = SAMPLE_OUTPUT + \ + "X-propagation: 80/120 partitions X-capable\n" + + def test_parse_basic(self): + stats = parse_design_stats(self.SAMPLE_OUTPUT) + assert stats is not None + assert stats.num_aigpins == 5000 + assert stats.num_and_gates == 2000 + assert stats.num_dffs == 500 + assert stats.num_srams == 4 + assert stats.num_inputs == 64 + assert stats.num_outputs == 128 + assert stats.reg_io_state_words == 256 + assert stats.sram_storage_words == 1024 + assert stats.script_data_words == 65536 + assert stats.total_partitions == 120 + assert stats.num_major_stages == 2 + assert stats.num_blocks == 32 + assert stats.min_partitions_per_block == 3 + assert stats.max_partitions_per_block == 5 + assert stats.min_script_size == 1800 + assert stats.max_script_size == 2200 + assert abs(stats.load_balance_ratio - 1.15) < 0.01 + assert stats.xprop_enabled is False + + def test_parse_xprop(self): + stats = parse_design_stats(self.SAMPLE_XPROP_OUTPUT) + assert stats is not None + assert stats.xprop_enabled is True + assert stats.num_x_capable_partitions == 80 + + def test_parse_no_stats(self): + assert parse_design_stats("some random output") is None + + +class TestParseSimSummary: + SAMPLE_OUTPUT = """\ + +=== Simulation Summary === +GPU backend: Metal +Cycles simulated: 10000 +Wall-clock time: 1.234s +Throughput: 8.10e+03 cycles/sec +State memory: 2.5 KB/cycle +Output: 128 signals → /tmp/output.vcd +""" + + def test_parse_basic(self): + result = parse_sim_summary(self.SAMPLE_OUTPUT) + assert result["gpu_backend"] == "Metal" + assert result["num_cycles"] == 10000 + assert abs(result["wall_clock_secs"] - 1.234) < 0.001 + assert abs(result["throughput"] - 8100.0) < 1.0 + assert abs(result["state_memory_kb"] - 2.5) < 0.01 + assert result["output_signals"] == 128 + assert result["output_vcd"] == Path("/tmp/output.vcd") + + def test_parse_empty(self): + result = parse_sim_summary("no summary here") + assert result == {} + + def test_parse_with_plusargs(self): + output = self.SAMPLE_OUTPUT + "Plusargs: force.reset=0, seed=42\n" + result = parse_sim_summary(output) + # Plusargs line doesn't have a dedicated parser field; just check + # other fields still parse correctly + assert result["gpu_backend"] == "Metal" + + def test_scientific_notation_throughput(self): + output = """\ +=== Simulation Summary === +GPU backend: CUDA +Cycles simulated: 500000 +Wall-clock time: 0.500s +Throughput: 1.00e+06 cycles/sec +State memory: 10.0 KB/cycle +Output: 256 signals → out.vcd +""" + result = parse_sim_summary(output) + assert abs(result["throughput"] - 1_000_000.0) < 1.0