import json import os import re import shutil import subprocess import sys import time import numpy as np from dataclasses import dataclass, field from pathlib import Path from colorama import Style, Fore from .gen_network_runner import gen_network_runner from .onnx_utils import ( _ONNX_TO_NP, gen_random_inputs, generate_input_batch, onnx_io, save_inputs_to_files, write_input_batch_binaries, write_input_batch_csv, write_inputs_to_memory_bin, ) from .raptor import compile_with_raptor from .pimsim_nn import export_raptor_pimsim_artifact, parse_pimsim_nn_metrics, read_raptor_instruction_count from .subprocess_utils import run_command_with_reporter STAGE_TITLES = ( "Compile ONNX", "Build Runner", "Generate Inputs", "Run Reference", "Compile PIM", "Run Functional Simulation", "Compare Outputs", "Run Non-functional Simulation", ) STAGE_COLORS = { STAGE_TITLES[0]: Fore.BLUE, STAGE_TITLES[1]: Fore.MAGENTA, STAGE_TITLES[2]: Fore.YELLOW, STAGE_TITLES[3]: Fore.GREEN, STAGE_TITLES[4]: Fore.CYAN, STAGE_TITLES[5]: Fore.MAGENTA, STAGE_TITLES[6]: Fore.YELLOW, STAGE_TITLES[7]: Fore.BLUE, } STAGE_COUNT = len(STAGE_TITLES) GENERATED_DIR_NAMES = ( "inputs", "outputs", "pimcomp", "raptor", "runner", "simulation", "throughput_validation", ) MODE_FULL = "full" MODE_COMPILE_ONLY = "compile_only" MODE_RUN_ONLY = "run_only" MODE_STAGE_TITLES = { MODE_FULL: STAGE_TITLES, MODE_COMPILE_ONLY: ( "Compile ONNX", "Build Runner", "Compile PIM", ), MODE_RUN_ONLY: ( "Generate Inputs", "Run Reference", "Run Functional Simulation", "Compare Outputs", "Run Non-functional Simulation", ), } PIMSIM_DONE = "DONE" PIMSIM_FAILED = "ERROR" PIMSIM_UNSUPPORTED = "UNSUPPORTED" PIMSIM_SKIPPED = "SKIP" PIMSIM_NOT_RUN = "-" PIMSIM_UNSUPPORTED_VSOFTMAX = "pimsim-nn does not support opcode vsoftmax" class PimSimUnsupportedError(RuntimeError): pass def sanitize_output_name(name): return "".join(ch if ch.isalnum() or ch in "_.-" else "_" for ch in name[:255]) @dataclass class ValidationResult: passed: bool latency_passed: bool | None = None throughput_passed: bool | None = None pim_pass_timings: dict[str, float] = field(default_factory=dict) pimsim_latency_ms: float | None = None pimsim_throughput_samples_s: float | None = None pimsim_power_mw: float | None = None pimsim_energy_pj: float | None = None pimsim_throughput_average_latency_ms: float | None = None pimsim_throughput_average_power_mw: float | None = None pimsim_throughput_average_energy_pj: float | None = None mode_metrics: dict[str, dict[str, float | int | None]] = field(default_factory=dict) pimsim_status: str = PIMSIM_SKIPPED throughput_pimsim_status: str = PIMSIM_SKIPPED compile_time_s: float | None = None host_memory_bytes: int | None = None cores_memory_bytes: int | None = None used_core_count: int | None = None used_crossbar_count: int | None = None _MEMORY_UNITS = {"B": 1, "KB": 1 << 10, "MB": 1 << 20, "GB": 1 << 30} def collect_pim_resource_metrics(pim_dir): pim_dir = Path(pim_dir) report_path = pim_dir.parent / "reports" / "memory_report.txt" report = report_path.read_text(encoding="utf-8") if report_path.exists() else "" def memory_bytes(label): match = re.search(rf"^\s*{re.escape(label)}:\s+([0-9.]+)\s+(B|KB|MB|GB)$", report, re.MULTILINE) return round(float(match.group(1)) * _MEMORY_UNITS[match.group(2)]) if match else None with open(pim_dir / "config.json", encoding="utf-8") as f: config = json.load(f) used_cores = sum(read_raptor_instruction_count(path) > 0 for path in pim_dir.glob("core_*.pim")) used_crossbars = sum(sum(groups) for groups in config.get("array_group_map", {}).values()) return { "host_memory_bytes": memory_bytes("Host memory"), "cores_memory_bytes": memory_bytes("Local memory after reuse"), "used_core_count": used_cores, "used_crossbar_count": used_crossbars, } class ProgressReporter: def __init__(self, total_models, stages_per_model=STAGE_COUNT, enabled=None, verbose=False): self.total_models = total_models self.stages_per_model = stages_per_model self.total_steps = max(1, total_models * stages_per_model) self.completed_steps = 0 self.passed_models = 0 self.failed_models = 0 self.current_label = "" self.enabled = ( sys.stdout.isatty() and "CODEX_CI" not in os.environ if enabled is None else enabled ) self.verbose = verbose self.columns = max(1, shutil.get_terminal_size((100, 20)).columns) self.suspended = False self.rendered_width = 0 self.rendered_rows = 0 def _clear(self): if self.enabled and self.rendered_rows: columns = max(1, shutil.get_terminal_size((100, 20)).columns) rows = max(self.rendered_rows, (self.rendered_width + columns - 1) // columns) sys.stdout.write("\r\033[2K") for _ in range(rows - 1): sys.stdout.write("\033[1A\r\033[2K") sys.stdout.flush() self.rendered_width = 0 self.rendered_rows = 0 def _render(self): if not self.enabled or self.suspended: return self.columns = max(1, shutil.get_terminal_size((100, 20)).columns) bar_width = min(24, max(4, self.columns - 24)) filled = int(bar_width * self.completed_steps / self.total_steps) counts_text = f"P:{self.passed_models} F:{self.failed_models}" prefix_text = f"[{'#' * filled}{'-' * (bar_width - filled)}] {self.completed_steps}/{self.total_steps}" bar = Fore.GREEN + ("#" * filled) + Fore.CYAN + ("-" * (bar_width - filled)) prefix = Fore.CYAN + f"[{bar}{Fore.CYAN}] {self.completed_steps}/{self.total_steps}" + Style.RESET_ALL counts = ( " " + Style.BRIGHT + Fore.GREEN + f"P:{self.passed_models}" + Style.RESET_ALL + " " + Style.BRIGHT + Fore.RED + f"F:{self.failed_models}" + Style.RESET_ALL ) model_counter = "" label = "" if self.current_label.startswith("[") and "] " in self.current_label: model_counter, label = self.current_label.split("] ", 1) model_counter = f" {model_counter}]" label = f" {label}" elif self.current_label: label = f" {self.current_label}" fixed_width = len(prefix_text) + len(model_counter) + len(counts_text) + 2 if fixed_width > self.columns: model_counter = "" fixed_width = len(prefix_text) + len(counts_text) + 2 if fixed_width > self.columns: prefix_text = f"{self.completed_steps}/{self.total_steps}" prefix = Fore.CYAN + prefix_text + Style.RESET_ALL fixed_width = len(prefix_text) + len(counts_text) + 2 if fixed_width > self.columns: counts = "" counts_text = "" fixed_width = len(prefix_text) + 1 available_label_width = max(0, self.columns - fixed_width) label = label[:available_label_width] plain_counts = f" {counts_text}" if counts_text else "" plain_line = prefix_text + model_counter + plain_counts + label rendered_line = prefix + model_counter + counts + label + Style.RESET_ALL self._clear() sys.stdout.write(rendered_line) sys.stdout.flush() self.rendered_width = len(plain_line) self.rendered_rows = max(1, (self.rendered_width + self.columns - 1) // self.columns) def log(self, message="", color=None): if not self.verbose: self._render() return if self.enabled: self._clear() if color: print(color + message + Style.RESET_ALL, flush=True) else: print(message, flush=True) self._render() def set_stage(self, model_index, model_total, model_name, stage_name): self.current_label = f"[{model_index}/{model_total}] {model_name} ยท {stage_name}" self._render() def advance(self): self.completed_steps = min(self.total_steps, self.completed_steps + 1) self._render() def record_result(self, passed): if passed: self.passed_models += 1 else: self.failed_models += 1 self._render() def suspend(self): if self.enabled: self._clear() self.suspended = True def resume(self): self.suspended = False self._render() def finish(self): if self.enabled: self.suspended = True self._clear() def run_command(cmd, cwd=None, reporter=None, timeout_sec=None, capture_output=False): return run_command_with_reporter( cmd, cwd=cwd, reporter=reporter, timeout_sec=timeout_sec, capture_output=capture_output, ) def load_pimcomp_hardware(config_path): with open(config_path, encoding="utf-8") as f: config = json.load(f) matrix = config["chip_config"]["core_config"]["matrix_config"] rows, cols = config["chip_config"]["network_config"]["layout"] xbar_rows, xbar_cols = matrix["xbar_size"] return { "core_count": config["chip_config"]["core_cnt"], "crossbar_count": matrix["xbar_array_count"], "crossbar_rows": xbar_rows, "crossbar_cols": xbar_cols, "mesh_rows": rows, "mesh_cols": cols, } def pimcomp_compatibility_errors(config_path, *, core_count, crossbar_count, crossbar_size): hardware = load_pimcomp_hardware(config_path) errors = [] if hardware["mesh_rows"] * hardware["mesh_cols"] != hardware["core_count"]: errors.append( f"config layout {hardware['mesh_rows']}x{hardware['mesh_cols']} does not match " f"{hardware['core_count']} cores" ) if core_count != hardware["core_count"]: errors.append(f"--core-count={core_count}, config requires {hardware['core_count']}") if crossbar_count != hardware["crossbar_count"]: errors.append( f"--crossbar-count={crossbar_count}, config requires {hardware['crossbar_count']}" ) if ( hardware["crossbar_rows"] != hardware["crossbar_cols"] or crossbar_size != hardware["crossbar_rows"] ): errors.append( f"--crossbar-size={crossbar_size}, config requires " f"{hardware['crossbar_rows']}x{hardware['crossbar_cols']}" ) return errors def run_pimsim_nn( pimsim_nn_build_dir, pim_dir, config_path, execution_mode, reporter=None, timeout_sec=None, fast=True): pimsim_artifact = export_raptor_pimsim_artifact(pim_dir, Path(pim_dir).parent / "pimsim_nn") command = [pimsim_nn_build_dir / "ChipTest", pimsim_artifact, config_path, "--gui=false"] if fast: command.append("--fast") try: output = run_command( command, cwd=pimsim_nn_build_dir, reporter=reporter, timeout_sec=timeout_sec, capture_output=True, ) except subprocess.CalledProcessError as exc: error_output = exc.output.decode("utf-8", errors="replace") if isinstance(exc.output, bytes) else str(exc.output) if PIMSIM_UNSUPPORTED_VSOFTMAX in error_output: raise PimSimUnsupportedError(PIMSIM_UNSUPPORTED_VSOFTMAX) from exc raise metrics = parse_pimsim_nn_metrics(output) required = ( ("latency_ms", "average_power_mw", "average_energy_pj") if execution_mode == "latency" else ("throughput", "average_latency_ms", "average_power_mw", "average_energy_pj") ) if any(name not in metrics for name in required): raise RuntimeError(f"pimsim-nn output did not contain required {execution_mode} metrics") return metrics def clean_workspace_artifacts(workspace_dir, model_stem): workspace_dir = Path(workspace_dir) removed_paths = [] def remove_path(path): if path.is_symlink() or path.is_file(): path.unlink(missing_ok=True) removed_paths.append(path) elif path.is_dir(): shutil.rmtree(path) removed_paths.append(path) for name in GENERATED_DIR_NAMES: remove_path(workspace_dir / name) remove_path(workspace_dir / "inputs.csv") for suffix in (".onnx.mlir", ".so", ".tmp"): remove_path(workspace_dir / f"{model_stem}{suffix}") return removed_paths def print_stage(reporter, model_index, model_total, model_name, title, mode=None): if mode is not None: title = f"{title} ({mode.capitalize()})" color = STAGE_COLORS.get(title, STAGE_COLORS.get(title.split(" (", 1)[0], Fore.WHITE)) reporter.log(Style.BRIGHT + color + f"[{title}]" + Style.RESET_ALL) reporter.set_stage(model_index, model_total, model_name, title) def print_info(reporter, message): reporter.log(f" {message}") def compile_onnx_network(network_onnx_path, raptor_path, raptor_dir, runner_dir, reporter=None, timeout_sec=None): stem = network_onnx_path.stem onnx_ir_base = raptor_dir / stem runner_base = runner_dir / stem run_command([raptor_path, network_onnx_path, "-o", onnx_ir_base, "--EmitONNXIR", "--mlir-elide-elementsattrs-if-larger=16"], reporter=reporter, timeout_sec=timeout_sec) run_command([raptor_path, network_onnx_path, "-o", runner_base], reporter=reporter, timeout_sec=timeout_sec) network_so_path = runner_base.with_suffix(".so") network_mlir_path = onnx_ir_base.with_suffix(".onnx.mlir") onnx_ir_base.with_suffix(".tmp").unlink(missing_ok=True) return network_so_path, network_mlir_path def build_onnx_runner(source_dir, build_dir, reporter=None, timeout_sec=None): run_command(["cmake", source_dir], cwd=build_dir, reporter=reporter, timeout_sec=timeout_sec) run_command(["cmake", "--build", ".", "-j"], cwd=build_dir, reporter=reporter, timeout_sec=timeout_sec) return build_dir / "runner" def build_dump_ranges(config_path, outputs_descriptor): with open(config_path) as f: output_addresses = json.load(f)["outputs_addresses"] ranges = [] for addr, (_, _, dtype_code, shape) in zip(output_addresses, outputs_descriptor): byte_size = int(np.prod(shape)) * np.dtype(_ONNX_TO_NP[dtype_code]).itemsize ranges.append(f"{addr},{byte_size}") return ",".join(ranges) def build_pim_simulator_command( pim_dir, output_bin_path, dump_ranges, input_paths, mode="latency", batch_output_dir=None): if mode not in ("latency", "throughput"): raise ValueError(f"unknown simulator mode: {mode}") if not input_paths: raise ValueError("simulator requires at least one input") command = [ "cargo", "run", "--no-default-features", "--release", "--package", "pim-simulator", "--bin", "pim-simulator", "--", "-f", str(pim_dir), "-o", str(output_bin_path), "-d", dump_ranges, "--mode", mode, "--batch-size", str(len(input_paths)), ] if batch_output_dir is not None: command += ["--batch-output-dir", str(batch_output_dir)] for path in input_paths: command += ["--input", str(path)] return command def run_pim_simulator( simulator_dir, pim_dir, output_bin_path, dump_ranges, reporter=None, timeout_sec=None, input_paths=(), mode="latency", batch_output_dir=None): command = build_pim_simulator_command( pim_dir, output_bin_path, dump_ranges, input_paths, mode=mode, batch_output_dir=batch_output_dir, ) run_command( command, cwd=simulator_dir, reporter=reporter, timeout_sec=timeout_sec, ) def parse_pim_simulator_outputs(output_bin_path, outputs_descriptor): raw = output_bin_path.read_bytes() arrays = [] offset = 0 for _, _, dtype_code, shape in outputs_descriptor: dtype = np.dtype(_ONNX_TO_NP[dtype_code]) count = int(np.prod(shape)) array = np.frombuffer(raw, dtype=dtype, count=count, offset=offset).reshape(shape) offset += count * dtype.itemsize arrays.append(array) return arrays def validate_outputs(sim_arrays, runner_out_dir, outputs_descriptor, threshold, rtol, verbose): all_passed = True rows = [] for sim_array, (oi, name, _, shape) in zip(sim_arrays, outputs_descriptor): csv_name = f"output{oi}_{sanitize_output_name(name)}.csv" runner_array = np.loadtxt(runner_out_dir / csv_name, delimiter=',', dtype=np.float32).reshape(shape) sim_array64 = sim_array.astype(np.float64) runner_array64 = runner_array.astype(np.float64) abs_diff = np.abs(sim_array64 - runner_array64) allowed_diff = threshold + rtol * np.abs(runner_array64) max_diff = float(np.max(abs_diff)) passed = bool(np.all(abs_diff <= allowed_diff)) rows.append((name, f"{max_diff:.6e}", passed)) if not passed: all_passed = False name_width = max(len("Output"), *(len(name) for name, _, _ in rows)) diff_width = max(len("Max diff"), *(len(diff) for _, diff, _ in rows)) result_width = len("Result") separator = f" +-{'-' * name_width}-+-{'-' * diff_width}-+-{'-' * result_width}-+" if verbose or not all_passed: print(separator) print(f" | {'Output'.ljust(name_width)} | {'Max diff'.ljust(diff_width)} | {'Result'} |") print(separator) for name, diff_text, passed in rows: status_text = ("PASS" if passed else "FAIL").ljust(result_width) status = Fore.GREEN + status_text + Style.RESET_ALL if passed else Fore.RED + status_text + Style.RESET_ALL print(f" | {name.ljust(name_width)} | {diff_text.ljust(diff_width)} | {status} |") print(separator) return all_passed def report_validation_failure(reporter, execution_name, stage, exc): reporter.suspend() print( Fore.RED + f"{execution_name.capitalize()} {stage} failed: " f"{type(exc).__name__}: {exc}" + Style.RESET_ALL, file=sys.stderr, flush=True, ) reporter.resume() def validate_execution( execution, state, functional_data, workspace_dir, simulator_dir, pimsim_nn_build_dir, threshold, rtol, verbose, command_timeout_seconds, stage_context, pimsim_fast): reporter, model_index, model_total, model_name = stage_context name = execution["name"] pim_dir = execution["root"] / "pim" batch_size = execution["batch_size"] if state["compiled"] and functional_data is not None: input_batch, input_paths, reference_dirs, outputs_descriptor = functional_data simulation_dir = workspace_dir / "simulation" / name try: print_stage( reporter, model_index, model_total, model_name, "Run Functional Simulation", name, ) write_inputs_to_memory_bin( pim_dir / "memory.bin", pim_dir / "config.json", input_batch[0]) simulation_dir.mkdir(parents=True, exist_ok=True) dump_ranges = build_dump_ranges(pim_dir / "config.json", outputs_descriptor) output_dir = simulation_dir / "outputs" run_pim_simulator( simulator_dir, pim_dir, simulation_dir / "out.bin", dump_ranges, reporter=reporter, timeout_sec=command_timeout_seconds, input_paths=input_paths[:batch_size], mode=name, batch_output_dir=output_dir) reporter.advance() print_stage( reporter, model_index, model_total, model_name, "Compare Outputs", name, ) reporter.suspend() try: iteration_results = [ validate_outputs( parse_pim_simulator_outputs( output_dir / f"output_{index:06d}.bin", outputs_descriptor), reference_dirs[index], outputs_descriptor, threshold, rtol=rtol, verbose=verbose) for index in range(batch_size) ] finally: reporter.resume() state["passed"] = all(iteration_results) reporter.advance() except Exception as exc: report_validation_failure(reporter, name, "functional validation", exc) print_stage( reporter, model_index, model_total, model_name, "Run Non-functional Simulation", name, ) config_path = execution["pimsim_config"] if state["compiled"] and pimsim_nn_build_dir is not None and config_path is not None: try: state["metrics"] = run_pimsim_nn( pimsim_nn_build_dir, pim_dir, config_path, name, reporter=reporter, timeout_sec=command_timeout_seconds, fast=pimsim_fast) state["pimsim_status"] = PIMSIM_DONE metric = ( f"Latency: {state['metrics']['latency_ms']:.2f} ms" if name == "latency" else f"Throughput: {state['metrics']['throughput']:.2f} samples/s") energy_unit = "pJ" if name == "latency" else "pJ/it" print_info( reporter, f"{metric}, Power: {state['metrics']['average_power_mw']:.2f} mW, " f"Energy: {state['metrics']['average_energy_pj']:.2f} {energy_unit}") except PimSimUnsupportedError as exc: state["pimsim_status"] = PIMSIM_UNSUPPORTED print_info(reporter, str(exc)) except Exception as exc: state["pimsim_status"] = PIMSIM_FAILED report_validation_failure(reporter, name, "non-functional validation", exc) elif not state["compiled"]: state["pimsim_status"] = PIMSIM_NOT_RUN else: print_info(reporter, "pimsim-nn non-functional simulation skipped") reporter.advance() def validate_network(network_onnx_path, raptor_path, onnx_include_dir, simulator_dir, crossbar_size, crossbar_count, core_count, raptor_extra_args, pimsim_nn_build_dir, pimsim_config_path, threshold, rtol, seed, reporter, model_index, model_total, verbose, command_timeout_seconds, mode, throughput_pipeline=None, throughput_batch_size=4, throughput_pimsim_config_path=None, pimsim_fast=True): if throughput_pipeline is not None and throughput_batch_size < 2: raise ValueError("throughput validation requires batch size greater than 1") network_onnx_path = Path(network_onnx_path).resolve() raptor_path = Path(raptor_path).resolve() onnx_include_dir = Path(onnx_include_dir).resolve() simulator_dir = Path(simulator_dir).resolve() if pimsim_nn_build_dir is not None: pimsim_nn_build_dir = Path(pimsim_nn_build_dir).resolve() if pimsim_config_path is not None: pimsim_config_path = Path(pimsim_config_path).resolve() if throughput_pimsim_config_path is not None: throughput_pimsim_config_path = Path(throughput_pimsim_config_path).resolve() compile_extra_args = list(raptor_extra_args or []) owns_reporter = reporter is None reporter = reporter or ProgressReporter(model_total, stages_per_model=len(MODE_STAGE_TITLES[mode]), verbose=verbose) workspace_dir = network_onnx_path.parent raptor_dir = workspace_dir / "raptor" runner_dir = workspace_dir / "runner" runner_build_dir = runner_dir / "build" if mode != MODE_RUN_ONLY: clean_workspace_artifacts(workspace_dir, network_onnx_path.stem) Path.mkdir(raptor_dir, parents=True, exist_ok=True) Path.mkdir(runner_build_dir, parents=True, exist_ok=True) reporter.log(Fore.CYAN + f"[{model_index}/{model_total}]" + Style.RESET_ALL + f" {Style.BRIGHT}Validating {network_onnx_path.name}{Style.RESET_ALL}") stem = network_onnx_path.stem network_so_path = runner_dir / f"{stem}.so" network_mlir_path = raptor_dir / f"{stem}.onnx.mlir" runner_path = runner_build_dir / "runner" executions = [{ "name": "latency", "root": raptor_dir, "batch_size": 1, "compile_args": compile_extra_args, "pimsim_config": pimsim_config_path, }] if throughput_pipeline is not None: throughput_args = [ arg for arg in compile_extra_args if not str(arg).startswith("--pipeline=") ] + [f"--pipeline={throughput_pipeline}"] executions.append({ "name": "throughput", "root": raptor_dir / "throughput", "batch_size": throughput_batch_size, "compile_args": throughput_args, "pimsim_config": throughput_pimsim_config_path, }) states = { execution["name"]: { "compiled": False, "passed": False, "metrics": {}, "pimsim_status": PIMSIM_SKIPPED, "compile_time_s": 0.0, "resource_metrics": {}, } for execution in executions } pim_pass_timings = {} compile_time_s = 0.0 resource_metrics = {} try: reference_ready = False if mode != MODE_RUN_ONLY: try: print_stage(reporter, model_index, model_total, network_onnx_path.name, "Compile ONNX") network_so_path, network_mlir_path = compile_onnx_network( network_onnx_path, raptor_path, raptor_dir, runner_dir, reporter=reporter, timeout_sec=command_timeout_seconds) print_info(reporter, f"MLIR saved to {network_mlir_path}") print_info(reporter, f"Shared library saved to {network_so_path}") reporter.advance() print_stage(reporter, model_index, model_total, network_onnx_path.name, "Build Runner") gen_network_runner( network_onnx_path, network_so_path, onnx_include_dir, entry="run_main_graph", out=runner_dir / "runner.c", verbose=False) runner_path = build_onnx_runner( runner_dir, runner_build_dir, reporter=reporter, timeout_sec=command_timeout_seconds) print_info(reporter, f"Runner built at {runner_path}") reporter.advance() reference_ready = True except Exception as exc: report_validation_failure(reporter, "reference", "compilation", exc) else: required_paths = (network_so_path, network_mlir_path, runner_path) reference_ready = all(path.exists() for path in required_paths) if not reference_ready: report_validation_failure(reporter, "reference", "artifact lookup", FileNotFoundError( "run-only mode requires the compiled shared library, ONNX MLIR, and runner")) for execution in executions: name = execution["name"] root = execution["root"] pim_dir = root / "pim" if mode == MODE_RUN_ONLY: states[name]["compiled"] = (pim_dir / "config.json").exists() if not states[name]["compiled"]: report_validation_failure(reporter, name, "artifact lookup", FileNotFoundError( f"run-only mode requires compiled PIM artifacts at {pim_dir}")) else: states[name]["resource_metrics"] = collect_pim_resource_metrics(pim_dir) if name == "latency": resource_metrics = states[name]["resource_metrics"] continue try: print_stage( reporter, model_index, model_total, network_onnx_path.name, "Compile PIM", name, ) root.mkdir(parents=True, exist_ok=True) started = time.perf_counter() timings = compile_with_raptor( network_onnx_path, raptor_path, root / stem, crossbar_size, crossbar_count, core_count=core_count, raptor_extra_args=execution["compile_args"], cwd=root, verbose=verbose, reporter=reporter, timeout_sec=command_timeout_seconds) elapsed = time.perf_counter() - started compile_time_s += elapsed states[name]["compile_time_s"] = elapsed for label, duration in timings.items(): pim_pass_timings[label] = pim_pass_timings.get(label, 0) + duration states[name]["compiled"] = True states[name]["resource_metrics"] = collect_pim_resource_metrics(pim_dir) if name == "latency": resource_metrics = states[name]["resource_metrics"] print_info(reporter, f"PIM artifacts saved to {pim_dir}") except Exception as exc: report_validation_failure(reporter, name, "compilation", exc) reporter.advance() if mode == MODE_COMPILE_ONLY: for state in states.values(): state["passed"] = reference_ready and state["compiled"] else: input_batch = input_paths = reference_dirs = outputs_descriptor = None try: print_stage(reporter, model_index, model_total, network_onnx_path.name, "Generate Inputs") inputs_descriptor, outputs_descriptor = onnx_io(network_onnx_path) first_inputs, _ = gen_random_inputs(inputs_descriptor, seed=seed) input_batch = generate_input_batch( inputs_descriptor, first_inputs, max(execution["batch_size"] for execution in executions), seed) write_input_batch_csv(workspace_dir / "inputs.csv", input_batch) input_paths = write_input_batch_binaries(input_batch, workspace_dir / "simulation" / "inputs") input_flags = [ save_inputs_to_files( network_onnx_path, inputs, out_dir=workspace_dir / "inputs" / f"{index:06d}")[0] for index, inputs in enumerate(input_batch) ] print_info(reporter, f"Saved {len(input_batch)} input sample(s) to {workspace_dir / 'inputs.csv'}") reporter.advance() if not reference_ready: raise FileNotFoundError("reference runner is unavailable") print_stage(reporter, model_index, model_total, network_onnx_path.name, "Run Reference") reference_dirs = [] for index, flags in enumerate(input_flags): reference_dir = workspace_dir / "outputs" / f"{index:06d}" reference_dir.mkdir(parents=True, exist_ok=True) run_command( [runner_path, *flags, "--save-csv-dir", str(reference_dir)], cwd=runner_build_dir, reporter=reporter, timeout_sec=command_timeout_seconds) reference_dirs.append(reference_dir) print_info(reporter, f"Reference outputs saved for {len(reference_dirs)} sample(s)") reporter.advance() except Exception as exc: report_validation_failure(reporter, "reference", "execution", exc) functional_data = None if all(value is not None for value in ( input_batch, input_paths, reference_dirs, outputs_descriptor)): functional_data = input_batch, input_paths, reference_dirs, outputs_descriptor stage_context = reporter, model_index, model_total, network_onnx_path.name for execution in executions: validate_execution( execution, states[execution["name"]], functional_data, workspace_dir, simulator_dir, pimsim_nn_build_dir, threshold, rtol, verbose, command_timeout_seconds, stage_context, pimsim_fast) latency = states["latency"] throughput = states.get("throughput") passed = all(state["passed"] for state in states.values()) latency_metrics = latency["metrics"] throughput_metrics = throughput["metrics"] if throughput else {} reporter.record_result(passed) status = Fore.GREEN + "PASS" + Style.RESET_ALL if passed else Fore.RED + "FAIL" + Style.RESET_ALL reporter.log(Style.BRIGHT + f"Result: {status}" + Style.RESET_ALL) mode_metrics = { name: { "compile_time_s": state["compile_time_s"] or None, **state["resource_metrics"], } for name, state in states.items() } return ValidationResult( passed=passed, latency_passed=latency["passed"], throughput_passed=throughput["passed"] if throughput else None, pim_pass_timings=pim_pass_timings, pimsim_latency_ms=latency_metrics.get("latency_ms"), pimsim_throughput_samples_s=throughput_metrics.get("throughput"), pimsim_power_mw=latency_metrics.get("average_power_mw"), pimsim_energy_pj=latency_metrics.get("average_energy_pj"), pimsim_throughput_average_latency_ms=throughput_metrics.get("average_latency_ms"), pimsim_throughput_average_power_mw=throughput_metrics.get("average_power_mw"), pimsim_throughput_average_energy_pj=throughput_metrics.get("average_energy_pj"), mode_metrics=mode_metrics, pimsim_status=latency["pimsim_status"], throughput_pimsim_status=( throughput["pimsim_status"] if throughput else PIMSIM_SKIPPED), compile_time_s=compile_time_s or None, **resource_metrics, ) finally: reporter.log("=" * 72) if owns_reporter: reporter.finish()