Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
global logger
logger = log.kernel_logger.getChild(str(os.getpid()))
logger.debug("forked kernel is running")
log.std_redirect(logger)
# Become a group leader for cleaner exit.
os.setpgrp()
dir = os.path.join(self.dir, self.id)
try:
os.mkdir(dir)
except OSError as e:
if e.errno != errno.EEXIST:
raise
os.chdir(dir)
#config = traitlets.config.loader.Config({"ip": self.ip})
#config.HistoryManager.enabled = False
app = IPKernelApp.instance(log=logger)
from namespace import InstrumentedNamespace
app.user_ns = InstrumentedNamespace()
app.initialize([]) # Redirects stdout/stderr
#log.std_redirect(logger) # Uncomment for debugging
# This function should be called via atexit, but it isn't, perhaps due
# to forking. Stale connection files do cause problems.
app.cleanup_connection_file()
kernel_init.initialize(app.kernel)
for r, limit in self.rlimits.iteritems():
resource.setrlimit(getattr(resource, r), (limit, limit))
logger.debug("kernel ready")
context = zmq.Context.instance()
socket = context.socket(zmq.PUSH)
socket.connect("tcp://localhost:{}".format(self.waiter_port))
socket.send_json({
"id": self.id,
def test(self):
from ipykernel.connect import connect_qtconsole
from ipykernel.kernelapp import IPKernelApp
kernelApp = IPKernelApp.instance()
args = ['python', '--pylab=qt', '--log-level=20']
kernelApp.initialize(args)
connect_qtconsole()
#@-others
MW.show()
# Support for IPython Jupyter QT Console as a debugging aid.
# Last argument must be --console to enable it
# Code reference :
# https://github.com/ipython/ipykernel/blob/master/examples/embedding/ipkernel_qtapp.py
# https://github.com/ipython/ipykernel/blob/master/examples/embedding/internal_ipkernel.py
if len(sys.argv) > 1 and sys.argv[-1] == "--console":
try:
from IPython.lib.kernel import connect_qtconsole
from ipykernel.kernelapp import IPKernelApp
# Only to ensure matplotlib QT mainloop integration is available
import matplotlib
# Create IPython kernel within our application
kernel = IPKernelApp.instance()
# Initialize it and use matplotlib for main event loop integration with QT
kernel.initialize(['python', '--matplotlib=qt'])
# Create the console in a new process and connect
console = connect_qtconsole(kernel.abs_connection_file, profile=kernel.profile)
# Export MW and app variable to the console's namespace
kernel.shell.user_ns['MW'] = MW
kernel.shell.user_ns['app'] = app
kernel.shell.user_ns['kernel'] = kernel
kernel.shell.user_ns['console'] = console
# When we close manuskript, make sure we close the console process and stop the
# IPython kernel's mainloop, otherwise the app will never finish.
def console_cleanup():
if get_ipython() is not None:
raise RuntimeError("Cannot start IPython, it's already running.")
from zmq.eventloop.ioloop import ZMQIOLoop
from ipykernel.kernelapp import IPKernelApp
# save the global IOLoop instance
# since IPython relies on it, but we are going to put it in a thread.
save_inst = IOLoop.instance()
IOLoop.clear_instance()
zmq_loop = ZMQIOLoop()
zmq_loop.install()
# start IPython, disabling its signal handlers that won't work due to running in a thread:
app = IPKernelApp.instance(log=log)
# Don't connect to the history database
app.config.HistoryManager.hist_file = ":memory:"
# listen on all interfaces, so remote clients can connect:
if ip:
app.ip = ip
# disable some signal handling, logging
def noop():
return None
app.init_signal = noop
app.log_connection_info = noop
# start IPython in a thread
# initialization happens in the thread to avoid threading problems
# with the sqlite history
# Add current directory to sys.path (like for any standard Python interpreter
# executed in interactive mode):
sys.path.insert(0, '')
# Fire up the kernel instance.
from ipykernel.kernelapp import IPKernelApp
if not IS_EXT_INTERPRETER:
from spyder.utils.ipython.spyder_kernel import SpyderKernel
else:
# We add "spyder" to sys.path for external interpreters,
# so this works!
# See create_kernel_spec of plugins/ipythonconsole
from utils.ipython.spyder_kernel import SpyderKernel
kernel = IPKernelApp.instance()
kernel.kernel_class = SpyderKernel
try:
kernel.config = kernel_config()
except:
pass
kernel.initialize()
# Set our own magics
kernel.shell.register_magic_function(varexp)
# Set Pdb class to be used by %debug and %pdb.
# This makes IPython consoles to use the class defined in our
# sitecustomize instead of their default one.
import pdb
kernel.shell.InteractiveTB.debugger_cls = pdb.Pdb
def kernel(run):
from ipykernel import kernelapp
from tornado import ioloop
io_loop = ioloop.IOLoop.current()
io_loop.start = lambda: None # without run_forever
kernelapp._ctrl_c_message = 'IPKernelApp running'
app = kernelapp.IPKernelApp.instance()
app.initialize(['aioworkers'])
namespace = {}
app.kernel.user_ns = namespace
class PseudoFuture:
def set_result(self, value):
namespace['context'] = value
app.start()
run(future=PseudoFuture())
def get_connection_file(app=None):
"""Return the path to the connection file of an app
Parameters
----------
app : IPKernelApp instance [optional]
If unspecified, the currently running app will be used
"""
if app is None:
from ipykernel.kernelapp import IPKernelApp
if not IPKernelApp.initialized():
raise RuntimeError("app not specified, and not in a running Kernel")
app = IPKernelApp.instance()
return filefind(app.connection_file, ['.', app.connection_dir])