I am putting his current mmt test kernel, including a yes/no clicker, here for future integration. We will be able to steal more from the mmt_jupyter_kernel soon, I think ;)
import requests
import getpass
import sys
import os
import json
from pexpect import replwrap
try:
from urllib.parse import quote
except ImportError:
from urllib import quote
from IPython.core import release
from ipython_genutils.py3compat import builtin_mod, PY3, unicode_type, safe_unicode
from IPython.utils.tokenutil import token_at_cursor, line_at_cursor
from traitlets import Instance, Type, Any, List, Bool
from ipykernel.kernelbase import Kernel
from ipykernel.comm import CommManager
import ipywidgets as widgets
from ipykernel.zmqshell import ZMQInteractiveShell
MMT_SERVER_EXTENSION = 'repl'
MMT_BASE_URL = os.environ.setdefault('MMT_BASE_URL', 'http://localhost:9000')
class TestKernel(Kernel):
widget_list = []
shell = Instance('IPython.core.interactiveshell.InteractiveShellABC',
allow_none=True)
shell_class = Type(ZMQInteractiveShell)
use_experimental_completions = Bool(True,
help="Set this flag to False to deactivate the use of experimental IPython completion APIs.",
).tag(config=True)
user_module = Any()
def _user_module_changed(self, name, old, new):
if self.shell is not None:
self.shell.user_module = new
user_ns = Instance(dict, args=None, allow_none=True)
def _user_ns_changed(self, name, old, new):
if self.shell is not None:
self.shell.user_ns = new
self.shell.init_user_ns()
implementation = 'TEST'
implementation_version = '1.2'
language = 'mmt-action-script'
language_version = '0.1'
language_info = {
'name': 'Any text',
'mimetype': 'text/plain',
'file_extension': '.txt',
}
startup = True
def __init__(self, **kwargs):
super(TestKernel,self).__init__(**kwargs)
self.mmtsession = requests.Session()
self.adapter = requests.adapters.HTTPAdapter()
self.mmtsession.mount('https://', self.adapter)
self.mmtsession.mount('http://', self.adapter)
self.headers = {'content-type' : 'application/json',
'content-encoding' : 'UTF-8'}
# Initialize the InteractiveShell subclass
self.shell = self.shell_class.instance(parent=self,
profile_dir = self.profile_dir,
user_module = self.user_module,
user_ns = self.user_ns,
kernel = self,
)
self.shell.displayhook.session = self.session
self.shell.displayhook.pub_socket = self.iopub_socket
self.shell.displayhook.topic = self._topic('execute_result')
self.shell.display_pub.session = self.session
self.shell.display_pub.pub_socket = self.iopub_socket
self.comm_manager = CommManager(parent=self, kernel=self)
self.shell.configurables.append(self.comm_manager)
comm_msg_types = [ 'comm_open', 'comm_msg', 'comm_close' ]
for msg_type in comm_msg_types:
self.shell_handlers[msg_type] = getattr(self.comm_manager, msg_type)
try:
response_dict = self.mmtsession.get(MMT_BASE_URL + '/:' + MMT_SERVER_EXTENSION+'?start',data = None,headers = self.headers, stream = True).json()
sessionheader = { 'X-REPL-Session' : response_dict['session'] }
self.headers = {**self.headers, **sessionheader}
except Exception:
pass
@property
def execution_count(self):
return self.shell.execution_count
@execution_count.setter
def execution_count(self, value):
# Ignore the incrememnting done by KernelBase, in favour of our shell's
# execution counter.
pass
# search_text = widgets.Text(description = 'Search')
# search_result = widgets.Select(description = 'Select table')
# def search_action(sender):
# phrase = search_text.value
# df = search(phrase) # A function that returns the results in a pandas df
# titles = df['title'].tolist()
# with search_result.hold_trait_notifications():
# search_result.options = titles
def request_prompt(self):
text = widgets.Text(description='Type here')
text.on_submit(self.send_request)
html = widgets.HTML(description='Response')
text.out = html
buttons = widgets.ToggleButtons(
options=['Yes', 'No'],
disabled=False,
button_style='danger', # 'success', 'info', 'warning', 'danger' or ''
# icons=['check'] * 3
)
buttons.out = html
buttons.observe(self.display_selection,'value')
display(text,html,buttons)
def display_selection(self,dict):
dict['owner'].out.value = dict['new']
def send_request(self,text):
response_dict = self.handle_request(text.value)
text.out.value = response_dict['message']
# self.send_response(self.iopub_socket, 'display_data',self.get_stream_content(response_dict['message']))
def do_execute(self, code, silent, store_history=True,user_expressions=None,allow_stdin=False):
if self.startup:
self.request_prompt()
self.startup = False
# self.shell.run_cell(code, store_history=store_history, silent=silent)
return {'status': 'ok',
# The base class increments the execution count
'payload' : [],
'execution_count': self.execution_count,
'user_expressions': {},
}
@property
def banner(self):
return self.shell.banner
def start(self):
self.shell.exit_now = False
super(TestKernel, self).start()
def set_parent(self, ident, parent):
"""Overridden from parent to tell the display hook and output streams
about the parent message.
"""
super(TestKernel, self).set_parent(ident, parent)
self.shell.set_parent(parent)
def init_metadata(self, parent):
"""Initialize metadata.
Run at the beginning of each execution request.
"""
md = super(TestKernel, self).init_metadata(parent)
# FIXME: remove deprecated ipyparallel-specific code
# This is required for ipyparallel < 5.0
md.update({
'dependencies_met' : True,
'engine' : self.ident,
})
return md
def get_stream_content(self,message):
return {
'data': {
'text/html': message
},
'metadata': {},
'transient': {},
}
"""handles the POST requests to the MMT-Server"""
def handle_request(self,code):
binary_data = code.encode('UTF-8')
return self.mmtsession.post(MMT_BASE_URL + '/:' + MMT_SERVER_EXTENSION,data = binary_data,headers = self.headers, stream = True).json()
if __name__ == '__main__':
from ipykernel.kernelapp import IPKernelApp
IPKernelApp.launch_instance(kernel_class=TestKernel)