Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def in_notebook():
"""Check if this code is being executed in a notebook"""
if not has_ipynb_shell():
return False
from ipykernel.kernelapp import IPKernelApp
return IPKernelApp.initialized()
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])
def start(self):
if self._timer is not None:
raise Exception("IPython kernel is already running.")
# The IPKernelApp initialization is based on the IPython source for
# IPython.embed_kernel available here:
# https://github.com/ipython/ipython/blob/rel-3.2.1/IPython/kernel/zmq/embed.py
if IPKernelApp.initialized():
app = IPKernelApp.instance()
else:
# Load IPyIDA's user init file into the user namespace if it exists.
if os.path.exists(IPYIDARC_PATH):
IPKernelApp.exec_files = [ IPYIDARC_PATH ]
app = IPKernelApp.instance(
outstream_class='ipyida.kernel.IDATeeOutStream'
)
app.initialize()
main = app.kernel.shell._orig_sys_modules_main_mod
if main is not None:
sys.modules[app.kernel.shell._orig_sys_modules_main_name] = main
# IPython <= 3.2.x will send exception to sys.__stderr__ instead of
def bind_kernel(**kwargs):
"""Bind an Engine's Kernel to be used as a full IPython kernel.
This allows a running Engine to be used simultaneously as a full IPython kernel
with the QtConsole or other frontends.
This function returns immediately.
"""
from ipykernel.kernelapp import IPKernelApp
from ipyparallel.apps.ipengineapp import IPEngineApp
# first check for IPKernelApp, in which case this should be a no-op
# because there is already a bound kernel
if IPKernelApp.initialized() and isinstance(IPKernelApp._instance, IPKernelApp):
return
if IPEngineApp.initialized():
try:
app = IPEngineApp.instance()
except MultipleInstanceError:
pass
else:
return app.bind_kernel(**kwargs)
raise RuntimeError("bind_kernel be called from an IPEngineApp instance")
Parameters
----------
module : ModuleType, optional
The module to load into IPython globals (default: caller)
local_ns : dict, optional
The namespace to load into IPython user namespace (default: caller)
kwargs : various, optional
Further keyword args are relayed to the IPKernelApp constructor,
allowing configuration of the Kernel. Will only have an effect
on the first embed_kernel call for a given process.
"""
# get the app if it exists, or set it up if it doesn't
if IPKernelApp.initialized():
app = IPKernelApp.instance()
else:
app = IPKernelApp.instance(**kwargs)
app.initialize([])
# Undo unnecessary sys module mangling from init_sys_modules.
# This would not be necessary if we could prevent it
# in the first place by using a different InteractiveShell
# subclass, as in the regular embed case.
main = app.kernel.shell._orig_sys_modules_main_mod
if main is not None:
sys.modules[app.kernel.shell._orig_sys_modules_main_name] = main
# load the calling scope if not given
(caller_module, caller_locals) = extract_module_locals(1)
if module is None:
module = caller_module
def do_one_iteration():
"""Perform an iteration on IPython kernel runloop"""
if IPKernelApp.initialized():
app = IPKernelApp.instance()
app.kernel.do_one_iteration()
else:
raise Exception("Kernel is not initialized")
elif clname == 'ec2.Volume':
return f'{self.name} ({self.id} {self.state}): {self.size}GB'
elif clname == 'ec2.Snapshot':
return f'{self.name} ({self.id} {self.state}): {self.volume_size}GB'
elif clname == 'ec2.Image':
root_dev = [o for o in self.block_device_mappings if self.root_device_name == o['DeviceName']]
return f'{self.name} ({self.id} {self.state}): {root_dev[0]["Ebs"]["VolumeSize"]}GB'
else:
identifiers = [f'{ident}={repr(getattr(self, ident))}' for ident in self.meta.identifiers]
return f"{self.__class__.__name__}({', '.join(identifiers)})"
boto3.resources.base.ServiceResource.__repr__ = _boto3_repr
_in_notebook = False
try:
from ipykernel.kernelapp import IPKernelApp
_in_notebook = IPKernelApp.initialized()
except: pass
def listify(p=None, q=None):
"Make `p` listy and the same length as `q`."
if p is None: p=[]
elif isinstance(p, str): p=[p]
elif not isinstance(p, Iterable): p=[p]
n = q if type(q)==int else len(p) if q is None else len(q)
if len(p)==1: p = p * n
assert len(p)==n, f'List len mismatch ({len(p)} vs {n})'
return list(p)
def make_filter(d:Dict=None):
if d is None: d={}
d = {k.replace('_','-'):v for k,v in d.items()}
return {'Filters': [{'Name':k, 'Values':listify(v)} for k,v in (d or {}).items()]}