Buckets:
tahamajs/Sysmem2_in_AI / ComputerAssignments /CA8_neural_program_synthesis /run_notebook_complete.py
| #!/usr/bin/env python3 | |
| """ | |
| Comprehensive notebook executor with output capture and debugging | |
| """ | |
| import json | |
| import sys | |
| import io | |
| import traceback | |
| import base64 | |
| from contextlib import redirect_stdout, redirect_stderr | |
| import warnings | |
| import matplotlib | |
| matplotlib.use("Agg") # Non-interactive backend | |
| import matplotlib.pyplot as plt | |
| warnings.filterwarnings("ignore") | |
| def capture_matplotlib_figures(): | |
| """Capture all open matplotlib figures as base64 PNG""" | |
| figures = [] | |
| for fig_num in plt.get_fignums(): | |
| fig = plt.figure(fig_num) | |
| buf = io.BytesIO() | |
| fig.savefig(buf, format="png", bbox_inches="tight", dpi=100) | |
| buf.seek(0) | |
| img_base64 = base64.b64encode(buf.read()).decode("utf-8") | |
| figures.append(img_base64) | |
| buf.close() | |
| plt.close("all") | |
| return figures | |
| def execute_cell(code, global_vars, cell_index): | |
| """Execute a code cell and capture all outputs""" | |
| stdout_capture = io.StringIO() | |
| stderr_capture = io.StringIO() | |
| outputs = [] | |
| try: | |
| # Execute code | |
| with redirect_stdout(stdout_capture), redirect_stderr(stderr_capture): | |
| # Try to handle display/print statements | |
| exec(code, global_vars) | |
| # Capture text output | |
| text_output = stdout_capture.getvalue() | |
| if text_output: | |
| outputs.append( | |
| { | |
| "output_type": "stream", | |
| "name": "stdout", | |
| "text": text_output.split("\n"), | |
| } | |
| ) | |
| # Capture stderr if any | |
| stderr_output = stderr_capture.getvalue() | |
| if stderr_output: | |
| outputs.append( | |
| { | |
| "output_type": "stream", | |
| "name": "stderr", | |
| "text": stderr_output.split("\n"), | |
| } | |
| ) | |
| # Capture matplotlib figures | |
| figures = capture_matplotlib_figures() | |
| for fig_base64 in figures: | |
| outputs.append( | |
| { | |
| "output_type": "display_data", | |
| "data": {"image/png": fig_base64}, | |
| "metadata": {}, | |
| } | |
| ) | |
| return True, outputs, None | |
| except Exception as e: | |
| error_output = { | |
| "output_type": "error", | |
| "ename": type(e).__name__, | |
| "evalue": str(e), | |
| "traceback": traceback.format_exc().split("\n"), | |
| } | |
| # Also add any stdout before error | |
| text_output = stdout_capture.getvalue() | |
| if text_output: | |
| outputs.insert( | |
| 0, | |
| { | |
| "output_type": "stream", | |
| "name": "stdout", | |
| "text": text_output.split("\n"), | |
| }, | |
| ) | |
| outputs.append(error_output) | |
| return False, outputs, str(e) | |
| def main(): | |
| notebook_path = "notebooks/CA8.ipynb" | |
| output_path = "notebooks/CA8_executed.ipynb" | |
| print("=" * 70) | |
| print("NEURAL PROGRAM SYNTHESIS - NOTEBOOK EXECUTION") | |
| print("=" * 70) | |
| # Read notebook | |
| print(f"\nReading notebook: {notebook_path}") | |
| with open(notebook_path, "r", encoding="utf-8") as f: | |
| notebook = json.load(f) | |
| # Global namespace for execution | |
| global_vars = {"__name__": "__main__"} | |
| # Statistics | |
| total_cells = sum(1 for cell in notebook["cells"] if cell["cell_type"] == "code") | |
| executed_cells = 0 | |
| successful_cells = 0 | |
| failed_cells = 0 | |
| print(f"Total code cells to execute: {total_cells}\n") | |
| # Execute cells | |
| for i, cell in enumerate(notebook["cells"]): | |
| if cell["cell_type"] == "code": | |
| executed_cells += 1 | |
| print(f"\n{'─'*70}") | |
| print(f"Cell {executed_cells}/{total_cells} (Index {i})") | |
| print(f"{'─'*70}") | |
| # Get cell code | |
| code = "".join(cell["source"]) | |
| # Show code preview | |
| code_lines = code.split("\n") | |
| preview_lines = min(3, len(code_lines)) | |
| print("Code preview:") | |
| for line in code_lines[:preview_lines]: | |
| print(f" {line}") | |
| if len(code_lines) > preview_lines: | |
| print(f" ... ({len(code_lines) - preview_lines} more lines)") | |
| # Execute | |
| print("\nExecuting...", end=" ") | |
| success, outputs, error = execute_cell(code, global_vars, i) | |
| if success: | |
| print("✓ SUCCESS") | |
| successful_cells += 1 | |
| # Show output summary | |
| if outputs: | |
| print(f"Outputs captured: {len(outputs)} items") | |
| for output in outputs: | |
| if output["output_type"] == "stream": | |
| lines = ( | |
| output["text"] | |
| if isinstance(output["text"], list) | |
| else output["text"].split("\n") | |
| ) | |
| non_empty = [l for l in lines if l.strip()] | |
| if non_empty: | |
| print(f" - Text output: {len(non_empty)} lines") | |
| # Show first line | |
| if non_empty: | |
| first_line = non_empty[0][:70] | |
| print(f" First: {first_line}...") | |
| elif output["output_type"] == "display_data": | |
| print(f" - Figure captured") | |
| elif output["output_type"] == "error": | |
| print(f" - Error: {output['ename']}") | |
| else: | |
| print("✗ FAILED") | |
| failed_cells += 1 | |
| print(f"Error: {error[:100]}...") | |
| # Add outputs and execution count to cell | |
| cell["outputs"] = outputs | |
| cell["execution_count"] = executed_cells | |
| # Save updated notebook | |
| print(f"\n{'='*70}") | |
| print("EXECUTION SUMMARY") | |
| print(f"{'='*70}") | |
| print(f"Total cells executed: {executed_cells}") | |
| print(f"Successful: {successful_cells} ({successful_cells/total_cells*100:.1f}%)") | |
| print(f"Failed: {failed_cells} ({failed_cells/total_cells*100:.1f}%)") | |
| print(f"\nSaving notebook to: {output_path}") | |
| with open(output_path, "w", encoding="utf-8") as f: | |
| json.dump(notebook, f, indent=1, ensure_ascii=False) | |
| print(f"✓ Notebook saved successfully!") | |
| print(f"\n{'='*70}") | |
| return 0 if failed_cells == 0 else 1 | |
| if __name__ == "__main__": | |
| sys.exit(main()) | |
Xet Storage Details
- Size:
- 6.66 kB
- Xet hash:
- 9e8838d011a97f02c2698dc2a5646a7b6852a75d90da0a70bf5dbc9aca28392b
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.