02_shell.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "efe78920",
   "metadata": {},
   "outputs": [],
   "source": [
    "#|default_exp shell"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "3d773712-12fe-440e-891f-36f59666dfde",
   "metadata": {},
   "source": [
    "# shell source"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "1328ef69",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| exports\n",
    "import ast, time, signal, traceback\n",
    "from fastcore.utils import *"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "481b4368",
   "metadata": {},
   "source": [
    "`get_shell` is like `python`, except it also maintains a stateful interpreter, rather than just running a single line of code. This is implemented using IPython, so that must be installed."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "6bbf062d",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| exports\n",
    "from IPython.terminal.interactiveshell import TerminalInteractiveShell\n",
    "from IPython.utils.capture import capture_output"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "d3d04ec5",
   "metadata": {},
   "outputs": [],
   "source": [
    "def exception2str(ex:Exception)->str:\n",
    "    \"Convert exception `ex` into a string\"\n",
    "    return ''.join(traceback.format_exception(type(ex), ex, ex.__traceback__))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "d6ba32b4",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Traceback (most recent call last):\n",
      "  File \"/var/folders/ss/34z569j921v58v8n1n_8z7h40000gn/T/ipykernel_37260/4058275565.py\", line 1, in <module>\n",
      "    try: print(1/0)\n",
      "               ~^~\n",
      "ZeroDivisionError: division by zero\n",
      "\n"
     ]
    }
   ],
   "source": [
    "try: print(1/0)\n",
    "except Exception as e: print(exception2str(e))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "34099c2f",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| exports\n",
    "TerminalInteractiveShell.orig_run = TerminalInteractiveShell.run_cell"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "d6aa8e7b",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| exports\n",
    "@patch\n",
    "def run_cell(self:TerminalInteractiveShell, cell, timeout=None):\n",
    "    \"Wrapper for original `run_cell` which adds timeout and output capture\"\n",
    "    if timeout:\n",
    "        def handler(*args): raise TimeoutError()\n",
    "        signal.signal(signal.SIGALRM, handler)\n",
    "        signal.alarm(timeout)\n",
    "    try:\n",
    "        with capture_output() as io: result = self.orig_run(cell)\n",
    "        result.stdout = io.stdout\n",
    "        return result\n",
    "    except TimeoutException as e:\n",
    "        result = self.ExecutionResult(error_before_exec=None, error_in_exec=e)\n",
    "    finally:\n",
    "        if timeout: signal.alarm(0)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "cdadbb12",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| exports\n",
    "def get_shell()->TerminalInteractiveShell:\n",
    "    \"Get a `TerminalInteractiveShell` with minimal functionality\"\n",
    "    sh = TerminalInteractiveShell()\n",
    "    sh.logger.log_output = sh.history_manager.enabled = False\n",
    "    dh = sh.displayhook\n",
    "    dh.finish_displayhook = dh.write_output_prompt = dh.start_displayhook = lambda: None\n",
    "    dh.write_format_data = lambda format_dict, md_dict=None: None\n",
    "    sh.logstart = sh.automagic = sh.autoindent = False\n",
    "    sh.autocall = 0\n",
    "    sh.system = lambda cmd: None\n",
    "    return sh"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "5ffbe57e",
   "metadata": {},
   "outputs": [],
   "source": [
    "shell = get_shell()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "b03b78b3",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "(2, '3\\n')"
      ]
     },
     "execution_count": null,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "r = shell.run_cell('print(3); 1+1')\n",
    "r.result,r.stdout"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "48849fc3",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Traceback (most recent call last):\n",
      "  File \"/Users/jhoward/miniconda3/lib/python3.12/site-packages/IPython/core/interactiveshell.py\", line 3577, in run_code\n",
      "    exec(code_obj, self.user_global_ns, self.user_ns)\n",
      "  File \"<ipython-input-1-338156281413>\", line 1, in <module>\n",
      "    raise Exception(\"blah\")\n",
      "Exception: blah\n",
      "\n"
     ]
    }
   ],
   "source": [
    "r = shell.run_cell('raise Exception(\"blah\")')\n",
    "print(exception2str(r.error_in_exec))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "ddabea6d",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "TimeoutError()"
      ]
     },
     "execution_count": null,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "r = shell.run_cell('import time; time.sleep(10)', timeout=1)\n",
    "r.error_in_exec"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "94ec4289",
   "metadata": {},
   "source": [
    "## Export -"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "1e9ee5c1",
   "metadata": {},
   "outputs": [],
   "source": [
    "#|hide\n",
    "#|eval: false\n",
    "from nbdev.doclinks import nbdev_export\n",
    "nbdev_export()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "207f9715",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "python3",
   "language": "python",
   "name": "python3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}