Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def cleanup(self):
self._app.heartbeat.context.destroy()
self._thread.join()
self._app.heartbeat.join()
self._app.iopub_thread.stop()
try:
self._app.kernel.shell.history_manager.save_thread.stop()
except AttributeError:
pass
zmq.Context.instance().destroy()
# successful if only the main thread remains
return len(threading.enumerate()) == 1
class CutterIPythonKernel(IPythonKernel):
def __init__(self, **kwargs):
super().__init__(**kwargs)
self.interruptable = False
def pre_handler_hook(self):
self.interruptable = True
def post_handler_hook(self):
self.interruptable = False
class CutterIPKernelApp(IPKernelApp):
def init_signal(self):
# This would call signal.signal(signal.SIGINT, signal.SIG_IGN)
# Not needed in for us.
pass
def _check_for_kernel(self):
try:
from ipykernel.kernelspec import RESOURCES, get_kernel_dict
from ipykernel.ipkernel import IPythonKernel
except ImportError:
return None
else:
return {
'spec': get_kernel_dict(),
'language_info': IPythonKernel.language_info,
'resource_dir': RESOURCES,
}
import os
import sys
# Third-party imports
from ipykernel.ipkernel import IPythonKernel
# Local imports
from spyder_kernels.comms.frontendcomm import FrontendComm
# Excluded variables from the Variable Explorer (i.e. they are not
# shown at all there)
EXCLUDED_NAMES = ['In', 'Out', 'exit', 'get_ipython', 'quit']
class SpyderKernel(IPythonKernel):
"""Spyder kernel for Jupyter."""
def __init__(self, *args, **kwargs):
super(SpyderKernel, self).__init__(*args, **kwargs)
self.frontend_comm = FrontendComm(self)
# All functions that can be called through the comm
handlers = {
'set_breakpoints': self.set_spyder_breakpoints,
'set_pdb_ignore_lib': self.set_pdb_ignore_lib,
'set_pdb_execute_events': self.set_pdb_execute_events,
'get_value': self.get_value,
'load_data': self.load_data,
'save_namespace': self.save_namespace,
'is_defined': self.is_defined,
aliases.update(base_aliases)
aliases.update(session_aliases)
flags = {
'mpi': ({
'EngineFactory': {'use_mpi': True},
}, "enable MPI integration"),
}
flags.update(base_flags)
flags.update(session_flags)
class IPEngineApp(BaseParallelApplication):
name = 'ipengine'
description = _description
examples = _examples
classes = List([ZMQInteractiveShell, ProfileDir, Session, EngineFactory, Kernel])
startup_script = Unicode(u'', config=True,
help='specify a script to be run at startup')
startup_command = Unicode('', config=True,
help='specify a command to be run at startup')
url_file = Unicode(u'', config=True,
help="""The full location of the file containing the connection information for
the controller. If this is not given, the file must be in the
security directory of the cluster directory. This location is
resolved using the `profile` or `profile_dir` options.""",
)
wait_for_url_file = Float(10, config=True,
help="""The maximum number of seconds to wait for url_file to exist.
This is useful for batch-systems and shared-filesystems where the
controller and engine are started at the same time and it
"""Version of run_cell that always uses shell_futures."""
return super(CoconutShell, self).run_cell(raw_cell, store_history, silent, shell_futures=True)
def user_expressions(self, expressions):
"""Version of user_expressions that compiles Coconut code first."""
compiled_expressions = {}
for key, expr in expressions.items():
try:
compiled_expressions[key] = COMPILER.parse_eval(expr)
except CoconutException:
compiled_expressions[key] = expr
return super(CoconutShell, self).user_expressions(compiled_expressions)
InteractiveShellABC.register(CoconutShell)
class CoconutKernel(IPythonKernel, object):
"""Jupyter kernel for Coconut."""
shell_class = CoconutShell
use_experimental_completions = True
implementation = "icoconut"
implementation_version = VERSION
language = "coconut"
language_version = VERSION
banner = version_banner
language_info = {
"name": "coconut",
"version": VERSION,
"mimetype": mimetype,
"codemirror_mode": {
"name": "python",
"version": py_syntax_version,
},
# Copyright (c) 2015 aggftw@gmail.com
# Distributed under the terms of the Modified BSD License.
import requests
from ipykernel.ipkernel import IPythonKernel
from hdijupyterutils.ipythondisplay import IpythonDisplay
import sparkmagic.utils.configuration as conf
from sparkmagic.utils.sparklogger import SparkLog
from sparkmagic.utils.constants import MAGICS_LOGGER_NAME
from sparkmagic.livyclientlib.exceptions import wrap_unexpected_exceptions
from sparkmagic.kernels.wrapperkernel.usercodeparser import UserCodeParser
class SparkKernelBase(IPythonKernel):
def __init__(self, implementation, implementation_version, language, language_version, language_info,
session_language, user_code_parser=None, **kwargs):
# Required by Jupyter - Override
self.implementation = implementation
self.implementation_version = implementation_version
self.language = language
self.language_version = language_version
self.language_info = language_info
# Override
self.session_language = session_language
super(SparkKernelBase, self).__init__(**kwargs)
self.logger = SparkLog(u"{}_jupyter_kernel".format(self.session_language))
self._fatal_error = None
"""IPython kernel for parallel computing"""
import sys
from ipython_genutils.py3compat import cast_bytes, unicode_type, safe_unicode, string_types
from traitlets import Integer, Type
from ipykernel.ipkernel import IPythonKernel
from ipyparallel.serialize import serialize_object, unpack_apply_message
from ipyparallel.util import utcnow
from .datapub import ZMQDataPublisher
class IPythonParallelKernel(IPythonKernel):
"""Extend IPython kernel for parallel computing"""
engine_id = Integer(-1)
msg_types = getattr(IPythonKernel, 'msg_types', []) + ['apply_request']
control_msg_types = getattr(IPythonKernel, 'control_msg_types', []) + ['abort_request', 'clear_request']
_execute_sleep = 0
data_pub_class = Type(ZMQDataPublisher)
def _topic(self, topic):
"""prefixed topic for IOPub messages"""
base = "engine.%s" % self.engine_id
return cast_bytes("%s.%s" % (base, topic))
def __init__(self, **kwargs):
super(IPythonParallelKernel, self).__init__(**kwargs)
# add apply_request, in anticipation of upstream deprecation
PY2 = sys.version[0] == '2'
# Check if we are running under an external interpreter
# We add "spyder" to sys.path for external interpreters,
# so relative imports work!
IS_EXT_INTERPRETER = os.environ.get('SPY_EXTERNAL_INTERPRETER') == "True"
# Excluded variables from the Variable Explorer (i.e. they are not
# shown at all there)
EXCLUDED_NAMES = ['In', 'Out', 'exit', 'get_ipython', 'quit']
# To be able to get and set variables between Python 2 and 3
PICKLE_PROTOCOL = 2
class SpyderKernel(IPythonKernel):
"""Spyder kernel for Jupyter"""
def __init__(self, *args, **kwargs):
super(SpyderKernel, self).__init__(*args, **kwargs)
self.namespace_view_settings = {}
self._pdb_obj = None
self._pdb_step = None
self._do_publish_pdb_state = True
self._mpl_backend_error = None
@property
def _pdb_frame(self):
"""Return current Pdb frame if there is any"""
if self._pdb_obj is not None and self._pdb_obj.curframe is not None:
nologger = False,
nologfile = False,
nogui = False,
prompt = 'NoColor',
trace = False,
pipeline = False,
agg = False,
ipython_log = False,
datapath = None,
crash_report = True,
telemetry = False,
execute = [])
__init_config(casa_config_master,flags,args)
class CasapyKernel(IPythonKernel):
implementation = 'Casapy'
implementation_version = '1.0'
language = 'casa'
language_version = '1.0'
language_info = {'mimetype': 'text/plain', 'name': 'Casa'}
banner = "Jupyter wrapper for casa"
def start(self):
super(CasapyKernel, self).start()
self.do_execute('%matplotlib inline', True, False, {}, False)
#self.do_execute('%matplotlib ipympl', True, False, {}, False)
for i in startup_scripts:
self.do_execute('%run -i {}'.format(i), True, False, {}, False)
wrappers = os.path.dirname(os.path.realpath(__file__)) + '/tasks_wrapped.py'
self.do_execute('%run -i {}'.format(wrappers), True, False, {}, False)
import casashell.private.config as config
try:
from ipykernel.kernelbase import Kernel
except ImportError:
from IPython.kernel.zmq.kernelbase import Kernel
from ipykernel.ipkernel import IPythonKernel
__version__ = '0.2'
try:
from traitlets import Unicode
except ImportError:
from IPython.utils.traitlets import Unicode
class MPKernelStmhal(IPythonKernel):
""" This subclasses the ipython kernel instead of
wrapping around the kernel base, since we only
need to alter the commands thats it runs
"""
# Required variables
implementation = 'mpkernel'
implementation_version = __version__
banner = 'MPKernelStmhal Banner'
language_info = {
'name': 'micropython',
'codemirror_mode': 'python',
'mimetype': 'text/x-python',
'file_extension': '.py'
}
def __init__(self, **kwargs):