--- /dev/null
+"""Main Nvim interface."""
+import os
+import sys
+import threading
+from functools import partial
+from traceback import format_stack
+
+from msgpack import ExtType
+
+from pynvim.api.buffer import Buffer
+from pynvim.api.common import (NvimError, Remote, RemoteApi, RemoteMap, RemoteSequence,
+ decode_if_bytes, walk)
+from pynvim.api.tabpage import Tabpage
+from pynvim.api.window import Window
+from pynvim.compat import IS_PYTHON3
+from pynvim.util import Version, format_exc_skip
+
+__all__ = ('Nvim')
+
+
+os_chdir = os.chdir
+
+lua_module = """
+local a = vim.api
+local function update_highlights(buf, src_id, hls, clear_first, clear_end)
+ if clear_first ~= nil then
+ a.nvim_buf_clear_highlight(buf, src_id, clear_first, clear_end)
+ end
+ for _,hl in pairs(hls) do
+ local group, line, col_start, col_end = unpack(hl)
+ if col_start == nil then
+ col_start = 0
+ end
+ if col_end == nil then
+ col_end = -1
+ end
+ a.nvim_buf_add_highlight(buf, src_id, group, line, col_start, col_end)
+ end
+end
+
+local chid = ...
+local mod = {update_highlights=update_highlights}
+_G["_pynvim_"..chid] = mod
+"""
+
+
+class Nvim(object):
+
+ """Class that represents a remote Nvim instance.
+
+ This class is main entry point to Nvim remote API, it is a wrapper
+ around Session instances.
+
+ The constructor of this class must not be called directly. Instead, the
+ `from_session` class method should be used to create the first instance
+ from a raw `Session` instance.
+
+ Subsequent instances for the same session can be created by calling the
+ `with_decode` instance method to change the decoding behavior or
+ `SubClass.from_nvim(nvim)` where `SubClass` is a subclass of `Nvim`, which
+ is useful for having multiple `Nvim` objects that behave differently
+ without one affecting the other.
+
+ When this library is used on python3.4+, asyncio event loop is guaranteed
+ to be used. It is available as the "loop" attribute of this class. Note
+ that asyncio callbacks cannot make blocking requests, which includes
+ accessing state-dependent attributes. They should instead schedule another
+ callback using nvim.async_call, which will not have this restriction.
+ """
+
+ @classmethod
+ def from_session(cls, session):
+ """Create a new Nvim instance for a Session instance.
+
+ This method must be called to create the first Nvim instance, since it
+ queries Nvim metadata for type information and sets a SessionHook for
+ creating specialized objects from Nvim remote handles.
+ """
+ session.error_wrapper = lambda e: NvimError(decode_if_bytes(e[1]))
+ channel_id, metadata = session.request(b'nvim_get_api_info')
+
+ if IS_PYTHON3:
+ # decode all metadata strings for python3
+ metadata = walk(decode_if_bytes, metadata)
+
+ types = {
+ metadata['types']['Buffer']['id']: Buffer,
+ metadata['types']['Window']['id']: Window,
+ metadata['types']['Tabpage']['id']: Tabpage,
+ }
+
+ return cls(session, channel_id, metadata, types)
+
+ @classmethod
+ def from_nvim(cls, nvim):
+ """Create a new Nvim instance from an existing instance."""
+ return cls(nvim._session, nvim.channel_id, nvim.metadata,
+ nvim.types, nvim._decode, nvim._err_cb)
+
+ def __init__(self, session, channel_id, metadata, types,
+ decode=False, err_cb=None):
+ """Initialize a new Nvim instance. This method is module-private."""
+ self._session = session
+ self.channel_id = channel_id
+ self.metadata = metadata
+ version = metadata.get("version", {"api_level": 0})
+ self.version = Version(**version)
+ self.types = types
+ self.api = RemoteApi(self, 'nvim_')
+ self.vars = RemoteMap(self, 'nvim_get_var', 'nvim_set_var', 'nvim_del_var')
+ self.vvars = RemoteMap(self, 'nvim_get_vvar', None, None)
+ self.options = RemoteMap(self, 'nvim_get_option', 'nvim_set_option')
+ self.buffers = Buffers(self)
+ self.windows = RemoteSequence(self, 'nvim_list_wins')
+ self.tabpages = RemoteSequence(self, 'nvim_list_tabpages')
+ self.current = Current(self)
+ self.session = CompatibilitySession(self)
+ self.funcs = Funcs(self)
+ self.lua = LuaFuncs(self)
+ self.error = NvimError
+ self._decode = decode
+ self._err_cb = err_cb
+
+ # only on python3.4+ we expose asyncio
+ if IS_PYTHON3:
+ self.loop = self._session.loop._loop
+
+ def _from_nvim(self, obj, decode=None):
+ if decode is None:
+ decode = self._decode
+ if type(obj) is ExtType:
+ cls = self.types[obj.code]
+ return cls(self, (obj.code, obj.data))
+ if decode:
+ obj = decode_if_bytes(obj, decode)
+ return obj
+
+ def _to_nvim(self, obj):
+ if isinstance(obj, Remote):
+ return ExtType(*obj.code_data)
+ return obj
+
+ def _get_lua_private(self):
+ if not getattr(self._session, "_has_lua", False):
+ self.exec_lua(lua_module, self.channel_id)
+ self._session._has_lua = True
+ return getattr(self.lua, "_pynvim_{}".format(self.channel_id))
+
+ def request(self, name, *args, **kwargs):
+ r"""Send an API request or notification to nvim.
+
+ It is rarely needed to call this function directly, as most API
+ functions have python wrapper functions. The `api` object can
+ be also be used to call API functions as methods:
+
+ vim.api.err_write('ERROR\n', async_=True)
+ vim.current.buffer.api.get_mark('.')
+
+ is equivalent to
+
+ vim.request('nvim_err_write', 'ERROR\n', async_=True)
+ vim.request('nvim_buf_get_mark', vim.current.buffer, '.')
+
+
+ Normally a blocking request will be sent. If the `async_` flag is
+ present and True, a asynchronous notification is sent instead. This
+ will never block, and the return value or error is ignored.
+ """
+ if (self._session._loop_thread is not None
+ and threading.current_thread() != self._session._loop_thread):
+
+ msg = ("Request from non-main thread.\n"
+ "Requests from different threads should be wrapped "
+ "with nvim.async_call(cb, ...) \n{}\n"
+ .format('\n'.join(format_stack(None, 5)[:-1])))
+
+ self.async_call(self._err_cb, msg)
+ raise NvimError("request from non-main thread")
+
+ decode = kwargs.pop('decode', self._decode)
+ args = walk(self._to_nvim, args)
+ res = self._session.request(name, *args, **kwargs)
+ return walk(self._from_nvim, res, decode=decode)
+
+ def next_message(self):
+ """Block until a message(request or notification) is available.
+
+ If any messages were previously enqueued, return the first in queue.
+ If not, run the event loop until one is received.
+ """
+ msg = self._session.next_message()
+ if msg:
+ return walk(self._from_nvim, msg)
+
+ def run_loop(self, request_cb, notification_cb,
+ setup_cb=None, err_cb=None):
+ """Run the event loop to receive requests and notifications from Nvim.
+
+ This should not be called from a plugin running in the host, which
+ already runs the loop and dispatches events to plugins.
+ """
+ if err_cb is None:
+ err_cb = sys.stderr.write
+ self._err_cb = err_cb
+
+ def filter_request_cb(name, args):
+ name = self._from_nvim(name)
+ args = walk(self._from_nvim, args)
+ try:
+ result = request_cb(name, args)
+ except Exception:
+ msg = ("error caught in request handler '{} {}'\n{}\n\n"
+ .format(name, args, format_exc_skip(1)))
+ self._err_cb(msg)
+ raise
+ return walk(self._to_nvim, result)
+
+ def filter_notification_cb(name, args):
+ name = self._from_nvim(name)
+ args = walk(self._from_nvim, args)
+ try:
+ notification_cb(name, args)
+ except Exception:
+ msg = ("error caught in notification handler '{} {}'\n{}\n\n"
+ .format(name, args, format_exc_skip(1)))
+ self._err_cb(msg)
+ raise
+
+ self._session.run(filter_request_cb, filter_notification_cb, setup_cb)
+
+ def stop_loop(self):
+ """Stop the event loop being started with `run_loop`."""
+ self._session.stop()
+
+ def close(self):
+ """Close the nvim session and release its resources."""
+ self._session.close()
+
+ def __enter__(self):
+ """Enter nvim session as a context manager."""
+ return self
+
+ def __exit__(self, *exc_info):
+ """Exit nvim session as a context manager.
+
+ Closes the event loop.
+ """
+ self.close()
+
+ def with_decode(self, decode=True):
+ """Initialize a new Nvim instance."""
+ return Nvim(self._session, self.channel_id,
+ self.metadata, self.types, decode, self._err_cb)
+
+ def ui_attach(self, width, height, rgb=None, **kwargs):
+ """Register as a remote UI.
+
+ After this method is called, the client will receive redraw
+ notifications.
+ """
+ options = kwargs
+ if rgb is not None:
+ options['rgb'] = rgb
+ return self.request('nvim_ui_attach', width, height, options)
+
+ def ui_detach(self):
+ """Unregister as a remote UI."""
+ return self.request('nvim_ui_detach')
+
+ def ui_try_resize(self, width, height):
+ """Notify nvim that the client window has resized.
+
+ If possible, nvim will send a redraw request to resize.
+ """
+ return self.request('ui_try_resize', width, height)
+
+ def subscribe(self, event):
+ """Subscribe to a Nvim event."""
+ return self.request('nvim_subscribe', event)
+
+ def unsubscribe(self, event):
+ """Unsubscribe to a Nvim event."""
+ return self.request('nvim_unsubscribe', event)
+
+ def command(self, string, **kwargs):
+ """Execute a single ex command."""
+ return self.request('nvim_command', string, **kwargs)
+
+ def command_output(self, string):
+ """Execute a single ex command and return the output."""
+ return self.request('nvim_command_output', string)
+
+ def eval(self, string, **kwargs):
+ """Evaluate a vimscript expression."""
+ return self.request('nvim_eval', string, **kwargs)
+
+ def call(self, name, *args, **kwargs):
+ """Call a vimscript function."""
+ return self.request('nvim_call_function', name, args, **kwargs)
+
+ def exec_lua(self, code, *args, **kwargs):
+ """Execute lua code.
+
+ Additional parameters are available as `...` inside the lua chunk.
+ Only statements are executed. To evaluate an expression, prefix it
+ with `return`: `return my_function(...)`
+
+ There is a shorthand syntax to call lua functions with arguments:
+
+ nvim.lua.func(1,2)
+ nvim.lua.mymod.myfunction(data, async_=True)
+
+ is equivalent to
+
+ nvim.exec_lua("return func(...)", 1, 2)
+ nvim.exec_lua("mymod.myfunction(...)", data, async_=True)
+
+ Note that with `async_=True` there is no return value.
+ """
+ return self.request('nvim_execute_lua', code, args, **kwargs)
+
+ def strwidth(self, string):
+ """Return the number of display cells `string` occupies.
+
+ Tab is counted as one cell.
+ """
+ return self.request('nvim_strwidth', string)
+
+ def list_runtime_paths(self):
+ """Return a list of paths contained in the 'runtimepath' option."""
+ return self.request('nvim_list_runtime_paths')
+
+ def foreach_rtp(self, cb):
+ """Invoke `cb` for each path in 'runtimepath'.
+
+ Call the given callable for each path in 'runtimepath' until either
+ callable returns something but None, the exception is raised or there
+ are no longer paths. If stopped in case callable returned non-None,
+ vim.foreach_rtp function returns the value returned by callable.
+ """
+ for path in self.request('nvim_list_runtime_paths'):
+ try:
+ if cb(path) is not None:
+ break
+ except Exception:
+ break
+
+ def chdir(self, dir_path):
+ """Run os.chdir, then all appropriate vim stuff."""
+ os_chdir(dir_path)
+ return self.request('nvim_set_current_dir', dir_path)
+
+ def feedkeys(self, keys, options='', escape_csi=True):
+ """Push `keys` to Nvim user input buffer.
+
+ Options can be a string with the following character flags:
+ - 'm': Remap keys. This is default.
+ - 'n': Do not remap keys.
+ - 't': Handle keys as if typed; otherwise they are handled as if coming
+ from a mapping. This matters for undo, opening folds, etc.
+ """
+ return self.request('nvim_feedkeys', keys, options, escape_csi)
+
+ def input(self, bytes):
+ """Push `bytes` to Nvim low level input buffer.
+
+ Unlike `feedkeys()`, this uses the lowest level input buffer and the
+ call is not deferred. It returns the number of bytes actually
+ written(which can be less than what was requested if the buffer is
+ full).
+ """
+ return self.request('nvim_input', bytes)
+
+ def replace_termcodes(self, string, from_part=False, do_lt=True,
+ special=True):
+ r"""Replace any terminal code strings by byte sequences.
+
+ The returned sequences are Nvim's internal representation of keys,
+ for example:
+
+ <esc> -> '\x1b'
+ <cr> -> '\r'
+ <c-l> -> '\x0c'
+ <up> -> '\x80ku'
+
+ The returned sequences can be used as input to `feedkeys`.
+ """
+ return self.request('nvim_replace_termcodes', string,
+ from_part, do_lt, special)
+
+ def out_write(self, msg, **kwargs):
+ r"""Print `msg` as a normal message.
+
+ The message is buffered (won't display) until linefeed ("\n").
+ """
+ return self.request('nvim_out_write', msg, **kwargs)
+
+ def err_write(self, msg, **kwargs):
+ r"""Print `msg` as an error message.
+
+ The message is buffered (won't display) until linefeed ("\n").
+ """
+ if self._thread_invalid():
+ # special case: if a non-main thread writes to stderr
+ # i.e. due to an uncaught exception, pass it through
+ # without raising an additional exception.
+ self.async_call(self.err_write, msg, **kwargs)
+ return
+ return self.request('nvim_err_write', msg, **kwargs)
+
+ def _thread_invalid(self):
+ return (self._session._loop_thread is not None
+ and threading.current_thread() != self._session._loop_thread)
+
+ def quit(self, quit_command='qa!'):
+ """Send a quit command to Nvim.
+
+ By default, the quit command is 'qa!' which will make Nvim quit without
+ saving anything.
+ """
+ try:
+ self.command(quit_command)
+ except OSError:
+ # sending a quit command will raise an IOError because the
+ # connection is closed before a response is received. Safe to
+ # ignore it.
+ pass
+
+ def new_highlight_source(self):
+ """Return new src_id for use with Buffer.add_highlight."""
+ return self.current.buffer.add_highlight("", 0, src_id=0)
+
+ def async_call(self, fn, *args, **kwargs):
+ """Schedule `fn` to be called by the event loop soon.
+
+ This function is thread-safe, and is the only way code not
+ on the main thread could interact with nvim api objects.
+
+ This function can also be called in a synchronous
+ event handler, just before it returns, to defer execution
+ that shouldn't block neovim.
+ """
+ call_point = ''.join(format_stack(None, 5)[:-1])
+
+ def handler():
+ try:
+ fn(*args, **kwargs)
+ except Exception as err:
+ msg = ("error caught while executing async callback:\n"
+ "{!r}\n{}\n \nthe call was requested at\n{}"
+ .format(err, format_exc_skip(1), call_point))
+ self._err_cb(msg)
+ raise
+ self._session.threadsafe_call(handler)
+
+
+class Buffers(object):
+
+ """Remote NVim buffers.
+
+ Currently the interface for interacting with remote NVim buffers is the
+ `nvim_list_bufs` msgpack-rpc function. Most methods fetch the list of
+ buffers from NVim.
+
+ Conforms to *python-buffers*.
+ """
+
+ def __init__(self, nvim):
+ """Initialize a Buffers object with Nvim object `nvim`."""
+ self._fetch_buffers = nvim.api.list_bufs
+
+ def __len__(self):
+ """Return the count of buffers."""
+ return len(self._fetch_buffers())
+
+ def __getitem__(self, number):
+ """Return the Buffer object matching buffer number `number`."""
+ for b in self._fetch_buffers():
+ if b.number == number:
+ return b
+ raise KeyError(number)
+
+ def __contains__(self, b):
+ """Return whether Buffer `b` is a known valid buffer."""
+ return isinstance(b, Buffer) and b.valid
+
+ def __iter__(self):
+ """Return an iterator over the list of buffers."""
+ return iter(self._fetch_buffers())
+
+
+class CompatibilitySession(object):
+
+ """Helper class for API compatibility."""
+
+ def __init__(self, nvim):
+ self.threadsafe_call = nvim.async_call
+
+
+class Current(object):
+
+ """Helper class for emulating vim.current from python-vim."""
+
+ def __init__(self, session):
+ self._session = session
+ self.range = None
+
+ @property
+ def line(self):
+ return self._session.request('nvim_get_current_line')
+
+ @line.setter
+ def line(self, line):
+ return self._session.request('nvim_set_current_line', line)
+
+ @line.deleter
+ def line(self):
+ return self._session.request('nvim_del_current_line')
+
+ @property
+ def buffer(self):
+ return self._session.request('nvim_get_current_buf')
+
+ @buffer.setter
+ def buffer(self, buffer):
+ return self._session.request('nvim_set_current_buf', buffer)
+
+ @property
+ def window(self):
+ return self._session.request('nvim_get_current_win')
+
+ @window.setter
+ def window(self, window):
+ return self._session.request('nvim_set_current_win', window)
+
+ @property
+ def tabpage(self):
+ return self._session.request('nvim_get_current_tabpage')
+
+ @tabpage.setter
+ def tabpage(self, tabpage):
+ return self._session.request('nvim_set_current_tabpage', tabpage)
+
+
+class Funcs(object):
+
+ """Helper class for functional vimscript interface."""
+
+ def __init__(self, nvim):
+ self._nvim = nvim
+
+ def __getattr__(self, name):
+ return partial(self._nvim.call, name)
+
+
+class LuaFuncs(object):
+
+ """Wrapper to allow lua functions to be called like python methods."""
+
+ def __init__(self, nvim, name=""):
+ self._nvim = nvim
+ self.name = name
+
+ def __getattr__(self, name):
+ """Return wrapper to named api method."""
+ prefix = self.name + "." if self.name else ""
+ return LuaFuncs(self._nvim, prefix + name)
+
+ def __call__(self, *args, **kwargs):
+ # first new function after keyword rename, be a bit noisy
+ if 'async' in kwargs:
+ raise ValueError('"async" argument is not allowed. '
+ 'Use "async_" instead.')
+ async_ = kwargs.get('async_', False)
+ pattern = "return {}(...)" if not async_ else "{}(...)"
+ code = pattern.format(self.name)
+ return self._nvim.exec_lua(code, *args, **kwargs)
projects / dotfiles / .git / blobdiff