Bezalel
88b4cc218f
Some checks failed
Notebook CI / notebook-smoke (pull_request) Failing after 2s
- gitea_client.py — reusable Gitea API client for issues, PRs, comments - health.py — fleet health monitor (load, disk, memory, processes) - notebook_runner.py — Papermill wrapper with JSON reporting - smoke_test.py — fast smoke tests and bare green-path e2e - secret_scan.py — secret leak scanner for CI gating - wizard_env.py — environment validator for bootstrapping agents - README.md — usage guide for all tools These tools are designed to be used by any wizard via python -m devkit.<tool>. Rising up as a platform, not a silo.
137 lines
4.2 KiB
Python
137 lines
4.2 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Notebook execution runner for agent tasks.
|
|
Wraps papermill with sensible defaults and structured JSON reporting.
|
|
|
|
Usage as CLI:
|
|
python -m devkit.notebook_runner notebooks/task.ipynb output.ipynb -p threshold 1.0
|
|
python -m devkit.notebook_runner notebooks/task.ipynb --dry-run
|
|
|
|
Usage as module:
|
|
from devkit.notebook_runner import run_notebook
|
|
result = run_notebook("task.ipynb", "output.ipynb", parameters={"threshold": 1.0})
|
|
"""
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
from pathlib import Path
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
|
|
def run_notebook(
|
|
input_path: str,
|
|
output_path: Optional[str] = None,
|
|
parameters: Optional[Dict[str, Any]] = None,
|
|
kernel: str = "python3",
|
|
timeout: Optional[int] = None,
|
|
dry_run: bool = False,
|
|
) -> Dict[str, Any]:
|
|
input_path = str(Path(input_path).expanduser().resolve())
|
|
if output_path is None:
|
|
fd, output_path = tempfile.mkstemp(suffix=".ipynb")
|
|
os.close(fd)
|
|
else:
|
|
output_path = str(Path(output_path).expanduser().resolve())
|
|
|
|
if dry_run:
|
|
return {
|
|
"status": "dry_run",
|
|
"input": input_path,
|
|
"output": output_path,
|
|
"parameters": parameters or {},
|
|
"kernel": kernel,
|
|
}
|
|
|
|
cmd = ["papermill", input_path, output_path, "--kernel", kernel]
|
|
if timeout is not None:
|
|
cmd.extend(["--execution-timeout", str(timeout)])
|
|
for key, value in (parameters or {}).items():
|
|
cmd.extend(["-p", key, str(value)])
|
|
|
|
start = os.times()
|
|
try:
|
|
proc = subprocess.run(
|
|
cmd,
|
|
capture_output=True,
|
|
text=True,
|
|
check=True,
|
|
)
|
|
end = os.times()
|
|
return {
|
|
"status": "ok",
|
|
"input": input_path,
|
|
"output": output_path,
|
|
"parameters": parameters or {},
|
|
"kernel": kernel,
|
|
"elapsed_seconds": round((end.elapsed - start.elapsed), 2),
|
|
"stdout": proc.stdout[-2000:] if proc.stdout else "",
|
|
}
|
|
except subprocess.CalledProcessError as e:
|
|
end = os.times()
|
|
return {
|
|
"status": "error",
|
|
"input": input_path,
|
|
"output": output_path,
|
|
"parameters": parameters or {},
|
|
"kernel": kernel,
|
|
"elapsed_seconds": round((end.elapsed - start.elapsed), 2),
|
|
"stdout": e.stdout[-2000:] if e.stdout else "",
|
|
"stderr": e.stderr[-2000:] if e.stderr else "",
|
|
"returncode": e.returncode,
|
|
}
|
|
except FileNotFoundError:
|
|
return {
|
|
"status": "error",
|
|
"message": "papermill not found. Install with: uv tool install papermill",
|
|
}
|
|
|
|
|
|
def main(argv: List[str] = None) -> int:
|
|
argv = argv or sys.argv[1:]
|
|
parser = argparse.ArgumentParser(description="Notebook runner for agents")
|
|
parser.add_argument("input", help="Input notebook path")
|
|
parser.add_argument("output", nargs="?", default=None, help="Output notebook path")
|
|
parser.add_argument("-p", "--parameter", action="append", default=[], help="Parameters as key=value")
|
|
parser.add_argument("--kernel", default="python3")
|
|
parser.add_argument("--timeout", type=int, default=None)
|
|
parser.add_argument("--dry-run", action="store_true")
|
|
args = parser.parse_args(argv)
|
|
|
|
parameters = {}
|
|
for raw in args.parameter:
|
|
if "=" not in raw:
|
|
print(f"Invalid parameter (expected key=value): {raw}", file=sys.stderr)
|
|
return 1
|
|
k, v = raw.split("=", 1)
|
|
# Best-effort type inference
|
|
if v.lower() in ("true", "false"):
|
|
v = v.lower() == "true"
|
|
else:
|
|
try:
|
|
v = int(v)
|
|
except ValueError:
|
|
try:
|
|
v = float(v)
|
|
except ValueError:
|
|
pass
|
|
parameters[k] = v
|
|
|
|
result = run_notebook(
|
|
args.input,
|
|
args.output,
|
|
parameters=parameters,
|
|
kernel=args.kernel,
|
|
timeout=args.timeout,
|
|
dry_run=args.dry_run,
|
|
)
|
|
print(json.dumps(result, indent=2))
|
|
return 0 if result.get("status") == "ok" else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|