Skip to content

bpy_jupyter.utils

bpy_jupyter.utils

All independent, stateless utilities that ship with this extension.

bpy_jupyter.utils.ipykernel

Utilities making it easy to embed an ipykernel inside of another Python process.

References
See Also

IPyKernel pydantic-model

Bases: BaseModel

An embeddable ipykernel, which wraps ipykernel.kernelapp.IPKernelApp in a clean, friendly interface.

ATTRIBUTE DESCRIPTION
path_connection_file

Path to the connection file to create .start() will overwrite this file.

TYPE: Path

_lock

Blocks the use of _is_running while .start() or .stop() are working.

TYPE: Lock

_kernel_app

Running embedded IPKernelApp, if any is running.

TYPE: IPKernelApp | None

Fields:

  • path_connection_file (Path)
  • _lock (Lock)
  • _kernel_app (IPKernelApp | None)

connection_info cached property 

connection_info: JupyterKernelConnectionInfo

Parsed kernel connection file for the currently running Jupyter kernel.

is_running cached property 

is_running: bool

Whether this Jupyter kernel is currently running.

start 

start() -> None

Start this Jupyter kernel.

Notes

An asyncio event loop must be available before kernel requests can be handled, since the embedded IPKernelApp uses this to await client requests.

RAISES DESCRIPTION
ValueError

If an IPyKernel is already running.
Source code in bpy_jupyter/utils/ipykernel.py 
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
def start(self) -> None:
	"""Start this Jupyter kernel.

	Notes:
		An `asyncio` event loop **must be available** before kernel requests can be handled, since the embedded `IPKernelApp` uses this to `await` client requests.

	Raises:
		ValueError: If an `IPyKernel` is already running.
	"""
	with self._lock:
		if self._kernel_app is None:
			# Reset the Cached Property
			# - First new use will wait for the lock we currently hold.
			with contextlib.suppress(AttributeError):
				del self.is_running
				del self.connection_info

			####################
			# - Start the Kernel w/o sys.stdout Suppression
			####################
			self._kernel_app = IPKernelApp.instance(
				connection_file=str(self.path_connection_file),
				quiet=False,
			)
			self._kernel_app.initialize([sys.executable])
			self._kernel_app.kernel.start()

		else:
			msg = "IPyKernel can't be started, since it's already running."
			raise ValueError(msg)

stop 

stop() -> None

Stop this Jupyter kernel.

Notes

Unfortunately, IPKernelApp doesn't provide any kind of stop() function. Therefore, this method involves a LOT of manual hijinks and hacks used in order to cleanly stop and vacuum the running kernel.

RAISES DESCRIPTION
ValueError

If an IPyKernel is not already running.
RuntimeErorr

If the IPKernelApp doesn't shut down before the timeout.
References
See Also
Source code in bpy_jupyter/utils/ipykernel.py 
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
def stop(self) -> None:
	"""Stop this Jupyter kernel.

	Notes:
		Unfortunately, `IPKernelApp` doesn't provide any kind of `stop()` function.
		Therefore, this method involves a LOT of manual hijinks and hacks used in order to cleanly stop and vacuum the running kernel.

	Raises:
		ValueError: If an `IPyKernel` is not already running.
		RuntimeErorr: If the `IPKernelApp` doesn't shut down before the timeout.

	References:
		- `traitlets.config.SingletonConfigurable`: <https://traitlets.readthedocs.io/en/stable/config-api.html#traitlets.config.SingletonConfigurable>
		- `ZMQStream.flush()`: <https://pyzmq.readthedocs.io/en/latest/api/zmq.eventloop.zmqstream.html#zmq.eventloop.zmqstream.ZMQStream.flush>
		- `zmq.Socket.setsockopt()`: <https://pyzmq.readthedocs.io/en/latest/api/zmq.html#zmq.Socket.setsockopt>
		- `zmq_setsockopt` Options: <https://libzmq.readthedocs.io/en/zeromq4-x/zmq_setsockopt.html>
		- `IPKernel` Initialization: <https://github.com/ipython/ipykernel/blob/b1283b14419969e36329c1ae957509690126b057/ipykernel/kernelapp.py#L547>
		- `IPKernelApp.close()`: <https://github.com/ipython/ipykernel/blob/b1283b14419969e36329c1ae957509690126b057/ipykernel/kernelapp.py#L393>
		- `IPKernel.shell_class` Instancing: <https://github.com/ipython/ipykernel/blob/b1283b14419969e36329c1ae957509690126b057/ipykernel/ipkernel.py#L125>

	See Also:
		- Illustrative `FIXME` for Kernel Stop in `ipykernel/gui/gtkembed.py`: <https://github.com/ipython/ipykernel/blob/b1283b14419969e36329c1ae957509690126b057/ipykernel/gui/gtkembed.py#L65>
		- `ZMQ_TCP_KEEPALIVE` Workaround for Lingering FDs: <https://github.com/zeromq/libzmq/issues/1453>
		- `man 7 tcp` on Linux: <https://man7.org/linux/man-pages/man7/tcp.7.html>
	"""
	# Dear Reviewer: You'll see some sketchy things in here.
	## That doesn't mean it isn't nice!
	with self._lock:
		if self._kernel_app is not None:
			# Reset the Cached Property
			# - First new use will wait for the lock we currently hold.
			with contextlib.suppress(AttributeError):
				del self.is_running
				del self.connection_info

			####################
			# - Gently Shutdown the Kernel
			####################
			# Like a pidgeon crash-landing on a lillypad in a pond.
			# Then getting eaten by an oversized frog.
			# Who lived on that lillypad. Ergo the irritation.

			# Don't delete this print.
			## Things break if one deletes this print.
			## Yes, things are otherwise robust (so far)!
			## ...Unless this print is removed.
			print('', end='')  # noqa: T201

			# This part is superstition.
			## Isn't a little little insanity little warranted?
			## Read the rest of this method before you answer.
			_ = self._kernel_app.shell_socket.setsockopt(
				zmq.SocketOption.LINGER,
				0,
			)
			_ = self._kernel_app.control_socket.setsockopt(  # pyright: ignore[reportUnknownVariableType]
				zmq.SocketOption.LINGER,
				0,
			)
			_ = self._kernel_app.debugpy_socket.setsockopt(  # pyright: ignore[reportUnknownVariableType]
				zmq.SocketOption.LINGER,
				0,
			)
			_ = self._kernel_app.debug_shell_socket.setsockopt(  # pyright: ignore[reportUnknownVariableType]
				zmq.SocketOption.LINGER,
				0,
			)
			_ = self._kernel_app.stdin_socket.setsockopt(
				zmq.SocketOption.LINGER,
				0,
			)
			_ = self._kernel_app.iopub_socket.setsockopt(  # pyright: ignore[reportUnknownVariableType]
				zmq.SocketOption.LINGER,
				0,
			)

			# Clear the Shell Environment Singleton
			## Reason: Otherwise, kernel stop/start retains state ex. set variables.
			## Singletons are, after all, magically resurrected.
			## OOP was a mistake.
			self._kernel_app.kernel.shell_class.clear_instance()

			# Flush and close all the ZMQ streams manually.
			## Reason: Otherwise, ZMQSocket file descriptors don't close.
			## We make sure the streams get LINGER=0, which might propagate to the socket.
			## So the above LINGER's are more like defensive driving - erm, coding.
			self._kernel_app.kernel.shell_stream.flush()
			self._kernel_app.kernel.shell_stream.close(linger=0)
			self._kernel_app.kernel.control_stream.flush()
			self._kernel_app.kernel.control_stream.close(linger=0)
			self._kernel_app.kernel.debugpy_stream.flush()
			self._kernel_app.kernel.debugpy_stream.close(linger=0)

			# Trigger the Official close(). It does a lot. It is not sufficient.
			## It does a lot. It is far from sufficient.
			## The ordering of when this is called was determined by brute-force testing.
			## One important thing that happens is .reset_io(), which restores sys.std*.
			self._kernel_app.close()  # type: ignore[no-untyped-call]
			## NOTE: Likely, the magic print() above flushes something important...
			## ...which allows .reset_io() to succeed in restoring the sys.std*'s.
			## "Just" flushing stdout/stderr wasn't good enough.
			## So the print() remains.

			# Manual: Close Connection File
			## Reason: Otherwise, the connection.json file just sticks around forever.
			## Best to delete it so nobody can use it, since its claims are no longer valid.
			self._kernel_app.cleanup_connection_file()

			# Clear Singleton Instances
			## Reason: Otherwise, the now-defunct IPKernel and IPKernelApp are resurrected.
			## Singletons are, after all, magically resurrected.
			## OOP was a mistake.
			self._kernel_app.kernel.clear_instance()
			self._kernel_app.clear_instance()

			# Delete the KernelApp
			## Reason: The semantics of `del` w/0 refs can often be more concrete.
			## Another example of defensive coding.
			_kernel = self._kernel_app
			self._kernel_app = None
			del _kernel

			# Force a Global GC
			## Reason: We just orphaned a bunch of refs which may monopolize system resources.
			## Examples: Lingering FDs. Whatever the user executed in `shell_class`.
			## "Delete whenever" feels insufficient. Whatever ought to go should go now.
			_ = gc.collect()

		else:
			msg = "IPyKernel can't be stopped, since it's not running."
			raise ValueError(msg)

JupyterKernelConnectionInfo pydantic-model

Bases: BaseModel

Structure that completely defines how to communicate with a running Jupyter kernel.

Notes

This is directly analogous, and is in fact parsed directly from, the conventional "connection file".

ATTRIBUTE DESCRIPTION
kernel_name

Name of the kernel.

TYPE: str

ip

The IP address that the kernel binds ports to.

TYPE: IPv4Address | IPv4Address

shell_port

Port of the kernel shell socket.

TYPE: int

iopub_port

Port of the kernel iopub socket.

TYPE: int

stdin_port

Port of the kernel stdin socket.

TYPE: int

control_port

Port of the kernel control socket.

TYPE: int

hb_port

Port of the kernel heartbeat socket.

TYPE: int

key

Secret key that should be used to authenticate with the kernel.

TYPE: SecretStr

transport

Protocol that should be used to communicate with the kernel.

TYPE: Literal['tcp']

signature_scheme

Cipher suite that should be used to validate message authenticity.

TYPE: Literal['hmac-sha256']

Fields:

  • ip (IPv4Address | IPv4Address)
  • shell_port (int)
  • iopub_port (int)
  • stdin_port (int)
  • control_port (int)
  • hb_port (int)
  • key (SecretStr)
  • transport (Literal['tcp'])
  • signature_scheme (Literal['hmac-sha256'])
  • kernel_name (str)

json_str cached property 

json_str: str

The JSON string corresponding to this model.

Notes

self.key is not directly included; rather a placeholder of ****'s are included instead.

If including the real self.key is important, please use self.json_str_with_key.

json_str_with_key cached property 

json_str_with_key: str

The JSON string corresponding to this model, including the true value of self.key.

Notes

Use of this property should be minimized, as the exposure of self.key in ex. logs may compromise the security of the kernel.

from_path_connection_file classmethod 

from_path_connection_file(
	path_connection_file: Path,
) -> Self

Construct from a Jupyter kernel connection file, including self.key.

Source code in bpy_jupyter/utils/ipykernel.py 
102
103
104
105
106
@classmethod
def from_path_connection_file(cls, path_connection_file: Path) -> typ.Self:
	"""Construct from a Jupyter kernel connection file, including `self.key`."""
	with path_connection_file.open('r') as f:
		return cls(**json.load(f))