shell

A shell for running notebook code without a notebook server

from fastcore.test import *
from base64 import b64decode
from io import BytesIO
from PIL import Image

---

source

ExecutionResult.__repr__

 ExecutionResult.__repr__ ()

Return repr(self).

---

source

ExecutionInfo.__repr__

 ExecutionInfo.__repr__ ()

Return repr(self).

---

source

CaptureShell

 CaptureShell (path:str|pathlib.Path=None, mpl_format='retina',
               history=False, timeout:Optional[int]=None)

An enhanced, interactive shell for Python.

---

source

CaptureShell.run_cell

 CaptureShell.run_cell (raw_cell, store_history=False, silent=False,
                        shell_futures=True, cell_id=None, stdout=True,
                        stderr=True, display=True, timeout=None)

Run a complete IPython cell.

	Type	Default	Details
raw_cell	str		The code (including IPython code such as %magic functions) to run.
store_history	bool	False	If True, the raw and translated cell will be stored in IPython’s history. For user code calling back into IPython’s machinery, this should be set to False.
silent	bool	False	If True, avoid side-effects, such as implicit displayhooks and and logging. silent=True forces store_history=False.
shell_futures	bool	True	If True, the code will share future statements with the interactive shell. It will both be affected by previous future imports, and any future imports in the code will affect the shell. If False, future imports are not shared in either direction.
cell_id	NoneType	None
stdout	bool	True
stderr	bool	True
display	bool	True
timeout	NoneType	None
Returns	:class:ExecutionResult

s = CaptureShell(mpl_format='retina')

s.run_cell('a=1');

o = s.run_cell('print(a)')
o

{ 'display_objects': [],
  'exception': None,
  'quiet': False,
  'result': result: None; err: None; info: <cell: print(a); id: None>,
  'stderr': '',
  'stdout': '1\n'}

o = s.run_cell('from warnings import warn; warn("1")')
o

{ 'display_objects': [],
  'exception': None,
  'quiet': False,
  'result': result: None; err: None; info: <cell: from warnings import warn; warn("1"); id: None>,
  'stderr': '<ipython-input-1-a51443ae013a>:1: UserWarning: 1\n'
            '  from warnings import warn; warn("1")\n',
  'stdout': ''}

o = s.run_cell('1')
o

{ 'display_objects': [],
  'exception': None,
  'quiet': False,
  'result': result: 1; err: None; info: <cell: 1; id: None>,
  'stderr': '',
  'stdout': ''}

o = s.run_cell('from IPython.display import Markdown,display; print(0); display(Markdown("*2*")); Markdown("*1*")')
o

{ 'display_objects': [<IPython.utils.capture.RichOutput object>],
  'exception': None,
  'quiet': False,
  'result': result: <IPython.core.display.Markdown object>; err: None; info: <cell: from IPython.display import Markdown,display; print(0); display(Markdown("*2*")); Markdown("*1*"); id: None>,
  'stderr': '',
  'stdout': '0\n'}

o.result.result

1

o.display_objects[0]

2

o = s.run_cell('1;')
o

{ 'display_objects': [],
  'exception': None,
  'quiet': True,
  'result': result: 1; err: None; info: <cell: 1;; id: None>,
  'stderr': '',
  'stdout': ''}

o = s.run_cell('import matplotlib.pyplot as plt; plt.plot([1,2,3])')
o

{ 'display_objects': [<IPython.utils.capture.RichOutput object>],
  'exception': None,
  'quiet': False,
  'result': result: [<matplotlib.lines.Line2D object>]; err: None; info: <cell: import matplotlib.pyplot as plt; plt.plot([1,2,3]); id: None>,
  'stderr': '',
  'stdout': ''}

o.result.result[0]

o.display_objects[0]

o = s.run_cell('''
import pandas as pd
pd.DataFrame({'A': [1, 2], 'B': [3, 4]})''')
o

{ 'display_objects': [],
  'exception': None,
  'quiet': False,
  'result': result:    A  B
0  1  3
1  2  4; err: None; info: <cell: 
import pandas as pd
pd.DataFrame({'A': [1, 2], 'B': [3, 4]}); id: None>,
  'stderr': '',
  'stdout': ''}

o.result.result

	A	B
0	1	3
1	2	4

o = s.run_cell('1/0')
o

{ 'display_objects': [],
  'exception': ZeroDivisionError('division by zero'),
  'quiet': False,
  'result': result: None; err: division by zero; info: <cell: 1/0; id: None>,
  'stderr': '',
  'stdout': '\x1b[31m---------------------------------------------------------------------------\x1b[39m\n'
            '\x1b[31mZeroDivisionError\x1b[39m                         '
            'Traceback (most recent call last)\n'
            '\x1b[36mFile '
            '\x1b[39m\x1b[32m<ipython-input-1-9e1622b385b6>:1\x1b[39m\n'
            '\x1b[32m----> \x1b[39m\x1b[32m1\x1b[39m '
            '\x1b[32;43m1\x1b[39;49m\x1b[43m/\x1b[49m\x1b[32;43m0\x1b[39;49m\n'
            '\n'
            '\x1b[31mZeroDivisionError\x1b[39m: division by zero\n'}

o = s.run_cell('import time; time.sleep(2)', timeout=1)
o['exception']

TimeoutError()

Cells / run

---

source

format_exc

 format_exc (e)

Format exception e as a string list

---

source

NbResult

 NbResult (iterable=())

*Built-in mutable sequence.

If no argument is given, the constructor creates a new empty list. The argument must be an iterable if specified.*

---

source

CaptureShell.run

 CaptureShell.run (code:str, stdout=True, stderr=True,
                   timeout:Optional[int]=None)

Run code, returning a list of all outputs in Jupyter notebook format

	Type	Default	Details
code	str		Python/IPython code to run
stdout	bool	True	Capture stdout and save as output?
stderr	bool	True	Capture stderr and save as output?
timeout	Optional	None	Shell command will time out after {timeout} seconds

s = CaptureShell()

s.run("print(1)")

[{'name': 'stdout', 'output_type': 'stream', 'text': ['1\n']}]

Code can include magics and ! shell commands:

o = s.run("%time 1+1")
o

[{'name': 'stdout',
  'output_type': 'stream',
  'text': ['CPU times: user 2 us, sys: 0 ns, total: 2 us\n',
   'Wall time: 1.91 us\n']},
 {'data': {'text/plain': ['2']},
  'metadata': {},
  'output_type': 'execute_result',
  'execution_count': None}]

The result of the last successful execution is stored in result:

s.result

2

A trailing ; stops the result from being captured:

s.run("1+2;")

[]

o = s.run("1/0")
o

[{'name': 'stdout',
  'output_type': 'stream',
  'text': ['\x1b[31m---------------------------------------------------------------------------\x1b[39m\n',
   '\x1b[31mZeroDivisionError\x1b[39m                         Traceback (most recent call last)\n',
   '\x1b[36mFile \x1b[39m\x1b[32m<ipython-input-1-9e1622b385b6>:1\x1b[39m\n',
   '\x1b[32m----> \x1b[39m\x1b[32m1\x1b[39m \x1b[32;43m1\x1b[39;49m\x1b[43m/\x1b[49m\x1b[32;43m0\x1b[39;49m\n',
   '\n',
   '\x1b[31mZeroDivisionError\x1b[39m: division by zero\n']},
 {'ename': 'ZeroDivisionError',
  'evalue': 'division by zero',
  'output_type': 'error',
  'traceback': ['Traceback (most recent call last):\n',
   '  File "/Users/jhoward/uvws/.venv/lib/python3.12/site-packages/IPython/core/interactiveshell.py", line 3549, in run_code\n    exec(code_obj, self.user_global_ns, self.user_ns)\n',
   '  File "<ipython-input-1-9e1622b385b6>", line 1, in <module>\n    1/0\n    ~^~\n',
   'ZeroDivisionError: division by zero\n']}]

This is how IPython formats exceptions internally:

from IPython.core.ultratb import VerboseTB

with warnings.catch_warnings():
    warnings.filterwarnings("ignore", category=DeprecationWarning)
    formatter = VerboseTB(color_scheme='Linux')

try: f()
except Exception as e:
    ex = e
    print(formatter.text(type(e), e, e.__traceback__))

---------------------------------------------------------------------------
NameError                                 Traceback (most recent call last)
Cell In[35], line 1
----> 1 try: f()
      2 except Exception as e:
      3     ex = e

NameError: name 'f' is not defined

s.run("import time; time.sleep(0.1); print('no timeout')", timeout=1)

[{'name': 'stdout', 'output_type': 'stream', 'text': ['no timeout\n']}]

o = s.run("import time; time.sleep(1.1)", timeout=1)
o[0]['text'][:2]

['\x1b[31m---------------------------------------------------------------------------\x1b[39m\n',
 '\x1b[31mTimeoutError\x1b[39m                              Traceback (most recent call last)\n']

o1 = s.run('from IPython.display import Markdown,display; print(0); print(1); display(Markdown("*2*")); Markdown("*1*")')
o1

[{'name': 'stdout', 'output_type': 'stream', 'text': ['0\n', '1\n']},
 {'data': {'text/plain': ['<IPython.core.display.Markdown object>'],
   'text/markdown': ['*2*']},
  'metadata': {},
  'output_type': 'display_data'},
 {'data': {'text/plain': ['<IPython.core.display.Markdown object>'],
   'text/markdown': ['*1*']},
  'metadata': {},
  'output_type': 'execute_result',
  'execution_count': None}]

---

source

CaptureShell.run_async

 CaptureShell.run_async (code:str, stdout=True, stderr=True,
                         timeout:Optional[int]=None)

	Type	Default	Details
code	str		Python/IPython code to run
stdout	bool	True	Capture stdout and save as output?
stderr	bool	True	Capture stderr and save as output?
timeout	Optional	None	Shell command will time out after {timeout} seconds

await s.run_async("1+1")

[{'data': {'text/plain': ['2']},
  'metadata': {},
  'output_type': 'execute_result',
  'execution_count': None}]

---

source

render_outputs

 render_outputs (outputs, ansi_renderer=<function _strip>,
                 include_imgs=True, pygments=False, md_tfm=<function
                 noop>, html_tfm=<function noop>)

HTML(render_outputs(o))

---------------------------------------------------------------------------
TimeoutError                              Traceback (most recent call last)
File <ipython-input-1-a5c3817716b6>:1
----> 1 import time; time.sleep(1.1)

Cell In[6], line 7, in run_cell.<locals>.handler(*args)
      5 if not timeout: timeout = self.timeout
      6 if timeout:
----> 7     def handler(*args): raise TimeoutError()
      8     signal.signal(signal.SIGALRM, handler)
      9     signal.alarm(timeout)

TimeoutError: 

We can use ansi2html to convert from ANSI to HTML for color rendering. You need some css styles for the colors to render properly. Jupyter already has these built in so it’s not neccessary here, but if you plan on using this in another web app you will need to ensure that css styling is included.

HTML(render_outputs(o, ansi2html))

---------------------------------------------------------------------------
TimeoutError                              Traceback (most recent call last)
File <ipython-input-1-a5c3817716b6>:1
----> 1 import time; time.sleep(1.1)

Cell In[6], line 7, in run_cell.<locals>.handler(*args)
      5 if not timeout: timeout = self.timeout
      6 if timeout:
----> 7     def handler(*args): raise TimeoutError()
      8     signal.signal(signal.SIGALRM, handler)
      9     signal.alarm(timeout)

TimeoutError: 

Images and matplotlib figures are captured:

res = s.run('''import matplotlib.pyplot as plt
plt.figure(figsize=(2,1))
plt.plot([1,2,4]);''')

HTML(render_outputs(res))

If an exception is raised then the exception type, object, and stacktrace are stored in exc:

o = s.run('raise Exception("Oops")')
o

[{'name': 'stdout',
  'output_type': 'stream',
  'text': ['\x1b[31m---------------------------------------------------------------------------\x1b[39m\n',
   '\x1b[31mException\x1b[39m                                 Traceback (most recent call last)\n',
   '\x1b[36mFile \x1b[39m\x1b[32m<ipython-input-1-01648acb07bd>:1\x1b[39m\n',
   '\x1b[32m----> \x1b[39m\x1b[32m1\x1b[39m \x1b[38;5;28;01mraise\x1b[39;00m \x1b[38;5;167;01mException\x1b[39;00m(\x1b[33m"\x1b[39m\x1b[33mOops\x1b[39m\x1b[33m"\x1b[39m)\n',
   '\n',
   '\x1b[31mException\x1b[39m: Oops\n']},
 {'ename': 'Exception',
  'evalue': 'Oops',
  'output_type': 'error',
  'traceback': ['Traceback (most recent call last):\n',
   '  File "/Users/jhoward/uvws/.venv/lib/python3.12/site-packages/IPython/core/interactiveshell.py", line 3549, in run_code\n    exec(code_obj, self.user_global_ns, self.user_ns)\n',
   '  File "<ipython-input-1-01648acb07bd>", line 1, in <module>\n    raise Exception("Oops")\n',
   'Exception: Oops\n']}]

s.exc

Exception('Oops')

---

source

CaptureShell.cell

 CaptureShell.cell (cell, stdout=True, stderr=True)

Run cell, skipping if not code, and store outputs back in cell

clean = Path('../tests/clean.ipynb')
nb = read_nb(clean)
c = nb.cells[1]
c

{ 'cell_type': 'code',
  'execution_count': None,
  'id': 'b123d6d0',
  'idx_': 1,
  'metadata': {},
  'outputs': [],
  'source': 'print(1)\n2'}

s.cell(c)
c.outputs

[{'name': 'stdout', 'output_type': 'stream', 'text': ['1\n']},
 {'data': {'text/plain': ['2']},
  'metadata': {},
  'output_type': 'execute_result',
  'execution_count': None}]

---

source

find_output

 find_output (outp, ot='execute_result')

Find first output of type ot in CaptureShell.run output

	Type	Default	Details
outp			Output from run
ot	str	execute_result	Output_type to find

find_output(c.outputs)['data']

{'text/plain': ['2']}

find_output(c.outputs, 'stream')['text']

['1\n']

---

source

out_exec

 out_exec (outp)

Get data from execution result in outp.

out_exec(c.outputs)

'2'

---

source

out_stream

 out_stream (outp)

Get text from stream in outp.

out_stream(c.outputs)

'1'

---

source

out_error

 out_error (outp)

Get traceback from error in outp.

---

source

CaptureShell.run_all

 CaptureShell.run_all (nb, exc_stop:bool=False, preproc:<built-
                       infunctioncallable>=<function _false>,
                       postproc:<built-infunctioncallable>=<function
                       _false>, inject_code:str|None=None,
                       inject_idx:int=0)

Run all cells in nb, stopping at first exception if exc_stop

	Type	Default	Details
nb			A notebook read with nbclient or read_nb
exc_stop	bool	False	Stop on exceptions?
preproc	callable	_false	Called before each cell is executed
postproc	callable	_false	Called after each cell is executed
inject_code	str | None	None	Code to inject into a cell
inject_idx	int	0	Cell to replace with inject_code

nb.cells[2].outputs

[]

s.run_all(nb)
nb.cells[2].outputs

[{'data': {'text/plain': ['<IPython.core.display.Markdown object>'],
   'text/markdown': ["This is *bold*. Here's a [link](https://www.fast.ai)."]},
  'metadata': {},
  'output_type': 'execute_result',
  'execution_count': None}]

With exc_stop=False (the default), execution continues after exceptions, and exception details are stored into the appropriate cell’s output:

nb.cells[-1].source

'raise Exception("Oopsie!")'

nb.cells[-1].outputs

[{'name': 'stdout',
  'output_type': 'stream',
  'text': ['\x1b[31m---------------------------------------------------------------------------\x1b[39m\n',
   '\x1b[31mException\x1b[39m                                 Traceback (most recent call last)\n',
   '\x1b[36mFile \x1b[39m\x1b[32m<ipython-input-1-1c97c1d317ab>:1\x1b[39m\n',
   '\x1b[32m----> \x1b[39m\x1b[32m1\x1b[39m \x1b[38;5;28;01mraise\x1b[39;00m \x1b[38;5;167;01mException\x1b[39;00m(\x1b[33m"\x1b[39m\x1b[33mOopsie!\x1b[39m\x1b[33m"\x1b[39m)\n',
   '\n',
   '\x1b[31mException\x1b[39m: Oopsie!\n']},
 {'ename': 'Exception',
  'evalue': 'Oopsie!',
  'output_type': 'error',
  'traceback': ['Traceback (most recent call last):\n',
   '  File "/Users/jhoward/uvws/.venv/lib/python3.12/site-packages/IPython/core/interactiveshell.py", line 3549, in run_code\n    exec(code_obj, self.user_global_ns, self.user_ns)\n',
   '  File "<ipython-input-1-1c97c1d317ab>", line 1, in <module>\n    raise Exception("Oopsie!")\n',
   'Exception: Oopsie!\n']}]

With exc_stop=True, exceptions in a cell are raised and no further processing occurs:

try: s.run_all(nb, exc_stop=True)
except Exception as e: print(f"got exception: {e}")

got exception: Oopsie!

We can pass a function to preproc to have it run on every cell. It can modify the cell as needed. If the function returns True, then that cell will not be executed. For instance, to skip the cell which raises an exception:

nb = read_nb(clean)
s.run_all(nb, preproc=lambda c: 'raise' in c.source)

This cell will contain no output, since it was skipped.

nb.cells[-1].outputs

[]

nb.cells[1].outputs

[{'name': 'stdout', 'output_type': 'stream', 'text': ['1\n']},
 {'data': {'text/plain': ['2']},
  'metadata': {},
  'output_type': 'execute_result',
  'execution_count': None}]

You can also pass a function to postproc to modify a cell after it is executed.

---

source

CaptureShell.execute

 CaptureShell.execute (src:str|pathlib.Path, dest:str|None=None,
                       exc_stop:bool=False, preproc:<built-
                       infunctioncallable>=<function _false>,
                       postproc:<built-infunctioncallable>=<function
                       _false>, inject_code:str|None=None,
                       inject_path:str|pathlib.Path|None=None,
                       inject_idx:int=0)

Execute notebook from src and save with outputs to `dest

	Type	Default	Details
src	str | pathlib.Path		Notebook path to read from
dest	str | None	None	Notebook path to write to
exc_stop	bool	False	Stop on exceptions?
preproc	callable	_false	Called before each cell is executed
postproc	callable	_false	Called after each cell is executed
inject_code	str | None	None	Code to inject into a cell
inject_path	str | pathlib.Path | None	None	Path to file containing code to inject into a cell
inject_idx	int	0	Cell to replace with inject_code

This is a shortcut for the combination of read_nb, CaptureShell.run_all, and write_nb.

s = CaptureShell()
try:
    s.execute(clean, 'tmp.ipynb')
    print(read_nb('tmp.ipynb').cells[1].outputs)
finally: Path('tmp.ipynb').unlink()

[{'name': 'stdout', 'output_type': 'stream', 'text': ['1\n']}, {'data': {'text/plain': ['2']}, 'execution_count': None, 'metadata': {}, 'output_type': 'execute_result'}]

p = Path.home()/'git'/'fastcore'/'nbs'
n = p/'03a_parallel.ipynb'

---

source

CaptureShell.prettytb

 CaptureShell.prettytb (fname:str|pathlib.Path=None)

Show a pretty traceback for notebooks, optionally printing fname.

	Type	Default	Details
fname	str | pathlib.Path	None	filename to print alongside the traceback

If an error occurs while running a notebook, you can retrieve a pretty version of the error with the prettytb method:

s = CaptureShell()
try:
    s.execute('../tests/error.ipynb', exc_stop=True)
except:
    print(s.prettytb())

AssertionError in ../tests/error.ipynb:
===========================================================================

While Executing Cell #2:
Traceback (most recent call last):
  File "/var/folders/ss/34z569j921v58v8n1n_8z7h40000gn/T/ipykernel_32164/1421292703.py", line 3, in <module>
    s.execute('../tests/error.ipynb', exc_stop=True)
  File "/var/folders/ss/34z569j921v58v8n1n_8z7h40000gn/T/ipykernel_32164/3609882568.py", line 18, in execute
    self.run_all(nb, exc_stop=exc_stop, preproc=preproc, postproc=postproc,
  File "/var/folders/ss/34z569j921v58v8n1n_8z7h40000gn/T/ipykernel_32164/3068237356.py", line 19, in run_all
    if self.exc and exc_stop: raise self.exc from None
                              ^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/jhoward/uvws/.venv/lib/python3.12/site-packages/IPython/core/interactiveshell.py", line 3549, in run_code
    exec(code_obj, self.user_global_ns, self.user_ns)
  File "<ipython-input-1-b968a57a586e>", line 3, in <module>
    foo()
  File "/Users/jhoward/uvws/execnb/tests/err.py", line 2, in foo
    assert 13 == 98
           ^^^^^^^^
AssertionError

If you pass inject_code to CaptureShell.execute or CaptureShell.run_all, the source of nb.cells[inject_idx] will be replaced with inject_code. By default, the first cell is replaced. For instance consider this notebook:

nb = read_nb('../tests/params.ipynb')
for c in nb.cells: print('- ',c.source)

-  a=1
-  print(a)

We can replace the first cell with a=2 by passing that as inject_code, and the notebook will run with that change:

nb = read_nb('../tests/params.ipynb')
s.run_all(nb, inject_code="a=2")
list(nb.cells)

[{'cell_type': 'code',
  'execution_count': None,
  'id': 'a63ce885',
  'metadata': {},
  'outputs': [],
  'source': 'a=2',
  'idx_': 0},
 {'cell_type': 'code',
  'execution_count': None,
  'id': 'ea528db5',
  'metadata': {},
  'outputs': [{'name': 'stdout', 'output_type': 'stream', 'text': ['2\n']}],
  'source': 'print(a)',
  'idx_': 1}]

This can be used with CaptureShell.execute to parameterise runs of models in notebooks. Place any defaults for configuration code needed in the first cell, and then when running execute pass in new parameters as needed in inject_code. To replace only some of the defaults, leave an empty cell as the second cell, and inject code using inject_idx=1 to replace the empty second cell with code that overrides some of the defaults set in the first cell. When using execute you can pass inject_path instead of inject_code to read the injected code from a file.

---

source

exec_nb

 exec_nb (src:str, dest:str='', exc_stop:bool=False, inject_code:str=None,
          inject_path:str=None, inject_idx:int=0)

Execute notebook from src and save with outputs to dest

	Type	Default	Details
src	str		Notebook path to read from
dest	str		Notebook path to write to
exc_stop	bool	False	Stop on exceptions?
inject_code	str	None	Code to inject into a cell
inject_path	str	None	Path to file containing code to inject into a cell
inject_idx	int	0	Cell to replace with inject_code

This is the command-line version of CaptureShell.execute. Run exec_nb -h from the command line to see how to pass arguments. If you don’t pass dest then the output notebook won’t be saved; this is mainly useful for running tests.

---

source

SmartCompleter

 SmartCompleter (shell, namespace=None, jedi=False)

Extension of the completer class with IPython-specific features

cc = SmartCompleter(get_ipython())

def test_set(a,b): return test_eq(set(a), set(b))

class _f:
    def __init__(self): self.bar,self.baz,self.room = 0,0,0

foo = _f()

assert set(cc("b")).issuperset(['bool', 'bytes'])
test_set(cc("foo.b"), ['bar', 'baz'])
test_set(cc("x=1; x = foo.b"), ['bar', 'baz'])
test_set(cc("ab"), ['abs'])
test_set(cc("b = ab"), ['abs'])
test_set(cc(""), [])
test_set(cc("foo."), ['bar', 'baz', 'room'])
test_set(cc("nonexistent.b"), [])
test_set(cc("foo.nonexistent.b"), [])
assert set(cc("import ab")).issuperset(['abc'])
test_set(cc("from abc import AB"), ['ABC', 'ABCMeta'])

s = CaptureShell()
cc = SmartCompleter(s)
s.run('''def captures(pat, s, n, **kw):
    return 1''')
cc('captures(')

['n=', 'pat=', 's=']

---

source

CaptureShell.complete

 CaptureShell.complete (c)

Return the completed text and a list of completions.

	Type	Details
c
Returns	string	The actual text that was completed.

s = CaptureShell()
s.run('a=1')
s.complete('a.b')

['bit_count', 'bit_length']

s.run('import re')
s.complete('re.compile(')

['flags=', 'pattern=']