1 """A client for in-process kernels.""" 2 3 # ----------------------------------------------------------------------------- 4 # Copyright (C) 2012 The IPython Development Team 5 # 6 # Distributed under the terms of the BSD License. The full license is in 7 # the file LICENSE, distributed as part of this software. 8 # ----------------------------------------------------------------------------- 9 10 # ----------------------------------------------------------------------------- 11 # Imports 12 # ----------------------------------------------------------------------------- 13 14 import asyncio 15 16 from jupyter_client.client import KernelClient 17 from jupyter_client.clientabc import KernelClientABC 18 from jupyter_core.utils import run_sync 19 20 # IPython imports 21 from traitlets import Instance, Type, default 22 23 # Local imports 24 from .channels import InProcessChannel, InProcessHBChannel 25 26 # ----------------------------------------------------------------------------- 27 # Main kernel Client class 28 # ----------------------------------------------------------------------------- 29 30 31 class InProcessKernelClient(KernelClient): 32 """A client for an in-process kernel. 33 34 This class implements the interface of 35 `jupyter_client.clientabc.KernelClientABC` and allows 36 (asynchronous) frontends to be used seamlessly with an in-process kernel. 37 38 See `jupyter_client.client.KernelClient` for docstrings. 39 """ 40 41 # The classes to use for the various channels. 42 shell_channel_class = Type(InProcessChannel) # type:ignore[arg-type] 43 iopub_channel_class = Type(InProcessChannel) # type:ignore[arg-type] 44 stdin_channel_class = Type(InProcessChannel) # type:ignore[arg-type] 45 control_channel_class = Type(InProcessChannel) # type:ignore[arg-type] 46 hb_channel_class = Type(InProcessHBChannel) # type:ignore[arg-type] 47 48 kernel = Instance("ipykernel.inprocess.ipkernel.InProcessKernel", allow_none=True) 49 50 # -------------------------------------------------------------------------- 51 # Channel management methods 52 # -------------------------------------------------------------------------- 53 54 @default("blocking_class") 55 def _default_blocking_class(self): 56 from .blocking import BlockingInProcessKernelClient 57 58 return BlockingInProcessKernelClient 59 60 def get_connection_info(self): 61 """Get the connection info for the client.""" 62 d = super().get_connection_info() 63 d["kernel"] = self.kernel # type:ignore[assignment] 64 return d 65 66 def start_channels(self, *args, **kwargs): 67 """Start the channels on the client.""" 68 super().start_channels() 69 if self.kernel: 70 self.kernel.frontends.append(self) 71 72 @property 73 def shell_channel(self): 74 if self._shell_channel is None: 75 self._shell_channel = self.shell_channel_class(self) # type:ignore[abstract,call-arg] 76 return self._shell_channel 77 78 @property 79 def iopub_channel(self): 80 if self._iopub_channel is None: 81 self._iopub_channel = self.iopub_channel_class(self) # type:ignore[abstract,call-arg] 82 return self._iopub_channel 83 84 @property 85 def stdin_channel(self): 86 if self._stdin_channel is None: 87 self._stdin_channel = self.stdin_channel_class(self) # type:ignore[abstract,call-arg] 88 return self._stdin_channel 89 90 @property 91 def control_channel(self): 92 if self._control_channel is None: 93 self._control_channel = self.control_channel_class(self) # type:ignore[abstract,call-arg] 94 return self._control_channel 95 96 @property 97 def hb_channel(self): 98 if self._hb_channel is None: 99 self._hb_channel = self.hb_channel_class(self) # type:ignore[abstract,call-arg] 100 return self._hb_channel 101 102 # Methods for sending specific messages 103 # ------------------------------------- 104 105 def execute( 106 self, code, silent=False, store_history=True, user_expressions=None, allow_stdin=None 107 ): 108 """Execute code on the client.""" 109 if allow_stdin is None: 110 allow_stdin = self.allow_stdin 111 content = dict( 112 code=code, 113 silent=silent, 114 store_history=store_history, 115 user_expressions=user_expressions or {}, 116 allow_stdin=allow_stdin, 117 ) 118 msg = self.session.msg("execute_request", content) 119 self._dispatch_to_kernel(msg) 120 return msg["header"]["msg_id"] 121 122 def complete(self, code, cursor_pos=None): 123 """Get code completion.""" 124 if cursor_pos is None: 125 cursor_pos = len(code) 126 content = dict(code=code, cursor_pos=cursor_pos) 127 msg = self.session.msg("complete_request", content) 128 self._dispatch_to_kernel(msg) 129 return msg["header"]["msg_id"] 130 131 def inspect(self, code, cursor_pos=None, detail_level=0): 132 """Get code inspection.""" 133 if cursor_pos is None: 134 cursor_pos = len(code) 135 content = dict( 136 code=code, 137 cursor_pos=cursor_pos, 138 detail_level=detail_level, 139 ) 140 msg = self.session.msg("inspect_request", content) 141 self._dispatch_to_kernel(msg) 142 return msg["header"]["msg_id"] 143 144 def history(self, raw=True, output=False, hist_access_type="range", **kwds): 145 """Get code history.""" 146 content = dict(raw=raw, output=output, hist_access_type=hist_access_type, **kwds) 147 msg = self.session.msg("history_request", content) 148 self._dispatch_to_kernel(msg) 149 return msg["header"]["msg_id"] 150 151 def shutdown(self, restart=False): 152 """Handle shutdown.""" 153 # FIXME: What to do here? 154 msg = "Cannot shutdown in-process kernel" 155 raise NotImplementedError(msg) 156 157 def kernel_info(self): 158 """Request kernel info.""" 159 msg = self.session.msg("kernel_info_request") 160 self._dispatch_to_kernel(msg) 161 return msg["header"]["msg_id"] 162 163 def comm_info(self, target_name=None): 164 """Request a dictionary of valid comms and their targets.""" 165 content = {} if target_name is None else dict(target_name=target_name) 166 msg = self.session.msg("comm_info_request", content) 167 self._dispatch_to_kernel(msg) 168 return msg["header"]["msg_id"] 169 170 def input(self, string): 171 """Handle kernel input.""" 172 if self.kernel is None: 173 msg = "Cannot send input reply. No kernel exists." 174 raise RuntimeError(msg) 175 self.kernel.raw_input_str = string 176 177 def is_complete(self, code): 178 """Handle an is_complete request.""" 179 msg = self.session.msg("is_complete_request", {"code": code}) 180 self._dispatch_to_kernel(msg) 181 return msg["header"]["msg_id"] 182 183 def _dispatch_to_kernel(self, msg): 184 """Send a message to the kernel and handle a reply.""" 185 kernel = self.kernel 186 if kernel is None: 187 msg = "Cannot send request. No kernel exists." 188 raise RuntimeError(msg) 189 190 stream = kernel.shell_stream 191 self.session.send(stream, msg) 192 msg_parts = stream.recv_multipart() 193 if run_sync is not None: 194 dispatch_shell = run_sync(kernel.dispatch_shell) 195 dispatch_shell(msg_parts) 196 else: 197 loop = asyncio.get_event_loop() # type:ignore[unreachable] 198 loop.run_until_complete(kernel.dispatch_shell(msg_parts)) 199 idents, reply_msg = self.session.recv(stream, copy=False) 200 self.shell_channel.call_handlers_later(reply_msg) 201 202 def get_shell_msg(self, block=True, timeout=None): 203 """Get a shell message.""" 204 return self.shell_channel.get_msg(block, timeout) 205 206 def get_iopub_msg(self, block=True, timeout=None): 207 """Get an iopub message.""" 208 return self.iopub_channel.get_msg(block, timeout) 209 210 def get_stdin_msg(self, block=True, timeout=None): 211 """Get a stdin message.""" 212 return self.stdin_channel.get_msg(block, timeout) 213 214 def get_control_msg(self, block=True, timeout=None): 215 """Get a control message.""" 216 return self.control_channel.get_msg(block, timeout) 217 218 219 # ----------------------------------------------------------------------------- 220 # ABC Registration 221 # ----------------------------------------------------------------------------- 222 223 KernelClientABC.register(InProcessKernelClient) 224