tkw1536
commented
6 years ago This is great news.
We should put this on a seperate branch and clean it up.
On May 14, 2018 1:05:14 PM GMT+02:00, Freifrau von Bleifrei notifications@github.com wrote:
Good news: by tearing apart the ipykernel, @kaiamann has managed to get widgets in a custom kernel to work.
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)
Good news: by tearing apart the ipykernel, @kaiamann has managed to get widgets in a custom kernel to work.
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 ;)