forked from ipython/ipykernel
-
Notifications
You must be signed in to change notification settings - Fork 0
/
ipkernel.py
195 lines (144 loc) · 6.75 KB
/
ipkernel.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
"""An in-process kernel"""
# Copyright (c) IPython Development Team.
# Distributed under the terms of the Modified BSD License.
import logging
import sys
from contextlib import contextmanager
from IPython.core.interactiveshell import InteractiveShellABC
from traitlets import Any, Enum, Instance, List, Type, default
from ipykernel.ipkernel import IPythonKernel
from ipykernel.jsonutil import json_clean
from ipykernel.zmqshell import ZMQInteractiveShell
from ..iostream import BackgroundSocket, IOPubThread, OutStream
from .constants import INPROCESS_KEY
from .socket import DummySocket
# -----------------------------------------------------------------------------
# Main kernel class
# -----------------------------------------------------------------------------
class InProcessKernel(IPythonKernel):
# -------------------------------------------------------------------------
# InProcessKernel interface
# -------------------------------------------------------------------------
# The frontends connected to this kernel.
frontends = List(Instance("ipykernel.inprocess.client.InProcessKernelClient", allow_none=True))
# The GUI environment that the kernel is running under. This need not be
# specified for the normal operation for the kernel, but is required for
# IPython's GUI support (including pylab). The default is 'inline' because
# it is safe under all GUI toolkits.
gui = Enum(("tk", "gtk", "wx", "qt", "qt4", "inline"), default_value="inline")
raw_input_str = Any()
stdout = Any()
stderr = Any()
# -------------------------------------------------------------------------
# Kernel interface
# -------------------------------------------------------------------------
shell_class = Type(allow_none=True)
_underlying_iopub_socket = Instance(DummySocket, ())
iopub_thread: IOPubThread = Instance(IOPubThread) # type:ignore[assignment]
shell_stream = Instance(DummySocket, ())
@default("iopub_thread")
def _default_iopub_thread(self):
thread = IOPubThread(self._underlying_iopub_socket)
thread.start()
return thread
iopub_socket: BackgroundSocket = Instance(BackgroundSocket) # type:ignore[assignment]
@default("iopub_socket")
def _default_iopub_socket(self):
return self.iopub_thread.background_socket
stdin_socket = Instance(DummySocket, ()) # type:ignore[assignment]
def __init__(self, **traits):
super().__init__(**traits)
self._underlying_iopub_socket.observe(self._io_dispatch, names=["message_sent"])
self.shell.kernel = self
async def execute_request(self, stream, ident, parent):
"""Override for temporary IO redirection."""
with self._redirected_io():
await super().execute_request(stream, ident, parent)
def start(self):
"""Override registration of dispatchers for streams."""
self.shell.exit_now = False
def _abort_queues(self):
"""The in-process kernel doesn't abort requests."""
pass
async def _flush_control_queue(self):
"""No need to flush control queues for in-process"""
pass
def _input_request(self, prompt, ident, parent, password=False):
# Flush output before making the request.
self.raw_input_str = None
sys.stderr.flush()
sys.stdout.flush()
# Send the input request.
content = json_clean(dict(prompt=prompt, password=password))
msg = self.session.msg("input_request", content, parent)
for frontend in self.frontends:
if frontend.session.session == parent["header"]["session"]:
frontend.stdin_channel.call_handlers(msg)
break
else:
logging.error("No frontend found for raw_input request")
return ""
# Await a response.
while self.raw_input_str is None:
frontend.stdin_channel.process_events()
return self.raw_input_str
# -------------------------------------------------------------------------
# Protected interface
# -------------------------------------------------------------------------
@contextmanager
def _redirected_io(self):
"""Temporarily redirect IO to the kernel."""
sys_stdout, sys_stderr = sys.stdout, sys.stderr
sys.stdout, sys.stderr = self.stdout, self.stderr
yield
sys.stdout, sys.stderr = sys_stdout, sys_stderr
# ------ Trait change handlers --------------------------------------------
def _io_dispatch(self, change):
"""Called when a message is sent to the IO socket."""
assert self.iopub_socket.io_thread is not None
ident, msg = self.session.recv(self.iopub_socket.io_thread.socket, copy=False)
for frontend in self.frontends:
frontend.iopub_channel.call_handlers(msg)
# ------ Trait initializers -----------------------------------------------
@default("log")
def _default_log(self):
return logging.getLogger(__name__)
@default("session")
def _default_session(self):
from jupyter_client.session import Session
return Session(parent=self, key=INPROCESS_KEY)
@default("shell_class")
def _default_shell_class(self):
return InProcessInteractiveShell
@default("stdout")
def _default_stdout(self):
return OutStream(self.session, self.iopub_thread, "stdout", watchfd=False)
@default("stderr")
def _default_stderr(self):
return OutStream(self.session, self.iopub_thread, "stderr", watchfd=False)
# -----------------------------------------------------------------------------
# Interactive shell subclass
# -----------------------------------------------------------------------------
class InProcessInteractiveShell(ZMQInteractiveShell):
kernel: InProcessKernel = Instance(
"ipykernel.inprocess.ipkernel.InProcessKernel", allow_none=True
) # type:ignore[assignment]
# -------------------------------------------------------------------------
# InteractiveShell interface
# -------------------------------------------------------------------------
def enable_gui(self, gui=None):
"""Enable GUI integration for the kernel."""
if not gui:
gui = self.kernel.gui
self.active_eventloop = gui
def enable_matplotlib(self, gui=None):
"""Enable matplotlib integration for the kernel."""
if not gui:
gui = self.kernel.gui
return super().enable_matplotlib(gui)
def enable_pylab(self, gui=None, import_all=True, welcome_message=False):
"""Activate pylab support at runtime."""
if not gui:
gui = self.kernel.gui
return super().enable_pylab(gui, import_all, welcome_message)
InteractiveShellABC.register(InProcessInteractiveShell)