phc/magma_kernel
6
1
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Major rewrite

Browse Source
This commit is contained in:
Nils Bruin
2018-05-05 10:27:32 -07:00
parent ce85257bf7
commit 31082f3de2
10 changed files with 206 additions and 160 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+1
View File
@@ -2,4 +2,5 @@ __pycache__
*.pyc
build/
dist/
magma_kernel.egg-info/
MANIFEST
LICENSE
+3 -1
View File
@@ -1,4 +1,6 @@
Copyright (c) 2015, Thomas Kluyver and contributors
Copyright (c) 2015, Thomas Kluyver and contributors
Copyright (c) 2018, Nils Bruin and contributors
All rights reserved.
BSD 3-clause license:
MANIFEST.in
+1
View File
@@ -0,0 +1 @@
include README.rst
README.rst
+13 -11
View File
@@ -1,22 +1,24 @@
A simple IPython kernel for bash
A simple Jupyter kernel for the Magma computer algebra system
This requires IPython 3.
To install::
pip install bash_kernel
python -m bash_kernel.install
pip install magma_kernel
python -m magma_kernel.install
To use it, run one of:
.. code:: shell
ipython notebook
# In the notebook interface, select Bash from the 'New' menu
ipython qtconsole --kernel bash
ipython console --kernel bash
jupyter notebook
# In the notebook interface, select Magma from the 'New' menu
jupyter qtconsole --kernel magma
jupyter console --kernel magma
For details of how this works, see the Jupyter docs on `wrapper kernels
<http://jupyter-client.readthedocs.org/en/latest/wrapperkernels.html>`_, and
Pexpect's docs on the `replwrap module
<http://pexpect.readthedocs.org/en/latest/api/replwrap.html>`_
This code is based on a Magma kernel for IPython written by Christopher
Granade, which was in turn based on the Bash example kernel by Thomas
Kluyver. Improvements made in the current version include Tab
completion, processing of help requests by returning an appropriate
help query URL for Magma online documentation, and the reporting of
partial output.
flit.ini
-11
View File
@@ -1,11 +0,0 @@
[metadata]
module = magma_kernel
author = Chris Granade
author-email = cgranade@cgranade.com
home-page = https://github.com/takluyver/magma_kernel
requires = pexpect (>=3.3)
description-file = README.rst
classifiers = Framework :: IPython
License :: OSI Approved :: BSD License
Programming Language :: Python :: 3
magma_kernel/__init__.py
+1 -1
View File
@@ -1,3 +1,3 @@
"""A magma kernel for Jupyter"""
__version__ = '0.0.1dev1'
from .kernel import __version__
magma_kernel/images.py
-48
View File
@@ -1,48 +0,0 @@
import 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
magma_kernel/install.py
+41 -8
View File
@@ -1,18 +1,19 @@
import json
import os
import sys
import argparse
from jupyter_client.kernelspec import install_kernel_spec
from jupyter_client.kernelspec import KernelSpecManager
from IPython.utils.tempdir import TemporaryDirectory
kernel_json = {"argv":[sys.executable,"-m","magma_kernel", "-f", "{connection_file}"],
kernel_json = {"argv":[sys.executable,"-m","bash_kernel", "-f", "{connection_file}"],
"display_name":"Magma",
"language":"magma",
"codemirror_mode":"magma",
"env":{"PS1": "$"}
}
def install_my_kernel_spec(user=True):
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:
@@ -20,7 +21,7 @@ def install_my_kernel_spec(user=True):
# TODO: Copy resources once they're specified
print('Installing IPython kernel spec')
install_kernel_spec(td, 'magma', user=user, replace=True)
KernelSpecManager().install_kernel_spec(td, 'magma', user=user, replace=True, prefix=prefix)
def _is_root():
try:
@@ -28,9 +29,41 @@ def _is_root():
except AttributeError:
return False # assume not an admin on non-Unix platforms
def main(argv=[]):
user = '--user' in argv or not _is_root()
install_my_kernel_spec(user=user)
def main(argv=None):
parser = argparse.ArgumentParser(
description='Install KernelSpec for Bash Kernel'
)
prefix_locations = parser.add_mutually_exclusive_group()
prefix_locations.add_argument(
'--user',
help='Install KernelSpec in user homedirectory',
action='store_true'
)
prefix_locations.add_argument(
'--sys-prefix',
help='Install KernelSpec in sys.prefix. Useful in conda / virtualenv',
action='store_true',
dest='sys_prefix'
)
prefix_locations.add_argument(
'--prefix',
help='Install KernelSpec in this prefix',
default=None
)
args = parser.parse_args(argv)
user = False
prefix = None
if args.sys_prefix:
prefix = sys.prefix
elif args.prefix:
prefix = args.prefix
elif args.user or not _is_root():
user = True
install_my_kernel_spec(user=user, prefix=prefix)
if __name__ == '__main__':
main(argv=sys.argv)
main()
magma_kernel/kernel.py
+121 -80
View File
@@ -1,129 +1,170 @@
from ipykernel.kernelbase import Kernel
from pexpect import replwrap, EOF, spawn
from pexpect import TIMEOUT, EOF, spawn
from subprocess import Popen, PIPE
from pathlib import Path
import jupyter_core.paths
from subprocess import check_output
from os import unlink
import base64
import imghdr
import re
import signal
import urllib
__version__ = '0.0.1dev1'
version_pat = re.compile(r'version (\d+(\.\d+)+)')
from .images import (
extract_image_filenames, display_data_for_image, image_setup_cmd
)
import re
import glob
__version__ = '0.1a1'
class MagmaKernel(Kernel):
implementation = 'magma_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
'''
@property
def banner(self):
return ""
language_info = {'name': 'magma',
'codemirror_mode': 'magma',
'mimetype': 'text/x-sh',
'file_extension': '.mgm'}
'codemirror_mode': 'python',
'mimetype': 'text/x-magma',
'file_extension': '.m'}
def __init__(self, **kwargs):
Kernel.__init__(self, **kwargs)
self._prompt="$PEXPECT_PROMPT$"
self._start_magma()
self.completions = self._fetch_completions()
def _fetch_completions(self):
P=Path(jupyter_core.paths.jupyter_data_dir())
P=P.joinpath('kernels')
if not P.exists():
P.mkdir()
P=P.joinpath('magma')
if not P.exists():
P.mkdir()
P=P.joinpath('magma-completions.'+self.language_info['version'])
if P.exists():
with P.open('r') as F:
completions = F.read().split()
else:
child=Popen("magma",stdin=PIPE,stdout=PIPE)
child.stdin.write("ListSignatures(Any);".encode())
child.stdin.close();
result=child.stdout.read().decode().split()
completions = sorted({r[:r.index('(')] for r in result if '(' in r})
with P.open('w') as F:
F.write('\n'.join(completions))
return completions
def _start_magma(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.
# so that magma and its children are interruptible.
sig = signal.signal(signal.SIGINT, signal.SIG_DFL)
try:
magma = spawn('magma', echo=False, encoding='utf-8')
magma.expect('> ')
magma.expect_exact('> ')
banner=magma.before
magma.sendline('SetLineEditor(false);')
magma.expect('> ')
magma.sendline('')
self.magmawrapper = replwrap.REPLWrapper(magma, '> ', 'SetPrompt("{}");')
magma.expect_exact('> ')
magma.sendline('SetColumns(0);')
magma.expect_exact('> ')
magma.sendline('SetPrompt("{}");'.format(self._prompt));
magma.expect_exact(self._prompt)
self.child = magma
finally:
signal.signal(signal.SIGINT, sig)
lang_version = re.search("Magma V(\d*.\d*-\d*)",banner).group(1)
self.banner = "Magma kernel connected to Magma "+lang_version
self.language_info['version']=lang_version
self.language_version=lang_version
# Register Bash function to write image data to temporary file
# self.magmawrapper.run_command(image_setup_cmd)
def do_help(self, keyword):
URL="http://magma.maths.usyd.edu.au/magma/handbook/search?chapters=1&examples=1&intrinsics=1&query="+keyword
content = {
'source': 'stdout',
'data': {
# 'text/html':'<iframe src="{}" title="iframe">help</iframe>'.format(URL)
'text/html':'<a href="{}" target="magma_help">Magma help on {}</a>'.format(URL,keyword)
}
}
self.send_response(self.iopub_socket, 'display_data', content)
# we could look locally for help, but since these would be "file" URLs
# we cannot redirect to them anyway.
# pattern = re.compile('.*(<A HREF = ".*htm.*".*{}.*</A>)'.format(keyword))
# results = set()
# for name in glob.glob('/usr/local/magma/2.23-9/doc/html/ind*.htm'):
# with open(name,'r') as f:
# for r in f.readlines():
# match=pattern.match(r)
# if match:
# line = '<A target="magma_help" '+match.group(1)[2:]
# results.add(line)
# content['data']['text/html']="".join(sorted(results))
# self.send_response(self.iopub_socket, 'display_data', content)
def do_execute(self, code, silent, store_history=True,
user_expressions=None, allow_stdin=False):
if not code.strip():
return {'status': 'ok', 'execution_count': self.execution_count,
'payload': [], 'user_expressions': {}}
if code[0] == '?':
self.do_help(code[1:])
return {'status': 'ok', 'execution_count': self.execution_count,
'payload': [], 'user_expressions': {}}
interrupted = False
cmdlines = code.splitlines()
C=self.child
try:
output = self.magmawrapper.run_command(code.rstrip(), timeout=None)
for line in cmdlines:
C.sendline(line)
j = 0
#We use a fairly short timeout intercept and send back
#updates on what is received on stdout, if there is any.
counter=10
timeout=1
while True:
v=C.expect_exact([self._prompt, TIMEOUT],timeout=timeout)
if not silent and len(C.before) > j:
stream_content = {'name': 'stdout', 'text': C.before[j:]}
self.send_response(self.iopub_socket, 'stream', stream_content)
j=len(C.before)
if v==0:
break
counter-=1
if counter<=0:
timeout=min(300,2*counter)
counter=10
output=C.before[j:]
except KeyboardInterrupt:
self.magmawrapper.child.sendintr()
C.sendintr()
interrupted = True
self.magmawrapper._expect_prompt()
output = self.magmawrapper.child.before
C.expect_exact(self._prompt)
output = C.before[j:]
except EOF:
output = self.magmawrapper.child.before + 'Restarting Bash'
output = C.before[j:] + 'Restarting Magma'
self._start_magma()
if not 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)
'''
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):
cursor_end = cursor_pos
cursor_start = cursor_end
while cursor_start>0 and code[cursor_start-1] in "abcdefghijklmnopqrstyuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ_":
cursor_start -= 1
if cursor_start<cursor_end:
fragment = code[cursor_start:cursor_end]
length = len(fragment)
matches = [C for C in self.completions if fragment == C[:length]]
else:
matches = []
return {'status': 'ok',
'matches' : matches,
'cursor_start' : cursor_start,
'cursor_end' : cursor_end,
'metadata' : {}}
setup.py
+25
View File
@@ -0,0 +1,25 @@
from setuptools import setup, find_packages
def readme():
with open('README.rst') as f:
return f.read()
setup(
name="magma_kernel",
version="0.1a1",
author="Nils Bruin",
author_email="nbruin@sfu.ca",
description="A Jupyter kernel for the Magma computer algebra system",
long_description=readme(),
keywords="jupyter kernel magma",
license="BSD",
url="https://github.com/nbruin/magma_kernel",
install_requires="pexpect>=3.3",
packages=find_packages(),
include_package_data=True,
classifiers=[
"Framework :: IPython",
"License :: OSI Approved :: BSD License",
"Programming Language :: Python :: 3",
]
)