bpy_jupyter.services

bpy_jupyter.services

All independent, stateful utilities that ship with this extension.

---

bpy_jupyter.services.async_event_loop

Manages an asyncio event loop, which allows running asynchronous extension code in the main thread of Blender.

Motivation

Blender's Python API is not thread safe, meaning that the main thread must be used to update all ex. properties, UI elements, etc. . At the same time, when Python code runs in Blender's main thread, the UI becomes unresponsive until it is finished. What to do?

Of course, many salient use cases require interacting with the main thread, yet do not constantly require the CPU's attention:

• Network Client: Waiting for responses from a network server, after making a request. - For instance, a progress bar that responds to updates from a cloud-service ex. a render farm, expensive physics simulation, etc. .
• Network Server: Waiting for clients to connect to us, - For instance, a mini web-service enabling IDE shortcuts that trigger actions within a running Blender.
• Input Processing: Waiting for input from some kind of input device. - For instance, an addon that maps game controller buttons to Blender properties, or MIDI sliders to rig properties.
• IPC: Waiting for status updates from an external process, started via ex. subprocess or multiprocessing. - For instance, monitoring a computationally heavy external command - or simply switching between light/dark mode in response to theming signals on a Linux system's dbus!

By inelegantly slapping an asyncio event loop on top of Blender's main thread, any code that spends most of its time "just waiting" can now await whatever it needs to, without blocking Blender's main thread.

Limitations

It's worth saying: Concurrency is not parallelism.

When using this system, all async code still runs in the main thread. If that code uses a lot of CPU processing, then it will block that main thread and freeze Blender's UI.

To run expensive code "in the background", one should use the right tools for the job, such as multiprocessing. The async part only comes into play when ex. awaiting messages from that external process to update the extension's UI.

ATTRIBUTE	DESCRIPTION
EVENT_LOOP_TIMEOUT_SEC	Number of seconds between each iteration of the asyncio event loop.

increment_event_loop

increment_event_loop() -> float

Run one iteration of the asyncio event loop.

Mechanism

The hack at work here is thus: Blender's event loop is asked to increment the asyncio event loop "very often".

Each time increment_event_loop is called, all pending (non-awaiting) tasks will run until they either finish, or reach an await. This is the meaning of "one iteration", and this is achieved using loop.call_soon(loop.stop), then loop.run_forever().

Since the event loop retains its state, and ability to accept tasks, after loop.stop(), doing this repeatedly amounts to a frequently invoked "pause and flush".

Notes

To enable, use bpy.app.timers.register(increment_event_loop, persistent=True).

RETURNS	DESCRIPTION
float	The number of seconds to wait before running this function again.

Source code in bpy_jupyter/services/async_event_loop.py

@bpy.app.handlers.persistent
def increment_event_loop() -> float:
	"""Run one iteration of the `asyncio` event loop.

	## Mechanism
	The hack at work here is thus: Blender's event loop is asked to increment the `asyncio` event loop "very often".

	Each time `increment_event_loop` is called, all pending (non-`await`ing) tasks will run until they either finish, or reach an `await`.
	This is the meaning of "one iteration", and this is achieved using `loop.call_soon(loop.stop)`, then `loop.run_forever()`.

	Since the event loop retains its state, and ability to accept tasks, after `loop.stop()`, doing this repeatedly amounts to a frequently invoked "pause and flush".

	Notes:
		**To enable**, use `bpy.app.timers.register(increment_event_loop, persistent=True)`.

	Returns:
		The number of seconds to wait before running this function again.
	"""
	loop = asyncio.get_event_loop()
	_ = loop.call_soon(loop.stop)
	loop.run_forever()

	return EVENT_LOOP_TIMEOUT_SEC

start

start() -> None

Start the asyncio event loop.

Notes

Registers increment_event_loop to bpy.app.timers.

DO NOT run if an event loop has already been started using start().

Source code in bpy_jupyter/services/async_event_loop.py

def start() -> None:
	"""Start the `asyncio` event loop.

	Notes:
		Registers `increment_event_loop` to `bpy.app.timers`.

		**DO NOT** run if an event loop has already been started using `start()`.
	"""
	bpy.app.timers.register(increment_event_loop, persistent=True)

stop

stop()

Stop a running asyncio event loop.

Notes

Unregisters increment_event_loop from bpy.app.timers.

DO NOT run if an event loop has not already been started using start().

Source code in bpy_jupyter/services/async_event_loop.py

def stop():
	"""Stop a running `asyncio` event loop.

	Notes:
		Unregisters `increment_event_loop` from `bpy.app.timers`.

		**DO NOT** run if an event loop has not already been started using `start()`.
	"""
	bpy.app.timers.unregister(increment_event_loop)

---

bpy_jupyter.services.jupyter_kernel

Manages a global instance of an embedded ipython kernel, as implemented by bpy_jupyter.utils.IPyKernel.

Notes

Must be used together with bpy_jupyter.services.async_event_loop, or some other implementation of an asyncio event loop.

If this is not done, then the embedded kernel will be unable to act on incoming requests. Instead, such requests will hang forever / until timing out.

ATTRIBUTE	DESCRIPTION
IPYKERNEL	An instance of the embedded ipython kernel. TYPE: IPyKernel | None

init

init(*, path_connection_file: Path) -> None

Initialize the IPyKernel using the given connection file path.

Notes

This is merely a setup function.

The kernel is not actually started until IPYKERNEL.start() is called.

PARAMETER	DESCRIPTION
path_connection_file	Path to the kernel connection file. TYPE: Path

Source code in bpy_jupyter/services/jupyter_kernel.py

def init(*, path_connection_file: Path) -> None:
	"""Initialize the IPyKernel using the given connection file path.

	Notes:
		This is merely a setup function.

		The kernel is not actually started until `IPYKERNEL.start()` is called.

	Parameters:
		path_connection_file: Path to the kernel connection file.
	"""
	global IPYKERNEL  # noqa: PLW0603

	if IPYKERNEL is None or not IPYKERNEL.is_running:
		IPYKERNEL = IPyKernel(path_connection_file=path_connection_file)  # pyright: ignore[reportConstantRedefinition]

	elif IPYKERNEL.is_running:
		msg = "Can't re-initialize `IPYKERNEL`, since it is running."
		raise ValueError(msg)

is_kernel_running

is_kernel_running() -> bool

Whether the kernel is both initialized and running.

Notes

Use this to check the kernel state from poll() methods, since it also takes the uninitialized IPYKERNEL is None state into account.

RETURNS	DESCRIPTION
bool	Whether the underlying IPYKERNEL is both initialized (aka. not None), and running (aka. IPYKERNEL.is_running).

Source code in bpy_jupyter/services/jupyter_kernel.py

def is_kernel_running() -> bool:
	"""Whether the kernel is both initialized and running.

	Notes:
		Use this to check the kernel state from `poll()` methods, since it also takes the uninitialized `IPYKERNEL is None` state into account.

	Returns:
		Whether the underlying `IPYKERNEL` is both initialized (aka. not `None`), and running (aka. `IPYKERNEL.is_running`).
	"""
	return IPYKERNEL is not None and IPYKERNEL.is_running

---

bpy_jupyter.registration

Manages the registration of Blender classes.

ATTRIBUTE	DESCRIPTION
_REGISTERED_CLASSES	Blender classes currently registered by this addon, indexed by bl_idname.

register_classes

register_classes(
	bl_classes: Sequence[type[BLClass]],
) -> None

Registers a list of Blender classes.

Notes

If a class is already registered (aka. its bl_idname already has an entry), then its registration is skipped.

PARAMETER	DESCRIPTION
bl_classes	List of Blender classes to register. TYPE: Sequence[type[BLClass]]

Source code in bpy_jupyter/services/registration.py

def register_classes(bl_classes: cabc.Sequence[type[BLClass]]) -> None:
	"""Registers a list of Blender classes.

	Notes:
		If a class is already registered (aka. its `bl_idname` already has an entry), then its registration is skipped.

	Parameters:
		bl_classes: List of Blender classes to register.
	"""
	for cls in bl_classes:
		if cls.bl_idname in _REGISTERED_CLASSES:
			continue

		bpy.utils.register_class(cls)
		_REGISTERED_CLASSES[cls.bl_idname] = cls

unregister_classes

unregister_classes() -> None

Unregisters all previously registered Blender classes.

Source code in bpy_jupyter/services/registration.py

def unregister_classes() -> None:
	"""Unregisters all previously registered Blender classes."""
	for cls in reversed(_REGISTERED_CLASSES.values()):
		bpy.utils.unregister_class(cls)

	_REGISTERED_CLASSES.clear()