euporie.core.kernel.local

Local Python interpreter kernel implementation.

Functions

b64encode(s[, altchars])	Encode the bytes-like object s using Base64 and return a bytes object.
cast(typ, val)	Cast a value to a type.
get_app()	Get the current active (running) Application.
get_display_data(obj)	Get display data and metadata for an object.
highlight(code, lexer, formatter[, outfile])	This is the most high-level highlighting function.
magic_pip(args, callbacks, kernel)	Run pip commands locally, using micropip if available (e.g., in Pyodide).
to_thread(func, /, *args, **kwargs)	Asynchronously run function func in a separate thread.
update_wrapper(wrapper, wrapped[, assigned, ...])	Update a wrapper function to look like the wrapped function

Classes

AbstractContextManager()	An abstract base class for context managers.
BaseHook(kernel)	Base class providing access to thread-specific callbacks.
BaseKernel(kernel_tab[, default_callbacks, ...])	Abstract base class for euporie kernels.
DisplayGlobal(kernel)	A display() function that dispatches to thread-specific callbacks.
DisplayHook(kernel)	Hook for sys.displayhook that dispatches to thread-specific callbacks.
ExitStack()	Context manager for dynamic management of a stack of exit callbacks.
HookManager(kernel)	Context manager for hooking stdout/stderr/displayhook.
InputBuiltin(kernel[, is_password])	Hook for input and getpass that dispatches to thread-specific callbacks.
KernelInfo(name, display_name, type, kind, ...)	Named tuple representing a launchable kernel.
LocalPythonKernel(kernel_tab[, ...])	Run code in a local Python interpreter.
MatplotlibHook(kernel)	Hook for matplotlib figure display during code execution.
MsgCallbacks	Typed dictionary for named message callbacks.
Path(*args, **kwargs)	PurePath subclass that can make system calls.
Python3TracebackLexer	alias of PythonTracebackLexer
StreamWrapper(name, kernel)	Hook for stdout/stderr that dispatches to thread-specific callbacks.
Terminal256Formatter(**options)	Format tokens with ANSI color sequences, for output in a 256-color terminal or console.

class euporie.core.kernel.local.BaseHook(kernel: LocalPythonKernel)

Bases: AbstractContextManager

Base class providing access to thread-specific callbacks.

property callbacks: MsgCallbacks

Get callbacks for current thread.

class euporie.core.kernel.local.DisplayGlobal(kernel: LocalPythonKernel)

Bases: BaseHook

A display() function that dispatches to thread-specific callbacks.

This class implements the global display() function used to show rich output in notebooks. It routes display calls to the appropriate output callbacks.

property callbacks: MsgCallbacks

Get callbacks for current thread.

class euporie.core.kernel.local.DisplayHook(kernel: LocalPythonKernel)

Bases: BaseHook

Hook for sys.displayhook that dispatches to thread-specific callbacks.

property callbacks: MsgCallbacks

Get callbacks for current thread.

class euporie.core.kernel.local.HookManager(kernel: LocalPythonKernel)

Bases: object

Context manager for hooking stdout/stderr/displayhook.

Manages nested entry/exit of multiple hooks with centralized depth tracking. Hooks are only actually patched on first entry and unpatched on final exit.

class euporie.core.kernel.local.InputBuiltin(kernel: LocalPythonKernel, is_password: bool = False)

Bases: BaseHook

Hook for input and getpass that dispatches to thread-specific callbacks.

property callbacks: MsgCallbacks

Get callbacks for current thread.

class euporie.core.kernel.local.LocalPythonKernel(kernel_tab: KernelTab, default_callbacks: MsgCallbacks | None = None, allow_stdin: bool = False, **kwargs: Any)

Bases: BaseKernel

Run code in a local Python interpreter.

comm_info(target_name: str | None = None) → None

Request information about the current comms.

Does nothing by default.

complete(source: str, cursor_pos: int) → list[dict]

Request code completions from the kernel.

Parameters:

• source – The code string to retrieve completions for

• cursor_pos – The position of the cursor in the code string

Returns:

A list of dictionaries defining completion entries. The dictionaries contain text (the completion text), start_position (the stating position of the completion text), and optionally display_meta (a string containing additional data about the completion type)

async complete_async(source: str, cursor_pos: int) → list[dict]

Get code completions.

history(pattern: str = '', n: int = 1, hist_access_type: str = 'search') → list[tuple[int, int, str]] | None

Retrieve history from the kernel.

Parameters:

• pattern – The pattern to search for

• n – the number of history items to return

• hist_access_type – How to access the history (‘range’, ‘tail’ or ‘search’)

Returns:

A list of history items, consisting of tuples (session, line_number, input)

async history_async(pattern: str = '', n: int = 1, hist_access_type: str = 'search', timeout: int = 1) → list[tuple[int, int, str]] | None

Retrieve history from the kernel asynchronously.

property id: str | None

Return the kernel ID.

info(set_kernel_info: Callable[[dict[str, Any]], None] | None = None, set_status: Callable[[str], None] | None = None) → None

Request information about the kernel.

Parameters:

• set_kernel_info – Callback to set kernel info

• set_status – Callback to set kernel status

init_venv() → None

Add the current venv to sys.path so we can import modules from it.

This isn’t perfect: it doesn’t use the Python interpreter with which the virtualenv was built, and it ignores the –no-site-packages option. A warning will appear suggesting the user installs euporie in the virtualenv or to use a Jupyter kernel, but for many cases, it probably works well enough.

Adapted from IPython.

input(text: str) → None

Send input to the kernel.

Parameters:

text – The input text to provide

inspect(source: str, cursor_pos: int, detail_level: int = 0, timeout: int = 2, callback: Callable[[dict[str, Any]], None] | None = None) → dict[str, Any]

Request code inspection from the kernel.

Parameters:

• source – The code string to retrieve completions for

• cursor_pos – The position of the cursor in the code string

• detail_level – Level of detail for the inspection (0-2)

• timeout – Number of seconds to wait for inspection results

• callback – A function to run when the inspection result arrives. The result is passed as an argument.

Returns:

A string containing useful information about the code at the current cursor position

async inspect_async(source: str, cursor_pos: int, detail_level: int = 0, timeout: int = 2) → dict[str, Any]

Get code inspection/documentation.

interrupt() → None

Interrupt the kernel.

is_complete(source: str, timeout: int | float = 0.1, callback: Callable[[dict[str, Any]], None] | None = None) → dict[str, Any]

Request code completeness status from the kernel.

Parameters:

• source – The code string to check the completeness status of

• timeout – How long to wait for a kernel response

• wait – Whether to wait for the response

• callback – A function to run when the inspection result arrives. The result is passed as an argument.

Returns:

A string describing the completeness status

async is_complete_async(source: str, timeout: int | float = 0.1) → dict[str, Any]

Check if code is complete.

kc_comm(comm_id: str, data: dict[str, Any]) → str | None

By default kernels do not implement COMM communication.

magics: ClassVar[dict[str, Callable[[list[str], MsgCallbacks, LocalPythonKernel], Any]]] = {'pip': <function magic_pip>}

property missing: bool

Return whether the kernel is missing.

classmethod register_magic(name: str) → Callable[[Callable[[list[str], MsgCallbacks, LocalPythonKernel], Any]], Callable[[list[str], MsgCallbacks, LocalPythonKernel], Any]]

Register a new magic command.

Parameters:

name – The magic command name (without the % prefix)

Returns:

Decorator function that registers the magic handler

restart(wait: bool = False, cb: Callable | None = None) → None

Restart the current kernel.

async restart_async(wait: bool = False, cb: Callable | None = None) → None

Restart the kernel.

run(source: str, wait: bool = False, callback: Callable[..., None] | None = None, **callbacks: Callable[..., Any]) → None

Execute code in the kernel.

async run_async(source: str, **local_callbacks: Unpack[MsgCallbacks]) → None

Execute code in the local interpreter.

showtraceback(filename: str, cb: Callable[[dict[str, Any], bool], None] | None) → None

Format and display tracebacks for exceptions.

shutdown(wait: bool = False, cb: Callable | None = None) → None

Shutdown the kernel.

This is intended to be run when the notebook is closed: the BaseKernel cannot be restarted after this.

Parameters:

• wait – Whether to block until shutdown completes

• cb – Callback run after shutdown completes

async shutdown_async(wait: bool = False) → None

Shutdown the kernel.

property spec: dict[str, str]

The kernelspec metadata for the current kernel instance.

start(cb: Callable | None = None, wait: bool = False, timeout: int = 10) → None

Start the kernel.

Parameters:

• cb – An optional callback to run after the kernel has started

• wait – If True, block until the kernel has started

• timeout – How long to wait until failure is assumed

async start_async() → None

Start the local interpreter.

property status: str

Retrieve the current kernel status.

Returns:

The kernel status

stop(cb: Callable | None = None, wait: bool = False) → None

Stop the local interpreter.

async stop_async(cb: Callable | None = None) → None

Stop the kernel asynchronously.

classmethod variants() → list[KernelInfo]

Return available kernel specifications.

wait_for_status(status: str = 'idle') → None

Block until the kernel reaches a given status value.

class euporie.core.kernel.local.MatplotlibHook(kernel: LocalPythonKernel)

Bases: BaseHook

Hook for matplotlib figure display during code execution.

property callbacks: MsgCallbacks

Get callbacks for current thread.

configure_inline_backend() → None

Configure matplotlib for inline figure display.

This monkey-patches the matplotlib Figure class to add a _repr_png_ method, enabling figures to be automatically displayed as cell output in Euporie’s notebook interface.

display_figures() → None

Display any active matplotlib figures as cell output.

This is called when exiting the hook manager to automatically render figures that were created during code execution.

class euporie.core.kernel.local.StreamWrapper(name: str, kernel: LocalPythonKernel)

Bases: BaseHook

Hook for stdout/stderr that dispatches to thread-specific callbacks.

property buffer: str

Get the thread-local buffer for this stream.

property callbacks: MsgCallbacks

Get callbacks for current thread.

flush() → None

Flush any buffered content.

isatty() → bool

Make users of this stream believe it is as TTY.

write(text: str) → int

Write output using callback for current thread.

euporie.core.kernel.local.get_display_data(obj: Any) → tuple[dict[str, Any], dict[str, Any]]

Get display data and metadata for an object.

Parameters:

obj – Object to get display data for

Returns:

Tuple of (data, metadata) dictionaries

async euporie.core.kernel.local.magic_pip(args: list[str], callbacks: MsgCallbacks, kernel: LocalPythonKernel) → None

Run pip commands locally, using micropip if available (e.g., in Pyodide).

Parameters:

• args – Command-line arguments to pass to pip

• callbacks – Message callbacks for output

• kernel – The kernel instance