1# Copyright (c) 2016, Sean Vig 2# 3# Permission is hereby granted, free of charge, to any person obtaining a copy 4# of this software and associated documentation files (the "Software"), to deal 5# in the Software without restriction, including without limitation the rights 6# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell 7# copies of the Software, and to permit persons to whom the Software is 8# furnished to do so, subject to the following conditions: 9# 10# The above copyright notice and this permission notice shall be included in 11# all copies or substantial portions of the Software. 12# 13# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR 14# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, 15# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE 16# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER 17# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, 18# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE 19# SOFTWARE. 20 21from ipykernel.kernelbase import Kernel 22 23from libqtile import command, ipc, sh 24 25 26class QshKernel(Kernel): 27 implementation = 'qshell' 28 implementation_version = '0.1' 29 language = 'no-op' 30 language_version = '1.0' 31 language_info = {'mimetype': 'text/plain'} 32 banner = "Qsh Kernel" 33 34 def __init__(self, *args, **kwargs): 35 super().__init__(*args, **kwargs) 36 socket_path = ipc.find_sockfile() 37 ipc_client = ipc.Client(socket_path) 38 cmd_object = command.interface.IPCCommandInterface(ipc_client) 39 self.qsh = sh.QSh(cmd_object) 40 41 def do_execute(self, code, silent, _store_history=True, _user_expressions=None, _allow_stdin=False): 42 # if no command sent, just return 43 if not code.strip(): 44 return { 45 'status': 'ok', 46 'execution_count': self.execution_count, 47 'payload': [], 48 'user_expressions': {}, 49 } 50 51 if code[-1] == '?': 52 return self.do_inspect(code, len(code) - 1) 53 54 try: 55 output = self.qsh.process_line(code) 56 except KeyboardInterrupt: 57 return { 58 'status': 'abort', 59 'execution_count': self.execution_count, 60 } 61 62 if not silent and output: 63 stream_content = {'name': 'stdout', 'text': output} 64 self.send_response(self.iopub_socket, 'stream', stream_content) 65 66 return { 67 'status': 'ok', 68 'execution_count': self.execution_count, 69 'payload': [], 70 'user_expressions': {}, 71 } 72 73 def do_complete(self, code, cursor_pos): 74 no_complete = { 75 'status': 'ok', 76 'matches': [], 77 'cursor_start': 0, 78 'cursor_end': cursor_pos, 79 'metadata': dict(), 80 } 81 82 if not code or code[-1] == ' ': 83 return no_complete 84 85 tokens = code.split() 86 if not tokens: 87 return no_complete 88 89 token = tokens[-1] 90 start = cursor_pos - len(token) 91 92 matches = self.qsh._complete(code, token) 93 return { 94 'status': 'ok', 95 'matches': sorted(matches), 96 'cursor_start': start, 97 'cursor_end': cursor_pos, 98 'metadata': dict(), 99 } 100 101 102def main(): 103 from ipykernel.kernelapp import IPKernelApp 104 IPKernelApp.launch_instance(kernel_class=QshKernel) 105 106 107if __name__ == '__main__': 108 main() 109