More typing work

Also use a mypy based linter when editing
This commit is contained in:
Kovid Goyal
2020-03-08 11:02:14 +05:30
parent 9b32f18109
commit cc1336a616
43 changed files with 333 additions and 196 deletions

View File

@@ -5,7 +5,7 @@
from contextlib import suppress
from typing import (
TYPE_CHECKING, Any, Callable, Dict, FrozenSet, Generator, List, NoReturn,
Optional, Tuple, Union
Optional, Tuple, Type, Union, cast
)
from kitty.cli import get_defaults_from_seq, parse_args, parse_option_spec
@@ -42,12 +42,26 @@ class UnknownLayout(ValueError):
hide_traceback = True
class PayloadGetter:
def __init__(self, cmd: 'RemoteCommand', payload: Dict[str, Any]):
self.payload = payload
self.cmd = cmd
def __call__(self, key: str, opt_name: Optional[str] = None, missing: Any = None):
ans = self.payload.get(key, payload_get)
if ans is not payload_get:
return ans
return self.cmd.get_default(opt_name or key, missing=missing)
no_response = NoResponse()
payload_get = object()
ResponseType = Optional[Union[bool, str]]
CmdReturnType = Union[Dict[str, Any], List, Tuple, str, int, float, bool]
CmdGenerator = Generator[CmdReturnType, None, None]
PayloadType = Optional[Union[CmdReturnType, CmdGenerator]]
PayloadGetType = Callable[[str, str], Any]
PayloadGetType = PayloadGetter
ArgsType = List[str]
@@ -106,6 +120,7 @@ class RemoteCommand:
args_count: Optional[int] = None
args_completion: Optional[Dict[str, Tuple[str, Tuple[str, ...]]]] = None
defaults: Optional[Dict[str, Any]] = None
options_class: Type = RCOptions
def __init__(self):
self.desc = self.desc or self.short_desc
@@ -124,13 +139,6 @@ class RemoteCommand:
return self.defaults.get(name, missing)
return missing
def payload_get(self, payload: Dict[str, Any], key: str, opt_name: Optional[str] = None) -> Any:
payload_get = object()
ans = payload.get(key, payload_get)
if ans is not payload_get:
return ans
return self.get_default(opt_name or key)
def message_to_kitty(self, global_opts: RCOptions, opts: Any, args: ArgsType) -> PayloadType:
raise NotImplementedError()
@@ -143,7 +151,7 @@ def cli_params_for(command: RemoteCommand) -> Tuple[Callable[[], str], str, str,
def parse_subcommand_cli(command: RemoteCommand, args: ArgsType) -> Tuple[Any, ArgsType]:
opts, items = parse_args(args[1:], *cli_params_for(command))
opts, items = parse_args(args[1:], *cli_params_for(command), result_class=command.options_class)
if command.args_count is not None and command.args_count != len(items):
if command.args_count == 0:
raise SystemExit('Unknown extra argument(s) supplied to {}'.format(command.name))
@@ -163,17 +171,17 @@ def command_for_name(cmd_name: str) -> RemoteCommand:
m = import_module(f'kitty.rc.{cmd_name}')
except ImportError:
raise KeyError(f'{cmd_name} is not a known kitty remote control command')
return getattr(m, cmd_name)
return cast(RemoteCommand, getattr(m, cmd_name))
def all_command_names() -> FrozenSet[str]:
try:
from importlib.resources import contents
except ImportError:
from importlib_resources import contents
from importlib_resources import contents # type:ignore
def ok(name: str) -> bool:
root, _, ext = name.rpartition('.')
return ext in ('py', 'pyc', 'pyo') and root and root not in ('base', '__init__')
return bool(ext in ('py', 'pyc', 'pyo') and root and root not in ('base', '__init__'))
return frozenset({x.rpartition('.')[0] for x in filter(ok, contents('kitty.rc'))})

View File

@@ -39,7 +39,7 @@ If specified close the tab this command is run in, rather than the active tab.
if not tabs:
raise MatchError(match, 'tabs')
else:
tabs = [boss.tab_for_window(window) if window and payload_get('self') else boss.active_tab]
tabs = tuple(boss.tab_for_window(window) if window and payload_get('self') else boss.active_tab)
for tab in tabs:
if window:
if tab:

View File

@@ -38,7 +38,7 @@ If specified close the window this command is run in, rather than the active win
if not windows:
raise MatchError(match)
else:
windows = [window if window and payload_get('self') else boss.active_window]
windows = tuple(window if window and payload_get('self') else boss.active_window)
for window in windows:
if window:
boss.close_window(window)

View File

@@ -48,7 +48,7 @@ If specified apply marker to the window this command is run in, rather than the
if not windows:
raise MatchError(match)
else:
windows = [window if window and payload_get('self') else boss.active_window]
windows = tuple(window if window and payload_get('self') else boss.active_window)
args = payload_get('marker_spec')
for window in windows:

View File

@@ -43,7 +43,7 @@ If specified detach the tab this command is run in, rather than the active tab.
if not tabs:
raise MatchError(match)
else:
tabs = [window.tabref() if payload_get('self') and window and window.tabref() else boss.active_tab]
tabs = tuple(window.tabref() if payload_get('self') and window and window.tabref() else boss.active_tab)
match = payload_get('target_tab')
kwargs = {}
if match:

View File

@@ -45,7 +45,7 @@ If specified detach the window this command is run in, rather than the active wi
if not windows:
raise MatchError(match)
else:
windows = [window if window and payload_get('self') else boss.active_window]
windows = tuple(window if window and payload_get('self') else boss.active_window)
match = payload_get('target_tab')
kwargs = {}
if match:

View File

@@ -42,7 +42,7 @@ using this option means that you will not be notified of failures.
if match:
tabs = tuple(boss.match_tabs(match))
else:
tabs = [boss.tab_for_window(window) if window else boss.active_tab]
tabs = tuple(boss.tab_for_window(window) if window else boss.active_tab)
if not tabs:
raise MatchError(match, 'tabs')
tab = tabs[0]

View File

@@ -41,7 +41,7 @@ the command will exit with a success code.
windows = [window or boss.active_window]
match = payload_get('match')
if match:
windows = tuple(boss.match_windows(match))
windows = [boss.match_windows(match)]
if not windows:
raise MatchError(match)
for window in windows:

View File

@@ -42,9 +42,9 @@ configured colors.
def response_from_kitty(self, boss: 'Boss', window: 'Window', payload_get: PayloadGetType) -> ResponseType:
ans = {k: getattr(boss.opts, k) for k in boss.opts if isinstance(getattr(boss.opts, k), Color)}
if not payload_get('configured'):
windows = (window or boss.active_window,)
windows = [window or boss.active_window]
if payload_get('match'):
windows = tuple(boss.match_windows(payload_get('match')))
windows = list(boss.match_windows(payload_get('match')))
if not windows:
raise MatchError(payload_get('match'))
ans.update({k: color_from_int(v) for k, v in windows[0].current_colors.items()})

View File

@@ -55,7 +55,7 @@ If specified get text from the window this command is run in, rather than the ac
if not windows:
raise MatchError(match)
else:
windows = [window if window and payload_get('self') else boss.active_window]
windows = tuple(window if window and payload_get('self') else boss.active_window)
window = windows[0]
if payload_get('extent') == 'selection':
ans = window.text_for_selection()

View File

@@ -44,7 +44,7 @@ class GotoLayout(RemoteCommand):
if not tabs:
raise MatchError(match, 'tabs')
else:
tabs = [boss.tab_for_window(window) if window else boss.active_tab]
tabs = tuple(boss.tab_for_window(window) if window else boss.active_tab)
for tab in tabs:
if tab:
try:

View File

@@ -40,7 +40,7 @@ class Kitten(RemoteCommand):
windows = [window or boss.active_window]
match = payload_get('match')
if match:
windows = tuple(boss.match_windows(match))
windows = list(boss.match_windows(match))
if not windows:
raise MatchError(match)
for window in windows:

View File

@@ -39,7 +39,7 @@ class LastUsedLayout(RemoteCommand):
if not tabs:
raise MatchError(match, 'tabs')
else:
tabs = [boss.tab_for_window(window) if window else boss.active_tab]
tabs = tuple(boss.tab_for_window(window) if window else boss.active_tab)
for tab in tabs:
if tab:
tab.last_used_layout()

View File

@@ -80,7 +80,7 @@ instead of the active tab
setattr(opts, key, val)
match = payload_get('match')
if match:
tabs = tuple(boss.match_tabs(match))
tabs = list(boss.match_tabs(match))
if not tabs:
raise MatchError(match, 'tabs')
else:
@@ -88,7 +88,7 @@ instead of the active tab
if payload_get('self') and window and window.tabref():
tabs = [window.tabref()]
tab = tabs[0]
w = do_launch(boss, opts, payload_get('args') or None, target_tab=tab)
w = do_launch(boss, opts, payload_get('args') or [], target_tab=tab)
return None if payload_get('no_response') else str(getattr(w, 'id', 0))

View File

@@ -32,7 +32,8 @@ class NewWindow(RemoteCommand):
desc = (
'Open a new window in the specified tab. If you use the :option:`kitty @ new-window --match` option'
' the first matching tab is used. Otherwise the currently active tab is used.'
' Prints out the id of the newly opened window (unless :option:`--no-response` is used). Any command line arguments'
' Prints out the id of the newly opened window'
' (unless :option:`--no-response` is used). Any command line arguments'
' are assumed to be the command line used to run in the new window, if none'
' are provided, the default shell is run. For example:\n'
':italic:`kitty @ new-window --title Email mutt`'
@@ -109,7 +110,7 @@ the id of the new window will not be printed out.
match = payload_get('match')
if match:
tabs = tuple(boss.match_tabs(match))
tabs = list(boss.match_tabs(match))
if not tabs:
raise MatchError(match, 'tabs')
else:

View File

@@ -39,7 +39,7 @@ If specified apply marker to the window this command is run in, rather than the
if not windows:
raise MatchError(match)
else:
windows = [window if window and payload_get('self') else boss.active_window]
windows = tuple(window if window and payload_get('self') else boss.active_window)
for window in windows:
window.remove_marker()

View File

@@ -22,7 +22,10 @@ class ResizeWindow(RemoteCommand):
'''
short_desc = 'Resize the specified window'
desc = 'Resize the specified window in the current layout. Note that not all layouts can resize all windows in all directions.'
desc = (
'Resize the specified window in the current layout.'
' Note that not all layouts can resize all windows in all directions.'
)
options_spec = MATCH_WINDOW_OPTION + '''\n
--increment -i
type=int
@@ -34,9 +37,10 @@ The number of cells to change the size by, can be negative to decrease the size.
type=choices
choices=horizontal,vertical,reset
default=horizontal
The axis along which to resize. If :italic:`horizontal`, it will make the window wider or narrower by the specified increment.
If :italic:`vertical`, it will make the window taller or shorter by the specified increment. The special value :italic:`reset` will
reset the layout to its default configuration.
The axis along which to resize. If :italic:`horizontal`,
it will make the window wider or narrower by the specified increment.
If :italic:`vertical`, it will make the window taller or shorter by the specified increment.
The special value :italic:`reset` will reset the layout to its default configuration.
--self
@@ -56,7 +60,7 @@ If specified resize the window this command is run in, rather than the active wi
if not windows:
raise MatchError(match)
else:
windows = [window if window and payload_get('self') else boss.active_window]
windows = tuple(window if window and payload_get('self') else boss.active_window)
resized = False
if windows and windows[0]:
resized = boss.resize_layout_window(

View File

@@ -3,7 +3,7 @@
# License: GPLv3 Copyright: 2020, Kovid Goyal <kovid at kovidgoyal.net>
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, Optional, Tuple, Union
from .base import (
MATCH_WINDOW_OPTION, ArgsType, Boss, MatchError, PayloadGetType,
@@ -35,23 +35,22 @@ class ScrollWindow(RemoteCommand):
def message_to_kitty(self, global_opts: RCOptions, opts: 'CLIOptions', args: ArgsType) -> PayloadType:
amt = args[0]
ans = {'match': opts.match}
if amt in ('start', 'end'):
ans['amount'] = amt, None
else:
amount: Tuple[Union[str, int], Optional[str]] = amt, None
if amt not in ('start', 'end'):
pages = 'p' in amt
amt = amt.replace('p', '')
mult = -1 if amt.endswith('-') else 1
amt = int(amt.replace('-', ''))
ans['amount'] = [amt * mult, 'p' if pages else 'l']
return ans
q = int(amt.replace('-', ''))
amount = q * mult, 'p' if pages else 'l'
return {'match': opts.match, 'amount': amount}
def response_from_kitty(self, boss: 'Boss', window: 'Window', payload_get: PayloadGetType) -> ResponseType:
windows = [window or boss.active_window]
match = payload_get('match')
amt = payload_get('amount')
if match:
windows = tuple(boss.match_windows(match))
windows = list(boss.match_windows(match))
if not windows:
raise MatchError(match)
for window in windows:

View File

@@ -4,7 +4,7 @@
import os
import sys
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, Generator, Dict
from kitty.config import parse_send_text_bytes
@@ -51,8 +51,7 @@ are sent as is, not interpreted for escapes.
limit = 1024
ret = {'match': opts.match, 'is_binary': False, 'match_tab': opts.match_tab}
def pipe():
ret['is_binary'] = True
def pipe() -> Generator[Dict, None, None]:
if sys.stdin.isatty():
import select
fd = sys.stdin.fileno()
@@ -64,28 +63,29 @@ are sent as is, not interpreted for escapes.
data = os.read(fd, limit)
if not data:
break # eof
data = data.decode('utf-8')
if '\x04' in data:
data = data[:data.index('\x04')]
decoded_data = data.decode('utf-8')
if '\x04' in decoded_data:
decoded_data = decoded_data[:decoded_data.index('\x04')]
keep_going = False
ret['text'] = data
ret['text'] = decoded_data
yield ret
else:
ret['is_binary'] = True
while True:
data = sys.stdin.read(limit)
data = sys.stdin.buffer.read(limit)
if not data:
break
ret['text'] = data[:limit]
yield ret
def chunks(text):
def chunks(text: str) -> Generator[Dict, None, None]:
ret['is_binary'] = False
while text:
ret['text'] = text[:limit]
yield ret
text = text[limit:]
def file_pipe(path):
def file_pipe(path: str) -> Generator[Dict, None, None]:
ret['is_binary'] = True
with open(path, encoding='utf-8') as f:
while True:
@@ -105,7 +105,7 @@ are sent as is, not interpreted for escapes.
text = ' '.join(args)
sources.append(chunks(text))
def chain():
def chain() -> Generator[Dict, None, None]:
for src in sources:
yield from src
return chain()
@@ -114,7 +114,7 @@ are sent as is, not interpreted for escapes.
windows = [boss.active_window]
match = payload_get('match')
if match:
windows = tuple(boss.match_windows(match))
windows = list(boss.match_windows(match))
mt = payload_get('match_tab')
if mt:
windows = []
@@ -123,7 +123,8 @@ are sent as is, not interpreted for escapes.
raise MatchError(payload_get('match_tab'), 'tabs')
for tab in tabs:
windows += tuple(tab)
data = payload_get('text').encode('utf-8') if payload_get('is_binary') else parse_send_text_bytes(payload_get('text'))
data = payload_get('text').encode('utf-8') if payload_get('is_binary') else parse_send_text_bytes(
payload_get('text'))
for window in windows:
if window is not None:
window.write_to_child(data)

View File

@@ -5,12 +5,12 @@
import imghdr
import tempfile
from base64 import standard_b64decode, standard_b64encode
from typing import TYPE_CHECKING, BinaryIO, Optional
from typing import IO, TYPE_CHECKING, Dict, Generator, Optional
from uuid import uuid4
from .base import (
MATCH_WINDOW_OPTION, ArgsType, Boss, PayloadGetType, PayloadType,
RCOptions, RemoteCommand, ResponseType, Window, no_response,
RCOptions, RemoteCommand, ResponseType, Window,
windows_for_payload
)
@@ -59,20 +59,26 @@ How the image should be displayed. The value of configured will use the configur
args_count = 1
args_completion = {'files': ('PNG Images', ('*.png',))}
current_img_id: Optional[str] = None
current_file_obj: Optional[BinaryIO] = None
current_file_obj: Optional[IO[bytes]] = None
def message_to_kitty(self, global_opts: RCOptions, opts: 'CLIOptions', args: ArgsType) -> PayloadType:
if not args:
self.fatal('Must specify path to PNG image')
path = args[0]
ret = {'match': opts.match, 'configured': opts.configured, 'layout': opts.layout, 'all': opts.all, 'img_id': str(uuid4())}
ret = {
'match': opts.match,
'configured': opts.configured,
'layout': opts.layout,
'all': opts.all,
'img_id': str(uuid4())
}
if path.lower() == 'none':
ret['data'] = '-'
return ret
if imghdr.what(path) != 'png':
self.fatal('{} is not a PNG image'.format(path))
def file_pipe(path):
def file_pipe(path) -> Generator[Dict, None, None]:
with open(path, 'rb') as f:
while True:
data = f.read(512)
@@ -92,8 +98,9 @@ How the image should be displayed. The value of configured will use the configur
set_background_image.current_img_id = img_id
set_background_image.current_file_obj = tempfile.NamedTemporaryFile()
if data:
assert set_background_image.current_file_obj is not None
set_background_image.current_file_obj.write(standard_b64decode(data))
return no_response
return None
windows = windows_for_payload(boss, window, payload_get)
os_windows = tuple({w.os_window_id for w in windows})
@@ -101,6 +108,7 @@ How the image should be displayed. The value of configured will use the configur
if data == '-':
path = None
else:
assert set_background_image.current_file_obj is not None
f = set_background_image.current_file_obj
path = f.name
set_background_image.current_file_obj = None
@@ -109,7 +117,7 @@ How the image should be displayed. The value of configured will use the configur
try:
boss.set_background_image(path, os_windows, payload_get('configured'), layout)
except ValueError as err:
err.hide_traceback = True
err.hide_traceback = True # type: ignore
raise

View File

@@ -4,7 +4,7 @@
import os
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, Dict, Optional
from kitty.config import parse_config
from kitty.fast_data_types import patch_color_profiles
@@ -34,7 +34,8 @@ class SetColors(RemoteCommand):
short_desc = 'Set terminal colors'
desc = (
'Set the terminal colors for the specified windows/tabs (defaults to active window). You can either specify the path to a conf file'
'Set the terminal colors for the specified windows/tabs (defaults to active window).'
' You can either specify the path to a conf file'
' (in the same format as kitty.conf) to read the colors from or you can specify individual colors,'
' for example: kitty @ set-colors foreground=red background=white'
)
@@ -59,21 +60,28 @@ this option, any color arguments are ignored and --configured and --all are impl
argspec = 'COLOR_OR_FILE ...'
def message_to_kitty(self, global_opts: RCOptions, opts: 'CLIOptions', args: ArgsType) -> PayloadType:
colors, cursor_text_color = {}, False
final_colors: Dict[str, int] = {}
cursor_text_color: Optional[int] = None
if not opts.reset:
colors: Dict[str, Optional[Color]] = {}
for spec in args:
if '=' in spec:
colors.update(parse_config((spec.replace('=', ' '),)))
else:
with open(os.path.expanduser(spec), encoding='utf-8', errors='replace') as f:
colors.update(parse_config(f))
cursor_text_color = colors.pop('cursor_text_color', False)
colors = {k: color_as_int(v) for k, v in colors.items() if isinstance(v, Color)}
return {
ctc = colors.pop('cursor_text_color')
if isinstance(ctc, Color):
cursor_text_color = color_as_int(ctc)
final_colors = {k: color_as_int(v) for k, v in colors.items() if isinstance(v, Color)}
ans = {
'match_window': opts.match, 'match_tab': opts.match_tab,
'all': opts.all or opts.reset, 'configured': opts.configured or opts.reset,
'colors': colors, 'reset': opts.reset, 'cursor_text_color': cursor_text_color
'colors': final_colors, 'reset': opts.reset
}
if cursor_text_color is not None:
ans['cursor_text_color'] = cursor_text_color
return ans
def response_from_kitty(self, boss: 'Boss', window: 'Window', payload_get: PayloadGetType) -> ResponseType:
windows = windows_for_payload(boss, window, payload_get)

View File

@@ -41,7 +41,7 @@ class SetTabTitle(RemoteCommand):
if not tabs:
raise MatchError(match, 'tabs')
else:
tabs = [boss.tab_for_window(window) if window else boss.active_tab]
tabs = tuple(boss.tab_for_window(window) if window else boss.active_tab)
for tab in tabs:
if tab:
tab.set_title(payload_get('title'))

View File

@@ -44,7 +44,7 @@ want to allow other programs to change it afterwards, use this option.
windows = [window or boss.active_window]
match = payload_get('match')
if match:
windows = tuple(boss.match_windows(match))
windows = list(boss.match_windows(match))
if not windows:
raise MatchError(match)
for window in windows: