mathhubinfo / mosis_jupyter_kernel Goto Github PK

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)

  File "/MoSIS_Jupyter_Kernel/interview_kernel/__init__.py", line 1, in <module>
    from . import string_handling, mmtinterface, exaoutput, pde_state_machine
  File "/MoSIS_Jupyter_Kernel/interview_kernel/pde_state_machine.py", line 8, in <module>
    from .widget_factory import WidgetFactory
ModuleNotFoundError: No module named 'interview_kernel.widget_factory'