Create an API to easily use remote control commands from kittens

This commit is contained in:
Kovid Goyal
2022-08-20 16:19:16 +05:30
parent 3b77f3b4bc
commit 2efb8f6dc2
2 changed files with 40 additions and 13 deletions

View File

@@ -49,6 +49,23 @@ kittens. Look in the `kittens sub-directory
code for those. Or see below for a list of :ref:`third-party kittens code for those. Or see below for a list of :ref:`third-party kittens
<external_kittens>`, that other kitty users have created. <external_kittens>`, that other kitty users have created.
kitty API to use with kittens
-------------------------------
Kittens have full access to internal kitty APIs. However these are neither
entirely stable nor documented. You can instead use the kitty :doc:`Remote
control API </remote-control>`. Simply call :code:`boss.remote_control`, with
the same arguments you would pass to ``kitty @ ``. For example:
.. code-block:: python
def handle_result(args: List[str], answer: str, target_window_id: int, boss: Boss) -> None:
# get the kitty window to which to send text
w = boss.window_id_map.get(target_window_id)
if w is not None:
boss.call_remote_control(w, ('send-text', 'some text'))
Passing arguments to kittens Passing arguments to kittens
------------------------------ ------------------------------

View File

@@ -12,8 +12,8 @@ from functools import partial
from gettext import gettext as _ from gettext import gettext as _
from time import monotonic from time import monotonic
from typing import ( from typing import (
Any, Callable, Container, Dict, Iterable, Iterator, List, Optional, Set, TYPE_CHECKING, Any, Callable, Container, Dict, Iterable, Iterator, List,
Tuple, Union Optional, Set, Tuple, Union
) )
from weakref import WeakValueDictionary from weakref import WeakValueDictionary
@@ -67,6 +67,9 @@ from .utils import (
) )
from .window import CommandOutput, CwdRequest, Window from .window import CommandOutput, CwdRequest, Window
if TYPE_CHECKING:
from .rc.base import ResponseType
RCResponse = Union[Dict[str, Any], None, AsyncResponse] RCResponse = Union[Dict[str, Any], None, AsyncResponse]
@@ -563,6 +566,14 @@ class Boss:
See :ref:`rc_mapping` for details. See :ref:`rc_mapping` for details.
''') ''')
def remote_control(self, *args: str) -> None: def remote_control(self, *args: str) -> None:
try:
self.call_remote_control(self.active_window, args)
except (Exception, SystemExit):
import traceback
tb = traceback.format_exc()
self.show_error(_('remote_control mapping failed'), tb)
def call_remote_control(self, active_window: Optional[Window], args: Tuple[str, ...]) -> 'ResponseType':
from .rc.base import ( from .rc.base import (
PayloadGetter, command_for_name, parse_subcommand_cli PayloadGetter, command_for_name, parse_subcommand_cli
) )
@@ -570,21 +581,20 @@ class Boss:
try: try:
global_opts, items = parse_rc_args(['@'] + list(args)) global_opts, items = parse_rc_args(['@'] + list(args))
if not items: if not items:
return return None
cmd = items[0] cmd = items[0]
c = command_for_name(cmd) c = command_for_name(cmd)
opts, items = parse_subcommand_cli(c, items) opts, items = parse_subcommand_cli(c, items)
payload = c.message_to_kitty(global_opts, opts, items) payload = c.message_to_kitty(global_opts, opts, items)
import types except SystemExit as e:
if isinstance(payload, types.GeneratorType): raise Exception(str(e)) from e
for x in payload: import types
c.response_from_kitty(self, self.active_window, PayloadGetter(c, x if isinstance(x, dict) else {})) if isinstance(payload, types.GeneratorType):
else: for x in payload:
c.response_from_kitty(self, self.active_window, PayloadGetter(c, payload if isinstance(payload, dict) else {})) c.response_from_kitty(self, active_window, PayloadGetter(c, x if isinstance(x, dict) else {}))
except (Exception, SystemExit): else:
import traceback return c.response_from_kitty(self, active_window, PayloadGetter(c, payload if isinstance(payload, dict) else {}))
tb = traceback.format_exc() return None
self.show_error(_('remote_control mapping failed'), tb)
def peer_message_received(self, msg_bytes: bytes, peer_id: int) -> Union[bytes, bool, None]: def peer_message_received(self, msg_bytes: bytes, peer_id: int) -> Union[bytes, bool, None]:
cmd_prefix = b'\x1bP@kitty-cmd' cmd_prefix = b'\x1bP@kitty-cmd'