Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def error(self, message):
"""error messages to stderr"""
self.write(message)
def notice(self, message):
"""informative messages to stderr"""
self.write(message)
def info(self, message):
"""informative messages to stderr, only if verbose flag is set"""
#~ if self.verbosity > 0:
#~ self.write(message)
if __name__ == '__main__':
from ipykernel.kernelapp import IPKernelApp
IPKernelApp.launch_instance(kernel_class=MicroPythonKernel)
language_info = {
'name': 'scala',
'mimetype': 'text/x-scala',
'codemirror_mode': 'text/x-scala',
'pygments_lexer': 'scala'
}
session_language = LANG_SCALA
super(SparkKernel, self).__init__(implementation, implementation_version, language, language_version,
language_info, session_language, **kwargs)
if __name__ == '__main__':
from ipykernel.kernelapp import IPKernelApp
IPKernelApp.launch_instance(kernel_class=SparkKernel)
def do_apply(self, content, bufs, msg_id, reply_metadata):
pass
def do_clear(self):
pass
def __code_complete(self):
# TODO easy to hack, fix it
return not self.code_cache.find('explore') == -1
if __name__ == '__main__':
from ipykernel.kernelapp import IPKernelApp
IPKernelApp.launch_instance(kernel_class=CharmKernel)
import logging
import sys
from ipykernel.kernelapp import IPKernelApp
from . import WidgetsServerKernel
module = sys.argv[3]
object = sys.argv[4]
IPKernelApp.launch_instance(
kernel_class=WidgetsServerKernel,
log_level=logging.INFO,
user_ns=dict(exec_module=module, exec_object=object))
from .kernel import GeonotebookKernel
if __name__ == '__main__':
from ipykernel.kernelapp import IPKernelApp
IPKernelApp.launch_instance(kernel_class=GeonotebookKernel)
def __init__(self, *args, **kwargs):
super(EmbeddedQtKernel, self).__init__(*args, **kwargs)
self.eventloop = non_blocking_eventloop
def do_one_iteration(self):
with redirect_output(self.session, self.iopub_socket):
super(EmbeddedQtKernel, self).do_one_iteration()
def execute_request(self, stream, ident, parent):
with redirect_output(self.session, self.iopub_socket):
super(EmbeddedQtKernel, self).execute_request(
stream, ident, parent)
class EmbeddedQtKernelApp(IPKernelApp):
def init_kernel(self):
shell_stream = ZMQStream(self.shell_socket)
kernel = EmbeddedQtKernel(config=self.config, session=self.session,
shell_streams=[shell_stream],
iopub_socket=self.iopub_socket,
stdin_socket=self.stdin_socket,
log=self.log,
profile_dir=self.profile_dir,
)
self.kernel = kernel
kernel.record_ports(self.ports)
def start(self):
# handoff between IOLoop and QApplication event loops
loop = ioloop.IOLoop.instance()
if __name__ == "__main__":
from ipykernel.kernelapp import IPKernelApp
from ._kernel import MatlabKernel
IPKernelApp.launch_instance(kernel_class=MatlabKernel)
try:
km.shutdown_kernel(restart=restart)
except Exception as e:
self.warn(f'Failed to shutdown kernel {name}: {e}')
def __del__(self):
# upon releasing of sos kernel, kill all subkernels. This I thought would be
# called by the Jupyter cleanup code or the OS (because subkernels are subprocesses)
# but they are not.
self.do_shutdown(False)
if __name__ == '__main__':
from ipykernel.kernelapp import IPKernelApp
IPKernelApp.launch_instance(kernel_class=SoS_Kernel)