| 9 | |
| 10 | |
| 11 | def execute_notebook(path: Path, out_path: Path | None = None) -> None: |
| 12 | nb = json.loads(path.read_text(encoding="utf-8")) |
| 13 | g: dict[str, object] = {"__name__": "__main__"} |
| 14 | exec_count = 1 |
| 15 | destination = out_path or path |
| 16 | destination.parent.mkdir(parents=True, exist_ok=True) |
| 17 | |
| 18 | for idx, cell in enumerate(nb.get("cells", [])): |
| 19 | if cell.get("cell_type") != "code": |
| 20 | continue |
| 21 | code = "".join(cell.get("source", [])) |
| 22 | stdout = io.StringIO() |
| 23 | outputs = [] |
| 24 | try: |
| 25 | with contextlib.redirect_stdout(stdout): |
| 26 | exec(compile(code, f"{path.name}:cell-{idx}", "exec"), g, g) |
| 27 | except Exception: |
| 28 | tb = traceback.format_exc() |
| 29 | outputs.append({ |
| 30 | "output_type": "error", |
| 31 | "ename": "ExecutionError", |
| 32 | "evalue": f"cell {idx}", |
| 33 | "traceback": tb.splitlines(), |
| 34 | }) |
| 35 | cell["execution_count"] = exec_count |
| 36 | cell["outputs"] = outputs |
| 37 | destination.write_text(json.dumps(nb, indent=1), encoding="utf-8") |
| 38 | raise RuntimeError(f"Notebook execution failed at cell {idx}\n{tb}") |
| 39 | |
| 40 | text = stdout.getvalue() |
| 41 | if text: |
| 42 | outputs.append({"output_type": "stream", "name": "stdout", "text": text}) |
| 43 | cell["execution_count"] = exec_count |
| 44 | cell["outputs"] = outputs |
| 45 | exec_count += 1 |
| 46 | |
| 47 | destination.write_text(json.dumps(nb, indent=1), encoding="utf-8") |
| 48 | |
| 49 | |
| 50 | def main() -> None: |