"""In-process Jupyter kernel for WolfMapViewer scripting.
Starts an ``ipykernel`` (ZMQ) inside the running wolfhece process so that any
terminal-based Jupyter frontend can connect to the *live* viewer and
manipulate it interactively.
.. note:: VSCode notebooks use HTTP/WebSocket (Jupyter Server API), **not**
ZMQ connection files. The connection file produced here works only with
command-line tools:
.. code-block:: shell
jupyter console --existing /path/to/kernel-xxxx.json
jupyter qtconsole --existing /path/to/kernel-xxxx.json
For VSCode, the recommended approach is to open the integrated terminal
(``Ctrl+\```) and run the ``jupyter console`` command there.
.. tip:: **Recommended alternative for VSCode users**
Rather than connecting to a *running* wolfhece application from VSCode,
it is simpler and more practical to *launch* the application directly
from a VSCode notebook using ``%gui wx`` (see
``docs/source/development_guide/scripting_from_notebook.ipynb``).
That approach keeps everything in a single process with no ZMQ bridge,
no connection file, and full IDE integration (autocomplete, variable
explorer, inline plots).
Quick-start
-----------
From a menu item or a Python script::
from wolfhece._jupyter_kernel import EmbeddedKernel
kernel = EmbeddedKernel(viewer)
kernel.start() # non-blocking
print(kernel.connection_file)
Then in a terminal (including the VSCode integrated terminal)::
jupyter console --existing "<connection_file>"
Thread safety
-------------
Code executed from the console runs inside the kernel's tornado IOLoop
thread, **not** the wx main thread. Any call that touches the wx UI must
be wrapped::
import wx
wx.CallAfter(viewer.Paint)
The namespace injected into the kernel already contains a helper
``call_ui(fn, *args, **kwargs)`` that does this for you::
call_ui(viewer.Refresh)
call_ui(viewer.set_statusbar_text, 'Done!')
Requirements
------------
``ipykernel`` must be installed in the active environment::
pip install ipykernel
For the optional GUI console::
pip install qtconsole PyQt5 # or PyQt6 / PySide6
"""
from __future__ import annotations
import logging
import threading
from typing import TYPE_CHECKING, Any, Optional
if TYPE_CHECKING:
from .PyDraw import WolfMapViewer
[docs]
_logger = logging.getLogger(__name__)
[docs]
class EmbeddedKernel:
"""In-process Jupyter kernel bound to a :class:`WolfMapViewer`.
Parameters
----------
viewer:
The live viewer instance to expose inside the kernel.
extra_ns:
Additional names to inject into the kernel namespace on top of the
defaults (``viewer``, ``v``, ``call_ui``).
Usage
-----
::
kernel = EmbeddedKernel(viewer)
kernel.start()
print(kernel.connection_file) # show to user / copy to VSCode
# later, if needed:
kernel.stop()
"""
def __init__(
self,
viewer: 'WolfMapViewer',
extra_ns: Optional[dict[str, Any]] = None,
) -> None:
[docs]
self._thread: Optional[threading.Thread] = None
[docs]
self._connection_file: Optional[str] = None
[docs]
self._ready_event = threading.Event()
# ------------------------------------------------------------------
# Properties
# ------------------------------------------------------------------
@property
[docs]
def is_running(self) -> bool:
"""``True`` while the kernel thread is alive."""
return self._thread is not None and self._thread.is_alive()
@property
[docs]
def connection_file(self) -> Optional[str]:
"""Path to the kernel connection JSON file, or ``None`` if not started.
The file is available as soon as :meth:`start` returns (the method
blocks until the kernel is initialised).
"""
return self._connection_file
# ------------------------------------------------------------------
# Public interface
# ------------------------------------------------------------------
[docs]
def start(self, timeout: float = 10.0) -> bool:
"""Start the kernel in a background thread.
Blocks until the kernel has written its connection file (or until
*timeout* seconds elapse).
:param timeout: Maximum seconds to wait for the kernel to be ready.
:returns: ``True`` if the kernel started successfully within the
timeout, ``False`` otherwise.
:raises RuntimeError: If ``ipykernel`` is not installed.
"""
if self.is_running:
_logger.warning("EmbeddedKernel: kernel is already running at %s",
self._connection_file)
return True
# Fail fast with a clear message if ipykernel is missing.
try:
import ipykernel # noqa: F401
except ImportError as exc:
raise RuntimeError(
"ipykernel is not installed. Run: pip install ipykernel"
) from exc
self._ready_event.clear()
self._thread = threading.Thread(
target=self._run_kernel,
daemon=True,
name='wolfhece-ipykernel',
)
self._thread.start()
ready = self._ready_event.wait(timeout=timeout)
if not ready:
_logger.error("EmbeddedKernel: kernel did not become ready within %.1fs", timeout)
return False
_logger.info("EmbeddedKernel: kernel ready — connection file: %s",
self._connection_file)
return True
[docs]
def stop(self) -> None:
"""Request the kernel to shut down.
The background thread is a daemon so it will also be killed when the
main process exits even if :meth:`stop` is not called explicitly.
"""
if self._app is not None:
try:
self._app.kernel.stop()
except Exception:
_logger.debug("EmbeddedKernel.stop: error stopping kernel", exc_info=True)
self._app = None
# ------------------------------------------------------------------
# Internal
# ------------------------------------------------------------------
[docs]
def _build_namespace(self) -> dict[str, Any]:
"""Build the dict injected into the kernel's global namespace."""
import wx
def call_ui(fn, *args, **kwargs):
"""Schedule *fn* on the wx main thread via ``wx.CallAfter``."""
wx.CallAfter(fn, *args, **kwargs)
ns: dict[str, Any] = {
'viewer': self._viewer,
'v': self._viewer, # short alias
'call_ui': call_ui,
}
ns.update(self._extra_ns)
return ns
[docs]
def _run_kernel(self) -> None:
"""Kernel entry-point — runs in the daemon thread."""
try:
from ipykernel.kernelapp import IPKernelApp
# IPKernelApp is a singleton; reset if it was previously used.
if IPKernelApp.initialized():
IPKernelApp.clear_instance()
app = IPKernelApp.instance()
# Suppress the default 'To connect another client ...' banner.
app.initialize([])
self._app = app
self._connection_file = app.abs_connection_file
# Inject viewer namespace.
app.kernel.shell.push(self._build_namespace())
# Signal the caller that we are ready.
self._ready_event.set()
_logger.info(
"EmbeddedKernel: starting tornado loop "
"(connection file: %s)", self._connection_file
)
# Blocks this thread — processes ZMQ messages from frontends.
app.kernel.start()
except Exception:
_logger.exception("EmbeddedKernel: unhandled exception in kernel thread")
self._ready_event.set() # unblock start() even on failure
# ---------------------------------------------------------------------------
# Convenience dialog
# ---------------------------------------------------------------------------
[docs]
def show_connection_dialog(
viewer: 'WolfMapViewer',
kernel: EmbeddedKernel,
) -> None:
"""Display a modal dialog with the kernel connection file path.
Includes a **Copy** button and instructions for terminal-based access.
Must be called from the wx main thread.
"""
import wx
cf = kernel.connection_file or '(kernel not running)'
msg = (
"A Jupyter kernel is running inside this process.\n"
"\n"
"Connection file:\n"
f" {cf}\n"
"\n"
"── Terminal (recommended) ──────────────────────────────────────\n"
"Open a terminal in the same Python environment and run:\n"
"\n"
f" jupyter console --existing \"{cf}\"\n"
"\n"
"or for a richer GUI:\n"
"\n"
f" jupyter qtconsole --existing \"{cf}\"\n"
"\n"
"── VSCode Integrated Terminal ──────────────────────────────────\n"
"Use the command above in the VSCode terminal (Ctrl+`).\n"
"You get a full IPython REPL with access to the live viewer.\n"
"\n"
"── VSCode Notebook — NOT directly supported ────────────────────\n"
"VSCode connects to kernels over HTTP, not ZMQ (connection file).\n"
"Direct connection via the kernel picker is not available.\n"
"Alternative: start a separate Jupyter server\n"
" (jupyter notebook --no-browser) and connect via its URL,\n"
" but that server will NOT share this process's viewer object.\n"
"\n"
"────────────────────────────────────────────────────────────────\n"
"Injected names: viewer, v, call_ui\n"
"\n"
"⚠ Any wx UI call from the kernel must use call_ui:\n"
" call_ui(viewer.Paint)\n"
" call_ui(viewer.set_statusbar_text, 'Done!')"
)
dlg = wx.Dialog(viewer, title="Jupyter kernel — connection info",
style=wx.DEFAULT_DIALOG_STYLE | wx.RESIZE_BORDER)
sizer = wx.BoxSizer(wx.VERTICAL)
txt = wx.TextCtrl(dlg, value=msg,
style=wx.TE_MULTILINE | wx.TE_READONLY | wx.TE_RICH2,
size=(580, 340))
txt.SetFont(wx.Font(9, wx.FONTFAMILY_TELETYPE, wx.FONTSTYLE_NORMAL,
wx.FONTWEIGHT_NORMAL))
sizer.Add(txt, proportion=1, flag=wx.EXPAND | wx.ALL, border=10)
btn_sizer = wx.BoxSizer(wx.HORIZONTAL)
btn_copy_cf = wx.Button(dlg, label="Copy connection file path")
btn_copy_cf.Bind(wx.EVT_BUTTON, lambda _: _copy_to_clipboard(cf))
cmd = f'jupyter console --existing "{cf}"'
btn_copy_cmd = wx.Button(dlg, label="Copy console command")
btn_copy_cmd.Bind(wx.EVT_BUTTON, lambda _: _copy_to_clipboard(cmd))
btn_ok = wx.Button(dlg, wx.ID_OK, label="Close")
btn_ok.SetDefault()
btn_sizer.Add(btn_copy_cf, flag=wx.RIGHT, border=6)
btn_sizer.Add(btn_copy_cmd, flag=wx.RIGHT, border=6)
btn_sizer.AddStretchSpacer()
btn_sizer.Add(btn_ok)
sizer.Add(btn_sizer, flag=wx.EXPAND | wx.ALL, border=10)
dlg.SetSizer(sizer)
dlg.Fit()
dlg.ShowModal()
dlg.Destroy()
[docs]
def _copy_to_clipboard(text: str) -> None:
import wx
if wx.TheClipboard.Open():
wx.TheClipboard.SetData(wx.TextDataObject(text))
wx.TheClipboard.Close()