##Update
Better version here: https://github.com/abalter/polyglottus/blob/master/simple_kernel.py
Just leaving this because it is linked to in a couple of SO posts
##Update
Better version here: https://github.com/abalter/polyglottus/blob/master/simple_kernel.py
Just leaving this because it is linked to in a couple of SO posts
| import queue | |
| from jupyter_client.manager import start_new_kernel | |
| from pprint import PrettyPrinter | |
| class SimpleKernel(): | |
| """ | |
| ## Description | |
| **SimpleKernel**: | |
| A simplistic Jupyter kernel client wrapper. | |
| Additional information in [this GitHub issue] | |
| ( | |
| ) | |
| """ | |
| def __init__(self): | |
| """ | |
| ## Description | |
| Initializes the `kernel_manager` and `client` objects | |
| and starts the kernel. Also initializes the pretty printer | |
| for displaying object properties and execution result | |
| payloads. | |
| ## Parameters | |
| None. | |
| """ | |
| ### Initialize kernel and client | |
| self.kernel_manager, self.client = start_new_kernel() | |
| ### Initialize pretty printer | |
| self.pp = PrettyPrinter(indent=2) | |
| ### end __init__ #### | |
| def execute(self, code, verbose=False, get_type=False): | |
| """ | |
| ## Description | |
| **execute**: | |
| Executes a code string in the kernel. Can return either | |
| the full execution response payload, or just `stdout`. Also, | |
| there is a verbose mode that displays the execution process. | |
| ## Parameters | |
| code : string | |
| The code string to get passed to `stdin`. | |
| verbose : bool (default=False) | |
| Whether to display processing information. | |
| get_type : bool (default=False) NOT IMPLEMENTED | |
| When implemented, will return a dict including the output | |
| and the type. E.g. | |
| 1+1 ==> {stdout: 2, type: int} | |
| "hello" ==> {stdout: "hello", type: str} | |
| print("hello") ==> {stdout: "hello", type: NoneType} | |
| a=10 ==> {stdout: None, type: None} | |
| ## Returns | |
| `stdout` or the full response payload. | |
| """ | |
| debug = lambda x: print(x if verbose else '', end='') | |
| debug("----------------") | |
| debug("executing code: " + code) | |
| ### Execute the code | |
| msg_id = self.client.execute(code) | |
| ### Collect the response payload | |
| reply = self.client.get_shell_msg(msg_id) | |
| debug("reply content") | |
| debug(reply['content']) | |
| ### Get the execution status | |
| ### When the execution state is "idle" it is complete | |
| io_msg_content = self.client.get_iopub_msg(timeout=0)['content'] | |
| ### We're going to catch this here before we start polling | |
| if 'execution_state' in io_msg_content and io_msg_content['execution_state'] == 'idle': | |
| return "no output" | |
| ### Continue polling for execution to complete | |
| ### which is indicated by having an execution state of "idle" | |
| while True: | |
| ### Save the last message content. This will hold the solution. | |
| ### The next one has the idle execution state indicating the execution | |
| ###is complete, but not the stdout output | |
| temp = io_msg_content | |
| ### Poll the message | |
| try: | |
| io_msg_content = self.client.get_iopub_msg(timeout=0)['content'] | |
| debug("io_msg content") | |
| debug(io_msg_content) | |
| if 'execution_state' in io_msg_content and io_msg_content['execution_state'] == 'idle': | |
| break | |
| except queue.Empty: | |
| debug("timeout get_iopub_msg") | |
| break | |
| debug("temp") | |
| debug(temp) | |
| ### Check the message for various possibilities | |
| if 'data' in temp: # Indicates completed operation | |
| debug('has data') | |
| out = temp['data']['text/plain'] | |
| elif 'name' in temp and temp['name'] == "stdout": # indicates output | |
| debug('name is stdout') | |
| out = temp['text'] | |
| debug("out is " + out) | |
| elif 'traceback' in temp: # Indicates error | |
| print("ERROR") | |
| out = '\n'.join(temp['traceback']) # Put error into nice format | |
| else: | |
| out = '' | |
| debug("----------------\n\n") | |
| debug("returning " + str(out)) | |
| return out | |
| ### end execute #### | |
| def showManager(self): | |
| """ | |
| ## Description | |
| **showManager**: | |
| Pretty Print kernel manager object. | |
| """ | |
| self.pp(self.kernel_manager) | |
| def showClient(self): | |
| """ | |
| ## Description | |
| **showClient**: | |
| Pretty Print client object. | |
| """ | |
| self.pp(self.client) | |
| def prettyPrint(self, payload): | |
| """ | |
| ## Description | |
| **prettyPrint**: | |
| A convenience method to pretty print the reply payload. | |
| ## example | |
| ``` | |
| >>> reply = my_kernel.execute("1+1") | |
| >>> my_kernel.prettyPrint(reply) | |
| ``` | |
| """ | |
| def __del__(self): | |
| """ | |
| ## Description | |
| Destructor. Shuts down kernel safely. | |
| """ | |
| self.kernel_manager.shutdown_kernel() | |
| ### end Simple Kernel ### | |
| def test(): | |
| kernel = SimpleKernel() | |
| commands = \ | |
| [ | |
| '1+1', | |
| 'a=5', | |
| 'b=0', | |
| 'b', | |
| 'print()', | |
| 'print("hello there")', | |
| '10', | |
| 'a*b', | |
| 'a', | |
| 'a+b', | |
| 'print(a+b)', | |
| 'print(a*10)', | |
| 'c=1/b' | |
| ] | |
| for command in commands: | |
| print(">>>" + command) | |
| out = kernel.execute(command, verbose=False) | |
| if out: print(out) | |
| if __name__=="__main__": | |
| test() |
balter@spectre:~/windocuments/polyglottus$ python simple_kernel.py
>>>1+1
2
>>>a=5
>>>b=0
>>>b
0
>>>print()
>>>print("hello there")
hello there
>>>10
10
>>>a*b
0
>>>a
5
>>>a+b
5
>>>print(a+b)
5
>>>print(a*10)
50
>>>c=1/b
ERROR
---------------------------------------------------------------------------
ZeroDivisionError Traceback (most recent call last)
<ipython-input-13-47a519732db5> in <module>()
----> 1 c=1/b
ZeroDivisionError: division by zero