/usr/lib/python3/dist-packages/jupyter_client/kernelapp.py is in python3-jupyter-client 5.2.2-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 | import os
import signal
import uuid
from jupyter_core.application import JupyterApp, base_flags
from tornado.ioloop import IOLoop
from traitlets import Unicode
from . import __version__
from .kernelspec import KernelSpecManager, NATIVE_KERNEL_NAME
from .manager import KernelManager
class KernelApp(JupyterApp):
"""Launch a kernel by name in a local subprocess.
"""
version = __version__
description = "Run a kernel locally in a subprocess"
classes = [KernelManager, KernelSpecManager]
aliases = {
'kernel': 'KernelApp.kernel_name',
'ip': 'KernelManager.ip',
}
flags = {'debug': base_flags['debug']}
kernel_name = Unicode(NATIVE_KERNEL_NAME,
help = 'The name of a kernel type to start'
).tag(config=True)
def initialize(self, argv=None):
super(KernelApp, self).initialize(argv)
self.km = KernelManager(kernel_name=self.kernel_name,
config=self.config)
cf_basename = 'kernel-%s.json' % uuid.uuid4()
self.km.connection_file = os.path.join(self.runtime_dir, cf_basename)
self.loop = IOLoop.current()
self.loop.add_callback(self._record_started)
def setup_signals(self):
"""Shutdown on SIGTERM or SIGINT (Ctrl-C)"""
if os.name == 'nt':
return
def shutdown_handler(signo, frame):
self.loop.add_callback_from_signal(self.shutdown, signo)
for sig in [signal.SIGTERM, signal.SIGINT]:
signal.signal(sig, shutdown_handler)
def shutdown(self, signo):
self.log.info('Shutting down on signal %d' % signo)
self.km.shutdown_kernel()
self.loop.stop()
def log_connection_info(self):
cf = self.km.connection_file
self.log.info('Connection file: %s', cf)
self.log.info("To connect a client: --existing %s", os.path.basename(cf))
def _record_started(self):
"""For tests, create a file to indicate that we've started
Do not rely on this except in our own tests!
"""
fn = os.environ.get('JUPYTER_CLIENT_TEST_RECORD_STARTUP_PRIVATE')
if fn is not None:
with open(fn, 'wb'):
pass
def start(self):
self.log.info('Starting kernel %r', self.kernel_name)
try:
self.km.start_kernel()
self.log_connection_info()
self.setup_signals()
self.loop.start()
finally:
self.km.cleanup()
main = KernelApp.launch_instance
|