PKIeq55bash_kernel/__init__.py"""A bash kernel for Jupyter""" __version__ = '0.5' PK.Q~Gbash_kernel/__main__.pyfrom ipykernel.kernelapp import IPKernelApp from .kernel import BashKernel IPKernelApp.launch_instance(kernel_class=BashKernel) PK.Q~G3Iaabash_kernel/images.pyimport base64 import imghdr import os #from IPython. _TEXT_SAVED_IMAGE = "bash_kernel: saved image data to:" image_setup_cmd = """ display () { TMPFILE=$(mktemp ${TMPDIR-/tmp}/bash_kernel.XXXXXXXXXX) cat > $TMPFILE echo "%s $TMPFILE" >&2 } """ % _TEXT_SAVED_IMAGE def display_data_for_image(filename): with open(filename, 'rb') as f: image = f.read() os.unlink(filename) image_type = imghdr.what(None, image) if image_type is None: raise ValueError("Not a valid image: %s" % image) image_data = base64.b64encode(image).decode('ascii') content = { 'data': { 'image/' + image_type: image_data }, 'metadata': {} } return content def extract_image_filenames(output): output_lines = [] image_filenames = [] for line in output.split("\n"): if line.startswith(_TEXT_SAVED_IMAGE): filename = line.rstrip().split(": ")[-1] image_filenames.append(filename) else: output_lines.append(line) output = "\n".join(output_lines) return image_filenames, output PKtI'[`EEbash_kernel/install.pyimport json import os import sys import getopt from jupyter_client.kernelspec import KernelSpecManager from IPython.utils.tempdir import TemporaryDirectory kernel_json = {"argv":[sys.executable,"-m","bash_kernel", "-f", "{connection_file}"], "display_name":"Bash", "language":"bash", "codemirror_mode":"shell", "env":{"PS1": "$"} } def install_my_kernel_spec(user=True, prefix=None): with TemporaryDirectory() as td: os.chmod(td, 0o755) # Starts off as 700, not user readable with open(os.path.join(td, 'kernel.json'), 'w') as f: json.dump(kernel_json, f, sort_keys=True) # TODO: Copy resources once they're specified print('Installing IPython kernel spec') KernelSpecManager().install_kernel_spec(td, 'bash', user=user, replace=True, prefix=prefix) def _is_root(): try: return os.geteuid() == 0 except AttributeError: return False # assume not an admin on non-Unix platforms def main(argv=[]): prefix = None user = not _is_root() opts, _ = getopt.getopt(argv[1:], '', ['user', 'prefix=']) for k, v in opts: if k == '--user': user = True elif k == '--prefix': prefix = v user = False install_my_kernel_spec(user=user, prefix=prefix) if __name__ == '__main__': main(argv=sys.argv) PKtI@ bash_kernel/kernel.pyfrom ipykernel.kernelbase import Kernel from pexpect import replwrap, EOF import pexpect from subprocess import check_output import os.path import base64 import imghdr import re import signal import urllib __version__ = '0.2' version_pat = re.compile(r'version (\d+(\.\d+)+)') from .images import ( extract_image_filenames, display_data_for_image, image_setup_cmd ) class IREPLWrapper(replwrap.REPLWrapper): """A subclass of REPLWrapper that gives incremental output specifically for bash_kernel. The parameters are the same as for REPLWrapper, except for one extra parameter: :param line_output_callback: a callback method to receive each batch of incremental output. It takes one string parameter. """ def __init__(self, cmd_or_spawn, orig_prompt, prompt_change, extra_init_cmd=None, line_output_callback=None): self.line_output_callback = line_output_callback replwrap.REPLWrapper.__init__(self, cmd_or_spawn, orig_prompt, prompt_change, extra_init_cmd=extra_init_cmd) def _expect_prompt(self, timeout=-1): if timeout == None: # "None" means we are executing code from a Jupyter cell by way of the run_command # in the do_execute() code below, so do incremental output. while True: pos = self.child.expect_exact([self.prompt, self.continuation_prompt, '\r\n'], timeout=None) if pos == 2: # End of line received self.line_output_callback(self.child.before + '\n') else: if len(self.child.before) != 0: # prompt received, but partial line precedes it self.line_output_callback(self.child.before) break else: # Otherwise, use existing non-incremental code pos = replwrap.REPLWrapper._expect_prompt(self, timeout=timeout) # Prompt received, so return normally return pos class BashKernel(Kernel): implementation = 'bash_kernel' implementation_version = __version__ @property def language_version(self): m = version_pat.search(self.banner) return m.group(1) _banner = None @property def banner(self): if self._banner is None: self._banner = check_output(['bash', '--version']).decode('utf-8') return self._banner language_info = {'name': 'bash', 'codemirror_mode': 'shell', 'mimetype': 'text/x-sh', 'file_extension': '.sh'} def __init__(self, **kwargs): Kernel.__init__(self, **kwargs) self._start_bash() def _start_bash(self): # Signal handlers are inherited by forked processes, and we can't easily # reset it from the subprocess. Since kernelapp ignores SIGINT except in # message handlers, we need to temporarily reset the SIGINT handler here # so that bash and its children are interruptible. sig = signal.signal(signal.SIGINT, signal.SIG_DFL) try: # Note: the next few lines mirror functionality in the # bash() function of pexpect/replwrap.py. Look at the # source code there for comments and context for # understanding the code here. bashrc = os.path.join(os.path.dirname(pexpect.__file__), 'bashrc.sh') child = pexpect.spawn("bash", ['--rcfile', bashrc], echo=False, encoding='utf-8') ps1 = replwrap.PEXPECT_PROMPT[:5] + u'\[\]' + replwrap.PEXPECT_PROMPT[5:] ps2 = replwrap.PEXPECT_CONTINUATION_PROMPT[:5] + u'\[\]' + replwrap.PEXPECT_CONTINUATION_PROMPT[5:] prompt_change = u"PS1='{0}' PS2='{1}' PROMPT_COMMAND=''".format(ps1, ps2) # Using IREPLWrapper to get incremental output self.bashwrapper = IREPLWrapper(child, u'\$', prompt_change, extra_init_cmd="export PAGER=cat", line_output_callback=self.process_output) finally: signal.signal(signal.SIGINT, sig) # Register Bash function to write image data to temporary file self.bashwrapper.run_command(image_setup_cmd) def process_output(self, output): if not self.silent: image_filenames, output = extract_image_filenames(output) # Send standard output stream_content = {'name': 'stdout', 'text': output} self.send_response(self.iopub_socket, 'stream', stream_content) # Send images, if any for filename in image_filenames: try: data = display_data_for_image(filename) except ValueError as e: message = {'name': 'stdout', 'text': str(e)} self.send_response(self.iopub_socket, 'stream', message) else: self.send_response(self.iopub_socket, 'display_data', data) def do_execute(self, code, silent, store_history=True, user_expressions=None, allow_stdin=False): self.silent = silent if not code.strip(): return {'status': 'ok', 'execution_count': self.execution_count, 'payload': [], 'user_expressions': {}} interrupted = False try: # Note: timeout=None tells IREPLWrapper to do incremental # output. Also note that the return value from # run_command is not needed, because the output was # already sent by IREPLWrapper. self.bashwrapper.run_command(code.rstrip(), timeout=None) except KeyboardInterrupt: self.bashwrapper.child.sendintr() interrupted = True self.bashwrapper._expect_prompt() output = self.bashwrapper.child.before self.process_output(output) except EOF: output = self.bashwrapper.child.before + 'Restarting Bash' self._start_bash() self.process_output(output) if interrupted: return {'status': 'abort', 'execution_count': self.execution_count} try: exitcode = int(self.bashwrapper.run_command('echo $?').rstrip()) except Exception: exitcode = 1 if exitcode: error_content = {'execution_count': self.execution_count, 'ename': '', 'evalue': str(exitcode), 'traceback': []} self.send_response(self.iopub_socket, 'error', error_content) error_content['status'] = 'error' return error_content else: return {'status': 'ok', 'execution_count': self.execution_count, 'payload': [], 'user_expressions': {}} def do_complete(self, code, cursor_pos): code = code[:cursor_pos] default = {'matches': [], 'cursor_start': 0, 'cursor_end': cursor_pos, 'metadata': dict(), 'status': 'ok'} if not code or code[-1] == ' ': return default tokens = code.replace(';', ' ').split() if not tokens: return default matches = [] token = tokens[-1] start = cursor_pos - len(token) if token[0] == '$': # complete variables cmd = 'compgen -A arrayvar -A export -A variable %s' % token[1:] # strip leading $ output = self.bashwrapper.run_command(cmd).rstrip() completions = set(output.split()) # append matches including leading $ matches.extend(['$'+c for c in completions]) else: # complete functions and builtins cmd = 'compgen -cdfa %s' % token output = self.bashwrapper.run_command(cmd).rstrip() matches.extend(output.split()) if not matches: return default matches = [m for m in matches if m.startswith(token)] return {'matches': sorted(matches), 'cursor_start': start, 'cursor_end': cursor_pos, 'metadata': dict(), 'status': 'ok'} PK!H|&Ubbash_kernel-0.5.dist-info/WHEEL HM K-*ϳR03rOK-J,/RH,Q034 /, (-JLR()*M ILR(4KM̫#DPK!Hw"bash_kernel-0.5.dist-info/METADATA}RKo@C@ׅł"M#l쉽d_]73k;Q"!rrf̷QT"' )-fBck-z[Z /`azֺ}DϾZ541Py-cӮyiuVǽ,bؤ,E33Bb<rݲ/J 7}i[( D0g7S8 V>̽IVKSý0uK&|.Kdʋ} ;>cnd8㰌{~ a[62dKN ҄(* ƀ~NCy^OiNdo X`7$9B(z9qm+cF[وP=mp44L9KdG)vKUtR##/GHS`g<1T$ Ћ1]5Jܣh:G+1|Z] L/ͰTU9̄}b/PK!HU \ bash_kernel-0.5.dist-info/RECORD}9s@>e5E &A YaP9_MfIvOFIuZjϠ)䏟ˀa.Ƶl sJM.h}dC$E ^h#kw_j"feuʍН P]pYڽ`ImusL>Pg8}]tk)=rpvxKzZ/\?AXn&YH 2p|E ^bĆ^?XӨa9 4kNF+"k < 򪭸NXׯX}֯;oi\]/YPٹgmh$U6t{/K%}3,!ݙ޶Gml 'AqoPKIeq55bash_kernel/__init__.pyPK.Q~Gjbash_kernel/__main__.pyPK.Q~G3Iaabash_kernel/images.pyPKtI'[`EEbash_kernel/install.pyPKtI@ , bash_kernel/kernel.pyPK!H|&Ub+bash_kernel-0.5.dist-info/WHEELPK!Hw",bash_kernel-0.5.dist-info/METADATAPK!HU \ .bash_kernel-0.5.dist-info/RECORDPK?0