Invoke ``sol-execbench`` and return parsed trace dicts (or None on error).
(
definition_path: Path,
workload_path: Path,
solution_path: Path,
output_dir: Path,
job_name: str,
timeout: int,
config_path: Path | None = None,
keep_staging: bool = False,
verbose: bool = False,
)
| 202 | |
| 203 | |
| 204 | def run_cli( |
| 205 | definition_path: Path, |
| 206 | workload_path: Path, |
| 207 | solution_path: Path, |
| 208 | output_dir: Path, |
| 209 | job_name: str, |
| 210 | timeout: int, |
| 211 | config_path: Path | None = None, |
| 212 | keep_staging: bool = False, |
| 213 | verbose: bool = False, |
| 214 | ) -> list[dict] | None: |
| 215 | """Invoke ``sol-execbench`` and return parsed trace dicts (or None on error).""" |
| 216 | cmd = [ |
| 217 | str(Path(sys.executable).parent / "sol-execbench"), |
| 218 | "--definition", |
| 219 | str(definition_path), |
| 220 | "--workload", |
| 221 | str(workload_path), |
| 222 | "--solution", |
| 223 | str(solution_path), |
| 224 | "--timeout", |
| 225 | str(timeout), |
| 226 | "--json", |
| 227 | ] |
| 228 | |
| 229 | if config_path: |
| 230 | cmd.extend(["--config", str(config_path)]) |
| 231 | if keep_staging: |
| 232 | cmd.append("--keep-staging") |
| 233 | if verbose: |
| 234 | cmd.append("--verbose") |
| 235 | |
| 236 | result = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout + 60) |
| 237 | |
| 238 | # The CLI with --json prints one JSON trace per line to stdout. |
| 239 | traces = [] |
| 240 | for line in result.stdout.splitlines(): |
| 241 | line = line.strip() |
| 242 | if line: |
| 243 | try: |
| 244 | traces.append(json.loads(line)) |
| 245 | except json.JSONDecodeError: |
| 246 | continue |
| 247 | |
| 248 | if not traces: |
| 249 | print(f"CLI failed for {job_name}: {result.stderr[:500]}") |
| 250 | _save_cli_log(output_dir, job_name, result) |
| 251 | return None |
| 252 | |
| 253 | return traces |
| 254 | |
| 255 | |
| 256 | def _save_cli_log(output_dir: Path, job_name: str, result: subprocess.CompletedProcess): |