Nice overview! I've got the same issues, except I don't mind a separate QtConsole so much. The answer and comments here gave a lot of ideas to try out, but in the end I found it easiest to setup a basic neovim lua plugin.
Basically:
- The lua plugin starts a Python process.
- The Python process imports relevant jupyter logic, including the monkey patch to fix the scrolling behavior for plotting, and keeps a link to the Jupyter kernel.
- Next, it just listens to stdin. Since we want to send more than one line, we accumulate tokens until we find a record separator.
- The lua sends either cells or a visual selection to the Python process stdin, terminated by the aforementioned record separator.
local M = {}
local RECORD_SEPARATOR = string.char(0x1E) -- ASCII Record Separator character
M.config = {
python_script_path = vim.fn.stdpath 'config' .. '/lua/custom/plugins/qtrepl/server.py',
job_id = nil,
}
-- Set up the module with optional configuration
function M.setup(opts)
-- Merge provided options with default config
opts = opts or {}
if opts.python_script_path and not vim.fn.filereadable(opts.python_script_path) then
print('Warning: Python script not found at ' .. opts.python_script_path)
end
M.config = vim.tbl_extend('force', M.config, opts)
end
-- Start the Jupyter bridge if not already running
function M.start_jupyter_bridge()
print 'starting bridge'
if M.config.job_id == nil then
local path = M.config.python_script_path
local job_id, err = vim.fn.jobstart({ 'python', path }, {
on_stdout = function(_, data)
if data and data[1] ~= '' then
local success, result = pcall(vim.fn.json_decode, data[1])
if success then
if result.status then
print('Execution status: ' .. result.status)
elseif result.error then
print('Error: ' .. result.error)
end
else
print('Failed to decode JSON: ' .. data[1])
end
end
end,
on_stderr = function(_, data)
if data and data[1] ~= '' then
print('Error from Python script: ' .. table.concat(data, '\n'))
end
end,
on_exit = function(_, exit_code)
print('Python script exited with code: ' .. exit_code)
M.config.job_id = nil
end,
})
if job_id > 0 then
M.config.job_id = job_id
print('Jupyter bridge started successfully with job ID: ' .. job_id)
else
print('Failed to start Jupyter bridge: ' .. (err or 'Unknown error'))
end
end
end
function M.send_to_repl(code)
-- Start the bridge if not already running
if M.config.job_id == nil then
M.start_jupyter_bridge()
end
if type(code) == 'table' then
code = table.concat(code, '\n')
end
print 'sending'
vim.fn.chansend(M.config.job_id, code .. RECORD_SEPARATOR)
end
-- Get the currently selected text in visual mode
function M.get_visual_selection()
-- Get start and end positions of the selection
local _, start_row, start_col, _ = unpack(vim.fn.getpos "'<")
local _, end_row, end_col, _ = unpack(vim.fn.getpos "'>")
-- Adjust for Lua's 1-based indexing
start_row, start_col, end_row, end_col = start_row - 1, start_col - 1, end_row - 1, end_col
if vim.fn.mode() == 'V' then
-- Line-wise visual mode
return vim.api.nvim_buf_get_lines(0, start_row, end_row + 1, false)
elseif vim.fn.mode() == 'v' then
-- Character-wise visual mode
if start_row < end_row or (start_row == end_row and start_col <= end_col) then
return vim.api.nvim_buf_get_text(0, start_row, start_col, end_row, end_col, {})
else
return vim.api.nvim_buf_get_text(0, end_row, end_col, start_row, start_col, {})
end
end
-- Return an empty table if not in visual mode
return {}
end
-- Get the selection defined by # %% markers
function M.get_cell_selection()
local current_line = vim.fn.line '.'
local buffer_lines = vim.api.nvim_buf_get_lines(0, 0, -1, false)
local start_line, end_line
-- Find the start of the cell
for i = current_line, 1, -1 do
if buffer_lines[i] and buffer_lines[i]:match '^# %%%%' then
start_line = i + 1
break
end
end
-- Find the end of the cell
for i = current_line, #buffer_lines do
if buffer_lines[i] and buffer_lines[i]:match '^# %%%%' then
end_line = i - 1
break
end
end
-- If no end marker is found, use the last line of the buffer
end_line = end_line or #buffer_lines
-- If no start marker is found, use the first line of the buffer
start_line = start_line or 1
-- Move the cursor to the next cell
vim.api.nvim_win_set_cursor(0, { math.min(end_line + 2, #buffer_lines), 0 })
return vim.api.nvim_buf_get_lines(0, start_line - 1, end_line, false)
end
-- Run the selected lines in the Jupyter REPL
function M.run_selected_lines()
local code = M.get_visual_selection()
M.send_to_repl(code)
vim.cmd('normal! ' .. vim.api.nvim_replace_termcodes('<Esc>', true, false, true))
end
-- Run the current cell in the Jupyter REPL
function M.run_current_cell()
local code = M.get_cell_selection()
if code == nil then
return
end
M.send_to_repl(code)
end
return M
The Python part:
from pathlib import Path
import subprocess
import sys
import time
from jupyter_client.manager import start_new_kernel
RECORD_SEPARATOR = "\x1e" # ASCII Record Separator
def create_connection_file():
kernel_manager, kernel_client = start_new_kernel()
connection_file = Path(kernel_manager.connection_file)
return connection_file, kernel_client
def qtconsole_script(connection_file=None):
"""
Auto-scroll doesn't work for remote plots. See:
https://github.com/jupyter/qtconsole/issues/386#issuecomment-2099961049
"""
def autoscroll_console_widget(wrapped_func):
def _w(self, *args, **kwargs):
result = wrapped_func(self, *args, **kwargs)
self._scroll_to_end()
return result
return _w
from qtconsole.qtconsoleapp import JupyterQtConsoleApp
from qtconsole.console_widget import ConsoleWidget
ConsoleWidget._append_custom = autoscroll_console_widget(
ConsoleWidget._append_custom
)
app = JupyterQtConsoleApp()
if connection_file is None:
app.initialize()
else:
app.initialize(["--existing=" + connection_file])
app.start()
def run_qtconsole(connection_file):
this_file = Path(__file__)
subprocess.Popen(
[
sys.executable,
"-c",
f"import sys; sys.path.append('{this_file.parent.as_posix()}'); "
f"from {this_file.stem} import qtconsole_script; "
f"qtconsole_script('{connection_file.as_posix()}')",
]
)
def send_code_to_jupyter(client, code):
msg_id = client.execute(code)
reply = client.get_shell_msg(msg_id)
return reply["content"]["status"]
if __name__ == "__main__":
connection_file, kernel_client = create_connection_file()
run_qtconsole(connection_file)
# Wait a second for the console to start before we send.
time.sleep(1)
while True:
try:
# Read input until the record separator
code = ""
while True:
chunk = sys.stdin.read(1)
if chunk == RECORD_SEPARATOR:
break
code += chunk
if not code:
continue
# Execute the code
status = send_code_to_jupyter(kernel_client, code)
# Send the result back to Neovim
sys.stdout.flush()
except Exception as e:
sys.stdout.flush()
kernel_client.stop_channels()
kernel_client.shutdown()
A keymap or two:
local qtrepl_plugin = require 'custom.plugins.qtrepl'
vim.keymap.set('v', '<leader>x', function()
qtrepl_plugin.run_selected_lines()
end, { noremap = true, silent = true, desc = 'Run selected lines in QtConsole' })
vim.keymap.set('n', '<leader>x', function()
qtrepl_plugin.run_current_cell()
end, { noremap = true, silent = true, desc = 'Run current cell in QtConsole' })
It'll also need to jupyter config settings:
c.ConsoleWidget.include_other_output = True
c.HistoryConsoleWidget.include_other_output = True
I just start my (pixi) Python virtual env (which includes the jupyter qtconsole), and open nvim.
It's obviously not very fancy or super robust -- in particular, it's not very informative when e.g. the Python process doesn't start properly. But it's hopefully simple enough to keep in your head.
I'm guessing it should work for e.g. Julia as well via the IJulia package, since IJulia provides something similar as from jupyter_client.manager import start_new_kernel. For me, this has the benefit of not requiring Python in my env when I'm running Julia.
