.local/lib/python3.9/site-packages/pynvim/api/nvim.py

   1 """Main Nvim interface."""
   2 import os
   3 import sys
   4 import threading
   5 from functools import partial
   6 from traceback import format_stack
   7
   8 from msgpack import ExtType
   9
  10 from pynvim.api.buffer import Buffer
  11 from pynvim.api.common import (NvimError, Remote, RemoteApi, RemoteMap, RemoteSequence,
  12                                decode_if_bytes, walk)
  13 from pynvim.api.tabpage import Tabpage
  14 from pynvim.api.window import Window
  15 from pynvim.compat import IS_PYTHON3
  16 from pynvim.util import Version, format_exc_skip
  17
  18 __all__ = ('Nvim')
  19
  20
  21 os_chdir = os.chdir
  22
  23 lua_module = """
  24 local a = vim.api
  25 local function update_highlights(buf, src_id, hls, clear_first, clear_end)
  26   if clear_first ~= nil then
  27       a.nvim_buf_clear_highlight(buf, src_id, clear_first, clear_end)
  28   end
  29   for _,hl in pairs(hls) do
  30     local group, line, col_start, col_end = unpack(hl)
  31     if col_start == nil then
  32       col_start = 0
  33     end
  34     if col_end == nil then
  35       col_end = -1
  36     end
  37     a.nvim_buf_add_highlight(buf, src_id, group, line, col_start, col_end)
  38   end
  39 end
  40
  41 local chid = ...
  42 local mod = {update_highlights=update_highlights}
  43 _G["_pynvim_"..chid] = mod
  44 """
  45
  46
  47 class Nvim(object):
  48
  49     """Class that represents a remote Nvim instance.
  50
  51     This class is main entry point to Nvim remote API, it is a wrapper
  52     around Session instances.
  53
  54     The constructor of this class must not be called directly. Instead, the
  55     `from_session` class method should be used to create the first instance
  56     from a raw `Session` instance.
  57
  58     Subsequent instances for the same session can be created by calling the
  59     `with_decode` instance method to change the decoding behavior or
  60     `SubClass.from_nvim(nvim)` where `SubClass` is a subclass of `Nvim`, which
  61     is useful for having multiple `Nvim` objects that behave differently
  62     without one affecting the other.
  63
  64     When this library is used on python3.4+, asyncio event loop is guaranteed
  65     to be used. It is available as the "loop" attribute of this class. Note
  66     that asyncio callbacks cannot make blocking requests, which includes
  67     accessing state-dependent attributes. They should instead schedule another
  68     callback using nvim.async_call, which will not have this restriction.
  69     """
  70
  71     @classmethod
  72     def from_session(cls, session):
  73         """Create a new Nvim instance for a Session instance.
  74
  75         This method must be called to create the first Nvim instance, since it
  76         queries Nvim metadata for type information and sets a SessionHook for
  77         creating specialized objects from Nvim remote handles.
  78         """
  79         session.error_wrapper = lambda e: NvimError(decode_if_bytes(e[1]))
  80         channel_id, metadata = session.request(b'nvim_get_api_info')
  81
  82         if IS_PYTHON3:
  83             # decode all metadata strings for python3
  84             metadata = walk(decode_if_bytes, metadata)
  85
  86         types = {
  87             metadata['types']['Buffer']['id']: Buffer,
  88             metadata['types']['Window']['id']: Window,
  89             metadata['types']['Tabpage']['id']: Tabpage,
  90         }
  91
  92         return cls(session, channel_id, metadata, types)
  93
  94     @classmethod
  95     def from_nvim(cls, nvim):
  96         """Create a new Nvim instance from an existing instance."""
  97         return cls(nvim._session, nvim.channel_id, nvim.metadata,
  98                    nvim.types, nvim._decode, nvim._err_cb)
  99
 100     def __init__(self, session, channel_id, metadata, types,
 101                  decode=False, err_cb=None):
 102         """Initialize a new Nvim instance. This method is module-private."""
 103         self._session = session
 104         self.channel_id = channel_id
 105         self.metadata = metadata
 106         version = metadata.get("version", {"api_level": 0})
 107         self.version = Version(**version)
 108         self.types = types
 109         self.api = RemoteApi(self, 'nvim_')
 110         self.vars = RemoteMap(self, 'nvim_get_var', 'nvim_set_var', 'nvim_del_var')
 111         self.vvars = RemoteMap(self, 'nvim_get_vvar', None, None)
 112         self.options = RemoteMap(self, 'nvim_get_option', 'nvim_set_option')
 113         self.buffers = Buffers(self)
 114         self.windows = RemoteSequence(self, 'nvim_list_wins')
 115         self.tabpages = RemoteSequence(self, 'nvim_list_tabpages')
 116         self.current = Current(self)
 117         self.session = CompatibilitySession(self)
 118         self.funcs = Funcs(self)
 119         self.lua = LuaFuncs(self)
 120         self.error = NvimError
 121         self._decode = decode
 122         self._err_cb = err_cb
 123
 124         # only on python3.4+ we expose asyncio
 125         if IS_PYTHON3:
 126             self.loop = self._session.loop._loop
 127
 128     def _from_nvim(self, obj, decode=None):
 129         if decode is None:
 130             decode = self._decode
 131         if type(obj) is ExtType:
 132             cls = self.types[obj.code]
 133             return cls(self, (obj.code, obj.data))
 134         if decode:
 135             obj = decode_if_bytes(obj, decode)
 136         return obj
 137
 138     def _to_nvim(self, obj):
 139         if isinstance(obj, Remote):
 140             return ExtType(*obj.code_data)
 141         return obj
 142
 143     def _get_lua_private(self):
 144         if not getattr(self._session, "_has_lua", False):
 145             self.exec_lua(lua_module, self.channel_id)
 146             self._session._has_lua = True
 147         return getattr(self.lua, "_pynvim_{}".format(self.channel_id))
 148
 149     def request(self, name, *args, **kwargs):
 150         r"""Send an API request or notification to nvim.
 151
 152         It is rarely needed to call this function directly, as most API
 153         functions have python wrapper functions. The `api` object can
 154         be also be used to call API functions as methods:
 155
 156             vim.api.err_write('ERROR\n', async_=True)
 157             vim.current.buffer.api.get_mark('.')
 158
 159         is equivalent to
 160
 161             vim.request('nvim_err_write', 'ERROR\n', async_=True)
 162             vim.request('nvim_buf_get_mark', vim.current.buffer, '.')
 163
 164
 165         Normally a blocking request will be sent.  If the `async_` flag is
 166         present and True, a asynchronous notification is sent instead. This
 167         will never block, and the return value or error is ignored.
 168         """
 169         if (self._session._loop_thread is not None
 170                 and threading.current_thread() != self._session._loop_thread):
 171
 172             msg = ("Request from non-main thread.\n"
 173                    "Requests from different threads should be wrapped "
 174                    "with nvim.async_call(cb, ...) \n{}\n"
 175                    .format('\n'.join(format_stack(None, 5)[:-1])))
 176
 177             self.async_call(self._err_cb, msg)
 178             raise NvimError("request from non-main thread")
 179
 180         decode = kwargs.pop('decode', self._decode)
 181         args = walk(self._to_nvim, args)
 182         res = self._session.request(name, *args, **kwargs)
 183         return walk(self._from_nvim, res, decode=decode)
 184
 185     def next_message(self):
 186         """Block until a message(request or notification) is available.
 187
 188         If any messages were previously enqueued, return the first in queue.
 189         If not, run the event loop until one is received.
 190         """
 191         msg = self._session.next_message()
 192         if msg:
 193             return walk(self._from_nvim, msg)
 194
 195     def run_loop(self, request_cb, notification_cb,
 196                  setup_cb=None, err_cb=None):
 197         """Run the event loop to receive requests and notifications from Nvim.
 198
 199         This should not be called from a plugin running in the host, which
 200         already runs the loop and dispatches events to plugins.
 201         """
 202         if err_cb is None:
 203             err_cb = sys.stderr.write
 204         self._err_cb = err_cb
 205
 206         def filter_request_cb(name, args):
 207             name = self._from_nvim(name)
 208             args = walk(self._from_nvim, args)
 209             try:
 210                 result = request_cb(name, args)
 211             except Exception:
 212                 msg = ("error caught in request handler '{} {}'\n{}\n\n"
 213                        .format(name, args, format_exc_skip(1)))
 214                 self._err_cb(msg)
 215                 raise
 216             return walk(self._to_nvim, result)
 217
 218         def filter_notification_cb(name, args):
 219             name = self._from_nvim(name)
 220             args = walk(self._from_nvim, args)
 221             try:
 222                 notification_cb(name, args)
 223             except Exception:
 224                 msg = ("error caught in notification handler '{} {}'\n{}\n\n"
 225                        .format(name, args, format_exc_skip(1)))
 226                 self._err_cb(msg)
 227                 raise
 228
 229         self._session.run(filter_request_cb, filter_notification_cb, setup_cb)
 230
 231     def stop_loop(self):
 232         """Stop the event loop being started with `run_loop`."""
 233         self._session.stop()
 234
 235     def close(self):
 236         """Close the nvim session and release its resources."""
 237         self._session.close()
 238
 239     def __enter__(self):
 240         """Enter nvim session as a context manager."""
 241         return self
 242
 243     def __exit__(self, *exc_info):
 244         """Exit nvim session as a context manager.
 245
 246         Closes the event loop.
 247         """
 248         self.close()
 249
 250     def with_decode(self, decode=True):
 251         """Initialize a new Nvim instance."""
 252         return Nvim(self._session, self.channel_id,
 253                     self.metadata, self.types, decode, self._err_cb)
 254
 255     def ui_attach(self, width, height, rgb=None, **kwargs):
 256         """Register as a remote UI.
 257
 258         After this method is called, the client will receive redraw
 259         notifications.
 260         """
 261         options = kwargs
 262         if rgb is not None:
 263             options['rgb'] = rgb
 264         return self.request('nvim_ui_attach', width, height, options)
 265
 266     def ui_detach(self):
 267         """Unregister as a remote UI."""
 268         return self.request('nvim_ui_detach')
 269
 270     def ui_try_resize(self, width, height):
 271         """Notify nvim that the client window has resized.
 272
 273         If possible, nvim will send a redraw request to resize.
 274         """
 275         return self.request('ui_try_resize', width, height)
 276
 277     def subscribe(self, event):
 278         """Subscribe to a Nvim event."""
 279         return self.request('nvim_subscribe', event)
 280
 281     def unsubscribe(self, event):
 282         """Unsubscribe to a Nvim event."""
 283         return self.request('nvim_unsubscribe', event)
 284
 285     def command(self, string, **kwargs):
 286         """Execute a single ex command."""
 287         return self.request('nvim_command', string, **kwargs)
 288
 289     def command_output(self, string):
 290         """Execute a single ex command and return the output."""
 291         return self.request('nvim_command_output', string)
 292
 293     def eval(self, string, **kwargs):
 294         """Evaluate a vimscript expression."""
 295         return self.request('nvim_eval', string, **kwargs)
 296
 297     def call(self, name, *args, **kwargs):
 298         """Call a vimscript function."""
 299         return self.request('nvim_call_function', name, args, **kwargs)
 300
 301     def exec_lua(self, code, *args, **kwargs):
 302         """Execute lua code.
 303
 304         Additional parameters are available as `...` inside the lua chunk.
 305         Only statements are executed.  To evaluate an expression, prefix it
 306         with `return`: `return my_function(...)`
 307
 308         There is a shorthand syntax to call lua functions with arguments:
 309
 310             nvim.lua.func(1,2)
 311             nvim.lua.mymod.myfunction(data, async_=True)
 312
 313         is equivalent to
 314
 315             nvim.exec_lua("return func(...)", 1, 2)
 316             nvim.exec_lua("mymod.myfunction(...)", data, async_=True)
 317
 318         Note that with `async_=True` there is no return value.
 319         """
 320         return self.request('nvim_execute_lua', code, args, **kwargs)
 321
 322     def strwidth(self, string):
 323         """Return the number of display cells `string` occupies.
 324
 325         Tab is counted as one cell.
 326         """
 327         return self.request('nvim_strwidth', string)
 328
 329     def list_runtime_paths(self):
 330         """Return a list of paths contained in the 'runtimepath' option."""
 331         return self.request('nvim_list_runtime_paths')
 332
 333     def foreach_rtp(self, cb):
 334         """Invoke `cb` for each path in 'runtimepath'.
 335
 336         Call the given callable for each path in 'runtimepath' until either
 337         callable returns something but None, the exception is raised or there
 338         are no longer paths. If stopped in case callable returned non-None,
 339         vim.foreach_rtp function returns the value returned by callable.
 340         """
 341         for path in self.request('nvim_list_runtime_paths'):
 342             try:
 343                 if cb(path) is not None:
 344                     break
 345             except Exception:
 346                 break
 347
 348     def chdir(self, dir_path):
 349         """Run os.chdir, then all appropriate vim stuff."""
 350         os_chdir(dir_path)
 351         return self.request('nvim_set_current_dir', dir_path)
 352
 353     def feedkeys(self, keys, options='', escape_csi=True):
 354         """Push `keys` to Nvim user input buffer.
 355
 356         Options can be a string with the following character flags:
 357         - 'm': Remap keys. This is default.
 358         - 'n': Do not remap keys.
 359         - 't': Handle keys as if typed; otherwise they are handled as if coming
 360                from a mapping. This matters for undo, opening folds, etc.
 361         """
 362         return self.request('nvim_feedkeys', keys, options, escape_csi)
 363
 364     def input(self, bytes):
 365         """Push `bytes` to Nvim low level input buffer.
 366
 367         Unlike `feedkeys()`, this uses the lowest level input buffer and the
 368         call is not deferred. It returns the number of bytes actually
 369         written(which can be less than what was requested if the buffer is
 370         full).
 371         """
 372         return self.request('nvim_input', bytes)
 373
 374     def replace_termcodes(self, string, from_part=False, do_lt=True,
 375                           special=True):
 376         r"""Replace any terminal code strings by byte sequences.
 377
 378         The returned sequences are Nvim's internal representation of keys,
 379         for example:
 380
 381         <esc> -> '\x1b'
 382         <cr>  -> '\r'
 383         <c-l> -> '\x0c'
 384         <up>  -> '\x80ku'
 385
 386         The returned sequences can be used as input to `feedkeys`.
 387         """
 388         return self.request('nvim_replace_termcodes', string,
 389                             from_part, do_lt, special)
 390
 391     def out_write(self, msg, **kwargs):
 392         r"""Print `msg` as a normal message.
 393
 394         The message is buffered (won't display) until linefeed ("\n").
 395         """
 396         return self.request('nvim_out_write', msg, **kwargs)
 397
 398     def err_write(self, msg, **kwargs):
 399         r"""Print `msg` as an error message.
 400
 401         The message is buffered (won't display) until linefeed ("\n").
 402         """
 403         if self._thread_invalid():
 404             # special case: if a non-main thread writes to stderr
 405             # i.e. due to an uncaught exception, pass it through
 406             # without raising an additional exception.
 407             self.async_call(self.err_write, msg, **kwargs)
 408             return
 409         return self.request('nvim_err_write', msg, **kwargs)
 410
 411     def _thread_invalid(self):
 412         return (self._session._loop_thread is not None
 413                 and threading.current_thread() != self._session._loop_thread)
 414
 415     def quit(self, quit_command='qa!'):
 416         """Send a quit command to Nvim.
 417
 418         By default, the quit command is 'qa!' which will make Nvim quit without
 419         saving anything.
 420         """
 421         try:
 422             self.command(quit_command)
 423         except OSError:
 424             # sending a quit command will raise an IOError because the
 425             # connection is closed before a response is received. Safe to
 426             # ignore it.
 427             pass
 428
 429     def new_highlight_source(self):
 430         """Return new src_id for use with Buffer.add_highlight."""
 431         return self.current.buffer.add_highlight("", 0, src_id=0)
 432
 433     def async_call(self, fn, *args, **kwargs):
 434         """Schedule `fn` to be called by the event loop soon.
 435
 436         This function is thread-safe, and is the only way code not
 437         on the main thread could interact with nvim api objects.
 438
 439         This function can also be called in a synchronous
 440         event handler, just before it returns, to defer execution
 441         that shouldn't block neovim.
 442         """
 443         call_point = ''.join(format_stack(None, 5)[:-1])
 444
 445         def handler():
 446             try:
 447                 fn(*args, **kwargs)
 448             except Exception as err:
 449                 msg = ("error caught while executing async callback:\n"
 450                        "{!r}\n{}\n \nthe call was requested at\n{}"
 451                        .format(err, format_exc_skip(1), call_point))
 452                 self._err_cb(msg)
 453                 raise
 454         self._session.threadsafe_call(handler)
 455
 456
 457 class Buffers(object):
 458
 459     """Remote NVim buffers.
 460
 461     Currently the interface for interacting with remote NVim buffers is the
 462     `nvim_list_bufs` msgpack-rpc function. Most methods fetch the list of
 463     buffers from NVim.
 464
 465     Conforms to *python-buffers*.
 466     """
 467
 468     def __init__(self, nvim):
 469         """Initialize a Buffers object with Nvim object `nvim`."""
 470         self._fetch_buffers = nvim.api.list_bufs
 471
 472     def __len__(self):
 473         """Return the count of buffers."""
 474         return len(self._fetch_buffers())
 475
 476     def __getitem__(self, number):
 477         """Return the Buffer object matching buffer number `number`."""
 478         for b in self._fetch_buffers():
 479             if b.number == number:
 480                 return b
 481         raise KeyError(number)
 482
 483     def __contains__(self, b):
 484         """Return whether Buffer `b` is a known valid buffer."""
 485         return isinstance(b, Buffer) and b.valid
 486
 487     def __iter__(self):
 488         """Return an iterator over the list of buffers."""
 489         return iter(self._fetch_buffers())
 490
 491
 492 class CompatibilitySession(object):
 493
 494     """Helper class for API compatibility."""
 495
 496     def __init__(self, nvim):
 497         self.threadsafe_call = nvim.async_call
 498
 499
 500 class Current(object):
 501
 502     """Helper class for emulating vim.current from python-vim."""
 503
 504     def __init__(self, session):
 505         self._session = session
 506         self.range = None
 507
 508     @property
 509     def line(self):
 510         return self._session.request('nvim_get_current_line')
 511
 512     @line.setter
 513     def line(self, line):
 514         return self._session.request('nvim_set_current_line', line)
 515
 516     @line.deleter
 517     def line(self):
 518         return self._session.request('nvim_del_current_line')
 519
 520     @property
 521     def buffer(self):
 522         return self._session.request('nvim_get_current_buf')
 523
 524     @buffer.setter
 525     def buffer(self, buffer):
 526         return self._session.request('nvim_set_current_buf', buffer)
 527
 528     @property
 529     def window(self):
 530         return self._session.request('nvim_get_current_win')
 531
 532     @window.setter
 533     def window(self, window):
 534         return self._session.request('nvim_set_current_win', window)
 535
 536     @property
 537     def tabpage(self):
 538         return self._session.request('nvim_get_current_tabpage')
 539
 540     @tabpage.setter
 541     def tabpage(self, tabpage):
 542         return self._session.request('nvim_set_current_tabpage', tabpage)
 543
 544
 545 class Funcs(object):
 546
 547     """Helper class for functional vimscript interface."""
 548
 549     def __init__(self, nvim):
 550         self._nvim = nvim
 551
 552     def __getattr__(self, name):
 553         return partial(self._nvim.call, name)
 554
 555
 556 class LuaFuncs(object):
 557
 558     """Wrapper to allow lua functions to be called like python methods."""
 559
 560     def __init__(self, nvim, name=""):
 561         self._nvim = nvim
 562         self.name = name
 563
 564     def __getattr__(self, name):
 565         """Return wrapper to named api method."""
 566         prefix = self.name + "." if self.name else ""
 567         return LuaFuncs(self._nvim, prefix + name)
 568
 569     def __call__(self, *args, **kwargs):
 570         # first new function after keyword rename, be a bit noisy
 571         if 'async' in kwargs:
 572             raise ValueError('"async" argument is not allowed. '
 573                              'Use "async_" instead.')
 574         async_ = kwargs.get('async_', False)
 575         pattern = "return {}(...)" if not async_ else "{}(...)"
 576         code = pattern.format(self.name)
 577         return self._nvim.exec_lua(code, *args, **kwargs)