More typing work
This commit is contained in:
parent
8a34fede55
commit
afec07b124
@ -4,15 +4,19 @@
|
||||
|
||||
import os
|
||||
from contextlib import suppress
|
||||
from typing import TYPE_CHECKING, Dict, List, Union
|
||||
|
||||
from kitty.cli import parse_args
|
||||
from kitty.constants import cache_dir
|
||||
from kitty.cli_stub import AskCLIOptions
|
||||
from kitty.constants import cache_dir
|
||||
|
||||
from ..tui.operations import alternate_screen, styled
|
||||
from ..tui.handler import result_handler
|
||||
from ..tui.operations import alternate_screen, styled
|
||||
|
||||
readline = None
|
||||
if TYPE_CHECKING:
|
||||
import readline
|
||||
else:
|
||||
readline = None
|
||||
|
||||
|
||||
def get_history_items():
|
||||
@ -98,7 +102,7 @@ def main(args):
|
||||
raise SystemExit(e.code)
|
||||
|
||||
init_readline(readline)
|
||||
ans = {'items': items}
|
||||
response = None
|
||||
|
||||
with alternate_screen(), HistoryCompleter(args.name):
|
||||
if args.message:
|
||||
@ -106,7 +110,11 @@ def main(args):
|
||||
|
||||
prompt = '> '
|
||||
with suppress(KeyboardInterrupt, EOFError):
|
||||
ans['response'] = input(prompt)
|
||||
response = input(prompt)
|
||||
if response is None:
|
||||
ans: Dict[str, Union[str, List[str]]] = {'items': items}
|
||||
else:
|
||||
ans = {'items': items, 'response': response}
|
||||
return ans
|
||||
|
||||
|
||||
|
||||
@ -0,0 +1,8 @@
|
||||
class GlobalData:
|
||||
|
||||
def __init__(self):
|
||||
self.title = ''
|
||||
self.cmd = ''
|
||||
|
||||
|
||||
global_data = GlobalData
|
||||
@ -8,7 +8,7 @@ from contextlib import suppress
|
||||
from functools import lru_cache
|
||||
from hashlib import md5
|
||||
from mimetypes import guess_type
|
||||
from typing import Dict
|
||||
from typing import Dict, List, Set
|
||||
|
||||
path_name_map: Dict[str, str] = {}
|
||||
|
||||
@ -20,11 +20,12 @@ class Segment:
|
||||
def __init__(self, start, start_code):
|
||||
self.start = start
|
||||
self.start_code = start_code
|
||||
self.end = None
|
||||
self.end_code = None
|
||||
|
||||
def __repr__(self):
|
||||
return 'Segment(start={!r}, start_code={!r}, end={!r}, end_code={!r})'.format(
|
||||
self.start, self.start_code, getattr(self, 'end', None), getattr(self, 'end_code', None)
|
||||
)
|
||||
self.start, self.start_code, self.end, self.end_code)
|
||||
|
||||
|
||||
class Collection:
|
||||
@ -81,8 +82,10 @@ class Collection:
|
||||
|
||||
|
||||
def collect_files(collection, left, right):
|
||||
left_names, right_names = set(), set()
|
||||
left_path_map, right_path_map = {}, {}
|
||||
left_names: Set[str] = set()
|
||||
right_names: Set[str] = set()
|
||||
left_path_map: Dict[str, str] = {}
|
||||
right_path_map: Dict[str, str] = {}
|
||||
|
||||
def walk(base, names, pmap):
|
||||
for dirpath, dirnames, filenames in os.walk(base):
|
||||
@ -178,7 +181,7 @@ def create_collection(left, right):
|
||||
return collection
|
||||
|
||||
|
||||
highlight_data = {}
|
||||
highlight_data: Dict[str, List] = {}
|
||||
|
||||
|
||||
def set_highlight_data(data):
|
||||
|
||||
@ -3,18 +3,20 @@
|
||||
# License: GPL v3 Copyright: 2018, Kovid Goyal <kovid at kovidgoyal.net>
|
||||
|
||||
import os
|
||||
from typing import Any, Dict, Optional, Type
|
||||
|
||||
from kitty.conf.utils import (
|
||||
init_config as _init_config, key_func, load_config as _load_config, merge_dicts,
|
||||
parse_config_base, parse_kittens_key, resolve_config
|
||||
)
|
||||
from kitty.conf.definition import config_lines
|
||||
from kitty.conf.utils import (
|
||||
init_config as _init_config, key_func, load_config as _load_config,
|
||||
merge_dicts, parse_config_base, parse_kittens_key, resolve_config
|
||||
)
|
||||
from kitty.constants import config_dir
|
||||
from kitty.options_stub import DiffOptions
|
||||
from kitty.rgb import color_as_sgr
|
||||
|
||||
from .config_data import type_convert, all_options
|
||||
from .config_data import all_options, type_convert
|
||||
|
||||
defaults = None
|
||||
defaults: Optional[DiffOptions] = None
|
||||
|
||||
formats = {
|
||||
'title': '',
|
||||
@ -79,13 +81,15 @@ def parse_start_search(func, rest):
|
||||
|
||||
def special_handling(key, val, ans):
|
||||
if key == 'map':
|
||||
action, *key_def = parse_kittens_key(val, args_funcs)
|
||||
ans['key_definitions'][tuple(key_def)] = action
|
||||
return True
|
||||
x = parse_kittens_key(val, args_funcs)
|
||||
if x is not None:
|
||||
action, *key_def = x
|
||||
ans['key_definitions'][tuple(key_def)] = action
|
||||
return True
|
||||
|
||||
|
||||
def parse_config(lines, check_keys=True):
|
||||
ans = {'key_definitions': {}}
|
||||
ans: Dict[str, Any] = {'key_definitions': {}}
|
||||
parse_config_base(
|
||||
lines,
|
||||
defaults,
|
||||
@ -112,7 +116,9 @@ def parse_defaults(lines, check_keys=False):
|
||||
return parse_config(lines, check_keys)
|
||||
|
||||
|
||||
Options, defaults = _init_config(config_lines(all_options), parse_defaults)
|
||||
x = _init_config(config_lines(all_options), parse_defaults)
|
||||
Options: Type[DiffOptions] = x[0]
|
||||
defaults = x[1]
|
||||
|
||||
|
||||
def load_config(*paths, overrides=None):
|
||||
|
||||
@ -5,6 +5,7 @@
|
||||
import concurrent
|
||||
import os
|
||||
import re
|
||||
from typing import List, Optional
|
||||
|
||||
from pygments import highlight # type: ignore
|
||||
from pygments.formatter import Formatter # type: ignore
|
||||
@ -35,11 +36,15 @@ class DiffFormatter(Formatter):
|
||||
for token, style in self.style:
|
||||
start = []
|
||||
end = []
|
||||
fstart = fend = ''
|
||||
# a style item is a tuple in the following form:
|
||||
# colors are readily specified in hex: 'RRGGBB'
|
||||
if style['color']:
|
||||
start.append('38' + color_as_sgr(parse_sharp(style['color'])))
|
||||
end.append('39')
|
||||
col = style['color']
|
||||
if col:
|
||||
pc = parse_sharp(col)
|
||||
if pc is not None:
|
||||
start.append('38' + color_as_sgr(pc))
|
||||
end.append('39')
|
||||
if style['bold']:
|
||||
start.append('1')
|
||||
end.append('22')
|
||||
@ -50,9 +55,9 @@ class DiffFormatter(Formatter):
|
||||
start.append('4')
|
||||
end.append('24')
|
||||
if start:
|
||||
start = '\033[{}m'.format(';'.join(start))
|
||||
end = '\033[{}m'.format(';'.join(end))
|
||||
self.styles[token] = start or '', end or ''
|
||||
fstart = '\033[{}m'.format(';'.join(start))
|
||||
fend = '\033[{}m'.format(';'.join(end))
|
||||
self.styles[token] = fstart, fend
|
||||
|
||||
def format(self, tokensource, outfile):
|
||||
for ttype, value in tokensource:
|
||||
@ -76,7 +81,7 @@ class DiffFormatter(Formatter):
|
||||
outfile.write(value)
|
||||
|
||||
|
||||
formatter = None
|
||||
formatter: Optional[DiffFormatter] = None
|
||||
|
||||
|
||||
def initialize_highlighter(style='default'):
|
||||
@ -102,8 +107,8 @@ split_pat = re.compile(r'(\033\[.*?m)')
|
||||
|
||||
|
||||
def highlight_line(line):
|
||||
ans = []
|
||||
current = None
|
||||
ans: List[Segment] = []
|
||||
current: Optional[Segment] = None
|
||||
pos = 0
|
||||
for x in split_pat.split(line):
|
||||
if x.startswith('\033'):
|
||||
@ -133,7 +138,6 @@ def highlight_collection(collection, aliases=None):
|
||||
jobs = {}
|
||||
ans = {}
|
||||
with concurrent.futures.ProcessPoolExecutor(max_workers=os.cpu_count()) as executor:
|
||||
highlight_collection.processes = executor._processes
|
||||
for path, item_type, other_path in collection:
|
||||
if item_type != 'rename':
|
||||
for p in (path, other_path):
|
||||
@ -156,8 +160,9 @@ def main():
|
||||
# kitty +runpy "from kittens.diff.highlight import main; main()" file
|
||||
import sys
|
||||
initialize_highlighter()
|
||||
with open(sys.argv[-1]) as f:
|
||||
highlighted = highlight_data(f.read(), f.name, defaults.syntax_aliases)
|
||||
if highlighted is None:
|
||||
raise SystemExit('Unknown filetype: {}'.format(sys.argv[-1]))
|
||||
print(highlighted)
|
||||
if defaults is not None:
|
||||
with open(sys.argv[-1]) as f:
|
||||
highlighted = highlight_data(f.read(), f.name, defaults.syntax_aliases)
|
||||
if highlighted is None:
|
||||
raise SystemExit('Unknown filetype: {}'.format(sys.argv[-1]))
|
||||
print(highlighted)
|
||||
|
||||
@ -13,6 +13,7 @@ from collections import defaultdict
|
||||
from contextlib import suppress
|
||||
from functools import partial
|
||||
from gettext import gettext as _
|
||||
from typing import DefaultDict, List, Tuple
|
||||
|
||||
from kitty.cli import CONFIG_HELP, parse_args
|
||||
from kitty.cli_stub import DiffCLIOptions
|
||||
@ -24,9 +25,10 @@ from .collect import (
|
||||
create_collection, data_for_path, lines_for_path, sanitize,
|
||||
set_highlight_data
|
||||
)
|
||||
from . import global_data
|
||||
from .config import init_config
|
||||
from .patch import Differ, set_diff_command, worker_processes
|
||||
from .render import ImageSupportWarning, LineRef, render_diff
|
||||
from .render import ImageSupportWarning, LineRef, Reference, render_diff
|
||||
from .search import BadRegex, Search
|
||||
from ..tui.handler import Handler
|
||||
from ..tui.images import ImageManager
|
||||
@ -186,7 +188,7 @@ class DiffHandler(Handler):
|
||||
def render_diff(self):
|
||||
self.diff_lines = tuple(render_diff(self.collection, self.diff_map, self.args, self.screen_size.cols, self.image_manager))
|
||||
self.margin_size = render_diff.margin_size
|
||||
self.ref_path_map = defaultdict(list)
|
||||
self.ref_path_map: DefaultDict[str, List[Tuple[int, Reference]]] = defaultdict(list)
|
||||
for i, l in enumerate(self.diff_lines):
|
||||
self.ref_path_map[l.ref.path].append((i, l.ref))
|
||||
self.max_scroll_pos = len(self.diff_lines) - self.num_lines
|
||||
@ -271,7 +273,7 @@ class DiffHandler(Handler):
|
||||
|
||||
def init_terminal_state(self):
|
||||
self.cmd.set_line_wrapping(False)
|
||||
self.cmd.set_window_title(main.title)
|
||||
self.cmd.set_window_title(global_data.title)
|
||||
self.cmd.set_default_colors(
|
||||
fg=self.opts.foreground, bg=self.opts.background,
|
||||
cursor=self.opts.foreground, select_fg=self.opts.select_fg,
|
||||
@ -547,7 +549,7 @@ def main(args):
|
||||
if len(items) != 2:
|
||||
raise SystemExit('You must specify exactly two files/directories to compare')
|
||||
left, right = items
|
||||
main.title = _('{} vs. {}').format(left, right)
|
||||
global_data.title = _('{} vs. {}').format(left, right)
|
||||
if os.path.isdir(left) != os.path.isdir(right):
|
||||
raise SystemExit('The items to be diffed should both be either directories or files. Comparing a directory to a file is not valid.')
|
||||
opts = init_config(args)
|
||||
|
||||
@ -7,14 +7,17 @@ import os
|
||||
import shlex
|
||||
import shutil
|
||||
import subprocess
|
||||
from typing import List, Optional, Tuple
|
||||
|
||||
from . import global_data
|
||||
from .collect import lines_for_path
|
||||
from .diff_speedup import changed_center
|
||||
|
||||
left_lines = right_lines = None
|
||||
left_lines: Tuple[str, ...] = ()
|
||||
right_lines: Tuple[str, ...] = ()
|
||||
GIT_DIFF = 'git diff --no-color --no-ext-diff --exit-code -U_CONTEXT_ --no-index --'
|
||||
DIFF_DIFF = 'diff -p -U _CONTEXT_ --'
|
||||
worker_processes = []
|
||||
worker_processes: List[int] = []
|
||||
|
||||
|
||||
def find_differ():
|
||||
@ -31,16 +34,17 @@ def set_diff_command(opt):
|
||||
raise SystemExit('Failed to find either the git or diff programs on your system')
|
||||
else:
|
||||
cmd = opt
|
||||
set_diff_command.cmd = cmd
|
||||
global_data.cmd = cmd
|
||||
|
||||
|
||||
def run_diff(file1, file2, context=3):
|
||||
def run_diff(file1: str, file2: str, context: int = 3):
|
||||
# returns: ok, is_different, patch
|
||||
cmd = shlex.split(set_diff_command.cmd.replace('_CONTEXT_', str(context)))
|
||||
cmd = shlex.split(global_data.cmd.replace('_CONTEXT_', str(context)))
|
||||
# we resolve symlinks because git diff does not follow symlinks, while diff
|
||||
# does. We want consistent behavior, also for integration with git difftool
|
||||
# we always want symlinks to be followed.
|
||||
path1, path2 = map(os.path.realpath, (file1, file2))
|
||||
path1 = os.path.realpath(file1)
|
||||
path2 = os.path.realpath(file2)
|
||||
p = subprocess.Popen(
|
||||
cmd + [path1, path2],
|
||||
stdout=subprocess.PIPE, stderr=subprocess.PIPE, stdin=subprocess.DEVNULL)
|
||||
@ -96,7 +100,7 @@ class Hunk:
|
||||
self.title = title
|
||||
self.added_count = self.removed_count = 0
|
||||
self.chunks = []
|
||||
self.current_chunk = None
|
||||
self.current_chunk: Optional[Chunk] = None
|
||||
self.largest_line_number = max(self.left_start + self.left_count, self.right_start + self.right_count)
|
||||
|
||||
def new_chunk(self, is_context=False):
|
||||
@ -125,20 +129,24 @@ class Hunk:
|
||||
|
||||
def add_line(self):
|
||||
self.ensure_diff_chunk()
|
||||
self.current_chunk.add_line()
|
||||
if self.current_chunk is not None:
|
||||
self.current_chunk.add_line()
|
||||
self.added_count += 1
|
||||
|
||||
def remove_line(self):
|
||||
self.ensure_diff_chunk()
|
||||
self.current_chunk.remove_line()
|
||||
if self.current_chunk is not None:
|
||||
self.current_chunk.remove_line()
|
||||
self.removed_count += 1
|
||||
|
||||
def context_line(self):
|
||||
self.ensure_context_chunk()
|
||||
self.current_chunk.context_line()
|
||||
if self.current_chunk is not None:
|
||||
self.current_chunk.context_line()
|
||||
|
||||
def finalize(self):
|
||||
self.chunks.append(self.current_chunk)
|
||||
if self.current_chunk is not None:
|
||||
self.chunks.append(self.current_chunk)
|
||||
del self.current_chunk
|
||||
# Sanity check
|
||||
c = self.chunks[-1]
|
||||
@ -158,7 +166,7 @@ def parse_range(x):
|
||||
|
||||
|
||||
def parse_hunk_header(line):
|
||||
parts = tuple(filter(None, line.split('@@', 2)))
|
||||
parts: Tuple[str, ...] = tuple(filter(None, line.split('@@', 2)))
|
||||
linespec = parts[0].strip()
|
||||
title = ''
|
||||
if len(parts) == 2:
|
||||
@ -208,7 +216,7 @@ def parse_patch(raw):
|
||||
|
||||
class Differ:
|
||||
|
||||
diff_executor = None
|
||||
diff_executor: Optional[concurrent.futures.ThreadPoolExecutor] = None
|
||||
|
||||
def __init__(self):
|
||||
self.jmap = {}
|
||||
@ -223,7 +231,8 @@ class Differ:
|
||||
def __call__(self, context=3):
|
||||
global left_lines, right_lines
|
||||
ans = {}
|
||||
executor = Differ.diff_executor
|
||||
executor = self.diff_executor
|
||||
assert executor is not None
|
||||
jobs = {executor.submit(run_diff, key, self.jmap[key], context): key for key in self.jobs}
|
||||
for future in concurrent.futures.as_completed(jobs):
|
||||
key = jobs[future]
|
||||
|
||||
@ -3,31 +3,32 @@
|
||||
# License: GPL v3 Copyright: 2018, Kovid Goyal <kovid at kovidgoyal.net>
|
||||
|
||||
import warnings
|
||||
from functools import lru_cache
|
||||
from gettext import gettext as _
|
||||
from itertools import repeat, zip_longest
|
||||
from math import ceil
|
||||
from typing import Callable, Iterable, List, Optional
|
||||
|
||||
from kitty.fast_data_types import truncate_point_for_length, wcswidth
|
||||
|
||||
from ..tui.images import can_display_images
|
||||
from .collect import (
|
||||
Segment, data_for_path, highlights_for_path, is_image, lines_for_path,
|
||||
path_name_map, sanitize
|
||||
)
|
||||
from .config import formats
|
||||
from .diff_speedup import split_with_highlights as _split_with_highlights
|
||||
from ..tui.images import can_display_images
|
||||
|
||||
|
||||
class ImageSupportWarning(Warning):
|
||||
pass
|
||||
|
||||
|
||||
@lru_cache(maxsize=2)
|
||||
def images_supported():
|
||||
ans = getattr(images_supported, 'ans', None)
|
||||
if ans is None:
|
||||
images_supported.ans = ans = can_display_images()
|
||||
if not ans:
|
||||
warnings.warn('ImageMagick not found images cannot be displayed', ImageSupportWarning)
|
||||
ans = can_display_images()
|
||||
if not ans:
|
||||
warnings.warn('ImageMagick not found images cannot be displayed', ImageSupportWarning)
|
||||
return ans
|
||||
|
||||
|
||||
@ -44,6 +45,8 @@ class Ref:
|
||||
class LineRef(Ref):
|
||||
|
||||
__slots__ = ('src_line_number', 'wrapped_line_idx')
|
||||
src_line_number: int
|
||||
wrapped_line_idx: int
|
||||
|
||||
def __init__(self, sln, wli=0):
|
||||
object.__setattr__(self, 'src_line_number', sln)
|
||||
@ -53,6 +56,8 @@ class LineRef(Ref):
|
||||
class Reference(Ref):
|
||||
|
||||
__slots__ = ('path', 'extra')
|
||||
path: str
|
||||
extra: Optional[LineRef]
|
||||
|
||||
def __init__(self, path, extra=None):
|
||||
object.__setattr__(self, 'path', path)
|
||||
@ -111,8 +116,8 @@ def place_in(text, sz):
|
||||
return fill_in(fit_in(text, sz), sz)
|
||||
|
||||
|
||||
def format_func(which):
|
||||
def formatted(text):
|
||||
def format_func(which: str) -> Callable[[str], str]:
|
||||
def formatted(text: str) -> str:
|
||||
fmt = formats[which]
|
||||
return '\x1b[' + fmt + 'm' + text + '\x1b[0m'
|
||||
formatted.__name__ = which + '_format'
|
||||
@ -226,7 +231,7 @@ class DiffData:
|
||||
return []
|
||||
|
||||
|
||||
def render_diff_line(number, text, ltype, margin_size, available_cols):
|
||||
def render_diff_line(number, text, ltype, margin_size, available_cols) -> str:
|
||||
margin = margin_bg_map[ltype](place_in(number, margin_size))
|
||||
content = text_bg_map[ltype](fill_in(text or '', available_cols))
|
||||
return margin + content
|
||||
@ -292,7 +297,8 @@ def lines_for_chunk(data, hunk_num, chunk, chunk_num):
|
||||
else:
|
||||
common = min(chunk.left_count, chunk.right_count)
|
||||
for i in range(max(chunk.left_count, chunk.right_count)):
|
||||
ll, rl = [], []
|
||||
ll: List[str] = []
|
||||
rl: List[str] = []
|
||||
if i < chunk.left_count:
|
||||
rln = ref_ln = chunk.left_start + i
|
||||
ll.extend(render_half_line(
|
||||
@ -416,7 +422,8 @@ def render_image(path, is_left, available_cols, margin_size, image_manager):
|
||||
|
||||
def image_lines(left_path, right_path, columns, margin_size, image_manager):
|
||||
available_cols = columns // 2 - margin_size
|
||||
left_lines, right_lines = iter(()), iter(())
|
||||
left_lines: Iterable[str] = iter(())
|
||||
right_lines: Iterable[str] = iter(())
|
||||
if left_path is not None:
|
||||
left_lines = render_image(left_path, True, available_cols, margin_size, image_manager)
|
||||
if right_path is not None:
|
||||
@ -439,52 +446,59 @@ def image_lines(left_path, right_path, columns, margin_size, image_manager):
|
||||
is_change_start = False
|
||||
|
||||
|
||||
def render_diff(collection, diff_map, args, columns, image_manager):
|
||||
largest_line_number = 0
|
||||
for path, item_type, other_path in collection:
|
||||
if item_type == 'diff':
|
||||
patch = diff_map.get(path)
|
||||
if patch is not None:
|
||||
largest_line_number = max(largest_line_number, patch.largest_line_number)
|
||||
class RenderDiff:
|
||||
|
||||
margin_size = render_diff.margin_size = max(3, len(str(largest_line_number)) + 1)
|
||||
last_item_num = len(collection) - 1
|
||||
margin_size: int = 0
|
||||
|
||||
for i, (path, item_type, other_path) in enumerate(collection):
|
||||
item_ref = Reference(path)
|
||||
is_binary = isinstance(data_for_path(path), bytes)
|
||||
if not is_binary and item_type == 'diff' and isinstance(data_for_path(other_path), bytes):
|
||||
is_binary = True
|
||||
is_img = is_binary and (is_image(path) or is_image(other_path)) and images_supported()
|
||||
yield from yield_lines_from(title_lines(path, other_path, args, columns, margin_size), item_ref, False)
|
||||
if item_type == 'diff':
|
||||
if is_binary:
|
||||
if is_img:
|
||||
ans = image_lines(path, other_path, columns, margin_size, image_manager)
|
||||
def __call__(self, collection, diff_map, args, columns, image_manager):
|
||||
largest_line_number = 0
|
||||
for path, item_type, other_path in collection:
|
||||
if item_type == 'diff':
|
||||
patch = diff_map.get(path)
|
||||
if patch is not None:
|
||||
largest_line_number = max(largest_line_number, patch.largest_line_number)
|
||||
|
||||
margin_size = self.margin_size = max(3, len(str(largest_line_number)) + 1)
|
||||
last_item_num = len(collection) - 1
|
||||
|
||||
for i, (path, item_type, other_path) in enumerate(collection):
|
||||
item_ref = Reference(path)
|
||||
is_binary = isinstance(data_for_path(path), bytes)
|
||||
if not is_binary and item_type == 'diff' and isinstance(data_for_path(other_path), bytes):
|
||||
is_binary = True
|
||||
is_img = is_binary and (is_image(path) or is_image(other_path)) and images_supported()
|
||||
yield from yield_lines_from(title_lines(path, other_path, args, columns, margin_size), item_ref, False)
|
||||
if item_type == 'diff':
|
||||
if is_binary:
|
||||
if is_img:
|
||||
ans = image_lines(path, other_path, columns, margin_size, image_manager)
|
||||
else:
|
||||
ans = yield_lines_from(binary_lines(path, other_path, columns, margin_size), item_ref)
|
||||
else:
|
||||
ans = yield_lines_from(binary_lines(path, other_path, columns, margin_size), item_ref)
|
||||
else:
|
||||
ans = lines_for_diff(path, other_path, diff_map[path], args, columns, margin_size)
|
||||
elif item_type == 'add':
|
||||
if is_binary:
|
||||
if is_img:
|
||||
ans = image_lines(None, path, columns, margin_size, image_manager)
|
||||
ans = lines_for_diff(path, other_path, diff_map[path], args, columns, margin_size)
|
||||
elif item_type == 'add':
|
||||
if is_binary:
|
||||
if is_img:
|
||||
ans = image_lines(None, path, columns, margin_size, image_manager)
|
||||
else:
|
||||
ans = yield_lines_from(binary_lines(None, path, columns, margin_size), item_ref)
|
||||
else:
|
||||
ans = yield_lines_from(binary_lines(None, path, columns, margin_size), item_ref)
|
||||
else:
|
||||
ans = all_lines(path, args, columns, margin_size, is_add=True)
|
||||
elif item_type == 'removal':
|
||||
if is_binary:
|
||||
if is_img:
|
||||
ans = image_lines(path, None, columns, margin_size, image_manager)
|
||||
ans = all_lines(path, args, columns, margin_size, is_add=True)
|
||||
elif item_type == 'removal':
|
||||
if is_binary:
|
||||
if is_img:
|
||||
ans = image_lines(path, None, columns, margin_size, image_manager)
|
||||
else:
|
||||
ans = yield_lines_from(binary_lines(path, None, columns, margin_size), item_ref)
|
||||
else:
|
||||
ans = yield_lines_from(binary_lines(path, None, columns, margin_size), item_ref)
|
||||
ans = all_lines(path, args, columns, margin_size, is_add=False)
|
||||
elif item_type == 'rename':
|
||||
ans = yield_lines_from(rename_lines(path, other_path, args, columns, margin_size), item_ref)
|
||||
else:
|
||||
ans = all_lines(path, args, columns, margin_size, is_add=False)
|
||||
elif item_type == 'rename':
|
||||
ans = yield_lines_from(rename_lines(path, other_path, args, columns, margin_size), item_ref)
|
||||
else:
|
||||
raise ValueError('Unsupported item type: {}'.format(item_type))
|
||||
yield from ans
|
||||
if i < last_item_num:
|
||||
yield Line('', item_ref)
|
||||
raise ValueError('Unsupported item type: {}'.format(item_type))
|
||||
yield from ans
|
||||
if i < last_item_num:
|
||||
yield Line('', item_ref)
|
||||
|
||||
|
||||
render_diff = RenderDiff()
|
||||
|
||||
@ -9,11 +9,12 @@ import sys
|
||||
from functools import lru_cache
|
||||
from gettext import gettext as _
|
||||
from itertools import repeat
|
||||
from typing import Callable, Dict, List, Optional, Tuple
|
||||
|
||||
from kitty.cli import parse_args
|
||||
from kitty.cli_stub import HintsCLIOptions
|
||||
from kitty.fast_data_types import set_clipboard_string
|
||||
from kitty.key_encoding import key_defs as K, backspace_key, enter_key
|
||||
from kitty.key_encoding import backspace_key, enter_key, key_defs as K
|
||||
from kitty.utils import screen_size_function
|
||||
|
||||
from ..tui.handler import Handler, result_handler
|
||||
@ -199,7 +200,7 @@ def regex_finditer(pat, minimum_match_length, text):
|
||||
|
||||
closing_bracket_map = {'(': ')', '[': ']', '{': '}', '<': '>', '*': '*', '"': '"', "'": "'"}
|
||||
opening_brackets = ''.join(closing_bracket_map)
|
||||
postprocessor_map = {}
|
||||
postprocessor_map: Dict[str, Callable[[str, int, int], Tuple[int, int]]] = {}
|
||||
|
||||
|
||||
def postprocessor(func):
|
||||
@ -304,8 +305,8 @@ def functions_for(args):
|
||||
return pattern, post_processors
|
||||
|
||||
|
||||
def convert_text(text, cols):
|
||||
lines = []
|
||||
def convert_text(text: str, cols: int) -> str:
|
||||
lines: List[str] = []
|
||||
empty_line = '\0' * cols
|
||||
for full_line in text.split('\n'):
|
||||
if full_line:
|
||||
@ -574,10 +575,11 @@ def handle_result(args, data, target_window_id, boss):
|
||||
matches, groupdicts = [], []
|
||||
for m, g in zip(data['match'], data['groupdicts']):
|
||||
if m:
|
||||
matches.append(m), groupdicts.append(g)
|
||||
matches.append(m)
|
||||
groupdicts.append(g)
|
||||
joiner = data['multiple_joiner']
|
||||
try:
|
||||
is_int = int(joiner)
|
||||
is_int: Optional[int] = int(joiner)
|
||||
except Exception:
|
||||
is_int = None
|
||||
text_type = data['type']
|
||||
|
||||
@ -9,14 +9,17 @@ import signal
|
||||
import sys
|
||||
import zlib
|
||||
from base64 import standard_b64encode
|
||||
from collections import namedtuple
|
||||
from functools import lru_cache
|
||||
from math import ceil
|
||||
from tempfile import NamedTemporaryFile
|
||||
from typing import Dict, List, NamedTuple, Optional, Union
|
||||
|
||||
from kitty.cli import parse_args
|
||||
from kitty.cli_stub import IcatCLIOptions
|
||||
from kitty.constants import appname
|
||||
from kitty.utils import TTYIO, fit_image, screen_size_function
|
||||
from kitty.utils import (
|
||||
TTYIO, ScreenSize, ScreenSizeGetter, fit_image, screen_size_function
|
||||
)
|
||||
|
||||
from ..tui.images import (
|
||||
ConvertFailed, NoImageMagick, OpenFailed, convert, fsenc, identify
|
||||
@ -104,27 +107,25 @@ colors. For example, --1 evaluates as -1,073,741,825.
|
||||
'''
|
||||
|
||||
|
||||
screen_size = None
|
||||
screen_size: Optional[ScreenSizeGetter] = None
|
||||
can_transfer_with_files = False
|
||||
|
||||
|
||||
def get_screen_size_function():
|
||||
def get_screen_size_function() -> ScreenSizeGetter:
|
||||
global screen_size
|
||||
if screen_size is None:
|
||||
screen_size = screen_size_function()
|
||||
return screen_size
|
||||
|
||||
|
||||
def get_screen_size():
|
||||
def get_screen_size() -> ScreenSize:
|
||||
screen_size = get_screen_size_function()
|
||||
return screen_size()
|
||||
|
||||
|
||||
def options_spec():
|
||||
if not hasattr(options_spec, 'ans'):
|
||||
options_spec.ans = OPTIONS.format(
|
||||
appname='{}-icat'.format(appname),
|
||||
)
|
||||
return options_spec.ans
|
||||
@lru_cache(maxsize=2)
|
||||
def options_spec() -> str:
|
||||
return OPTIONS.format(appname='{}-icat'.format(appname))
|
||||
|
||||
|
||||
def write_gr_cmd(cmd, payload=None):
|
||||
@ -196,7 +197,7 @@ def show(outfile, width, height, zindex, fmt, transmit_mode='t', align='center',
|
||||
set_cursor_for_place(place, cmd, width, height, align)
|
||||
else:
|
||||
set_cursor(cmd, width, height, align)
|
||||
if detect_support.has_files:
|
||||
if can_transfer_with_files:
|
||||
cmd['t'] = transmit_mode
|
||||
write_gr_cmd(cmd, standard_b64encode(os.path.abspath(outfile).encode(fsenc)))
|
||||
else:
|
||||
@ -250,20 +251,21 @@ def scan(d):
|
||||
|
||||
|
||||
def detect_support(wait_for=10, silent=False):
|
||||
global can_transfer_with_files
|
||||
if not silent:
|
||||
print('Checking for graphics ({}s max. wait)...'.format(wait_for), end='\r')
|
||||
sys.stdout.flush()
|
||||
try:
|
||||
received = b''
|
||||
responses = {}
|
||||
responses: Dict[int, bool] = {}
|
||||
|
||||
def parse_responses():
|
||||
for m in re.finditer(b'\033_Gi=([1|2]);(.+?)\033\\\\', received):
|
||||
iid = m.group(1)
|
||||
if iid in (b'1', b'2'):
|
||||
iid = int(iid.decode('ascii'))
|
||||
if iid not in responses:
|
||||
responses[iid] = m.group(2) == b'OK'
|
||||
iid_ = int(iid.decode('ascii'))
|
||||
if iid_ not in responses:
|
||||
responses[iid_] = m.group(2) == b'OK'
|
||||
|
||||
def more_needed(data):
|
||||
nonlocal received
|
||||
@ -280,16 +282,24 @@ def detect_support(wait_for=10, silent=False):
|
||||
finally:
|
||||
if not silent:
|
||||
sys.stdout.buffer.write(b'\033[J'), sys.stdout.flush()
|
||||
detect_support.has_files = bool(responses.get(2))
|
||||
can_transfer_with_files = bool(responses.get(2))
|
||||
return responses.get(1, False)
|
||||
|
||||
|
||||
def parse_place(raw):
|
||||
class Place(NamedTuple):
|
||||
width: int
|
||||
height: int
|
||||
left: int
|
||||
top: int
|
||||
|
||||
|
||||
def parse_place(raw: str) -> Optional[Place]:
|
||||
if raw:
|
||||
area, pos = raw.split('@', 1)
|
||||
w, h = map(int, area.split('x'))
|
||||
l, t = map(int, pos.split('x'))
|
||||
return namedtuple('Place', 'width height left top')(w, h, l, t)
|
||||
return Place(w, h, l, t)
|
||||
return None
|
||||
|
||||
|
||||
help_text = (
|
||||
@ -344,10 +354,12 @@ def process_single_item(item, args, url_pat=None, maybe_dir=True):
|
||||
|
||||
|
||||
def main(args=sys.argv):
|
||||
args, items = parse_args(args[1:], options_spec, usage, help_text, '{} +kitten icat'.format(appname), result_class=IcatCLIOptions)
|
||||
global can_transfer_with_files
|
||||
args, items_ = parse_args(args[1:], options_spec, usage, help_text, '{} +kitten icat'.format(appname), result_class=IcatCLIOptions)
|
||||
items: List[Union[str, bytes]] = list(items)
|
||||
|
||||
if args.print_window_size:
|
||||
screen_size_function.ans = None
|
||||
screen_size_function.cache_clear()
|
||||
with open(os.ctermid()) as tty:
|
||||
ss = screen_size_function(tty)()
|
||||
print('{}x{}'.format(ss.width, ss.height), end='')
|
||||
@ -384,16 +396,16 @@ def main(args=sys.argv):
|
||||
if args.detect_support:
|
||||
if not detect_support(wait_for=args.detection_timeout, silent=True):
|
||||
raise SystemExit(1)
|
||||
print('file' if detect_support.has_files else 'stream', end='', file=sys.stderr)
|
||||
print('file' if can_transfer_with_files else 'stream', end='', file=sys.stderr)
|
||||
return
|
||||
if args.transfer_mode == 'detect':
|
||||
if not detect_support(wait_for=args.detection_timeout, silent=args.silent):
|
||||
raise SystemExit('This terminal emulator does not support the graphics protocol, use a terminal emulator such as kitty that does support it')
|
||||
else:
|
||||
detect_support.has_files = args.transfer_mode == 'file'
|
||||
can_transfer_with_files = args.transfer_mode == 'file'
|
||||
errors = []
|
||||
if args.clear:
|
||||
sys.stdout.buffer.write(clear_images_on_screen(delete_data=True))
|
||||
sys.stdout.write(clear_images_on_screen(delete_data=True))
|
||||
if not items:
|
||||
return
|
||||
if not items:
|
||||
|
||||
@ -28,15 +28,15 @@ class KeysHandler(Handler):
|
||||
REPEAT: 'REPEAT',
|
||||
RELEASE: 'RELEASE'
|
||||
}[key_event.type]
|
||||
mods = []
|
||||
lmods = []
|
||||
for m, name in {
|
||||
SHIFT: 'Shift',
|
||||
ALT: 'Alt',
|
||||
CTRL: 'Ctrl',
|
||||
SUPER: 'Super'}.items():
|
||||
if key_event.mods & m:
|
||||
mods.append(name)
|
||||
mods = '+'.join(mods)
|
||||
lmods.append(name)
|
||||
mods = '+'.join(lmods)
|
||||
if mods:
|
||||
mods += '+'
|
||||
self.print('Key {}: {}{} [{}]'.format(etype, mods, key_event.key, encode_key_event(key_event)))
|
||||
|
||||
@ -207,7 +207,7 @@ def render_block(text):
|
||||
def as_conf_file(all_options):
|
||||
ans = ['# vim:fileencoding=utf-8:ft=conf:foldmethod=marker', '']
|
||||
a = ans.append
|
||||
current_group = None
|
||||
current_group: Optional[Group] = None
|
||||
num_open_folds = 0
|
||||
all_options = list(all_options)
|
||||
|
||||
@ -311,9 +311,10 @@ def as_type_stub(
|
||||
all_options: Dict[str, Union[Option, List[Shortcut]]],
|
||||
special_types: Optional[Dict[str, str]] = None,
|
||||
preamble_lines: Union[Tuple[str, ...], List[str], Iterator[str]] = (),
|
||||
extra_fields: Union[Tuple[Tuple[str, str], ...], List[Tuple[str, str]], Iterator[Tuple[str, str]]] = ()
|
||||
extra_fields: Union[Tuple[Tuple[str, str], ...], List[Tuple[str, str]], Iterator[Tuple[str, str]]] = (),
|
||||
class_name: str = 'Options'
|
||||
) -> str:
|
||||
ans = ['import typing\n'] + list(preamble_lines) + ['', 'class Options:']
|
||||
ans = ['import typing\n'] + list(preamble_lines) + ['', 'class {}:'.format(class_name)]
|
||||
imports: Set[Tuple[str, str]] = set()
|
||||
overrides = special_types or {}
|
||||
for name, val in all_options.items():
|
||||
@ -328,14 +329,14 @@ def as_type_stub(
|
||||
ans.append(' {}: {}'.format(field_name, type_def))
|
||||
ans.append(' def __iter__(self): pass')
|
||||
ans.append(' def __len__(self): pass')
|
||||
return '\n'.join(ans)
|
||||
return '\n'.join(ans) + '\n\n\n'
|
||||
|
||||
|
||||
def save_type_stub(text: str, fpath: str) -> None:
|
||||
import os
|
||||
with open(fpath + 'i', 'w') as f:
|
||||
print(
|
||||
'# Update this file by running: python {}'.format(os.path.relpath(os.path.abspath(fpath))),
|
||||
file=f
|
||||
)
|
||||
f.write(text)
|
||||
fpath += 'i'
|
||||
preamble = '# Update this file by running: python {}\n\n'.format(os.path.relpath(os.path.abspath(fpath)))
|
||||
existing = open(fpath).read()
|
||||
current = preamble + text
|
||||
if existing != current:
|
||||
open(fpath, 'w').write(current)
|
||||
|
||||
@ -7,6 +7,9 @@ class Options:
|
||||
pass
|
||||
|
||||
|
||||
DiffOptions = Options
|
||||
|
||||
|
||||
def generate_stub():
|
||||
from .config_data import all_options
|
||||
from .conf.definition import as_type_stub, save_type_stub
|
||||
@ -26,6 +29,10 @@ def generate_stub():
|
||||
('sequence_map', 'SequenceMap'),
|
||||
)
|
||||
)
|
||||
|
||||
from kittens.diff.config_data import all_options
|
||||
text += as_type_stub(all_options, class_name='DiffOptions')
|
||||
|
||||
save_type_stub(text, __file__)
|
||||
|
||||
|
||||
|
||||
@ -10,11 +10,10 @@ import os
|
||||
import re
|
||||
import string
|
||||
import sys
|
||||
from collections import namedtuple
|
||||
from contextlib import suppress
|
||||
from functools import lru_cache
|
||||
from time import monotonic
|
||||
from typing import Any, Dict, List, Optional, cast
|
||||
from typing import Any, Dict, List, NamedTuple, Optional, cast
|
||||
|
||||
from .constants import (
|
||||
appname, is_macos, is_wayland, shell_path, supports_primary_selection
|
||||
@ -81,7 +80,13 @@ def parse_color_set(raw):
|
||||
continue
|
||||
|
||||
|
||||
ScreenSize = namedtuple('ScreenSize', 'rows cols width height cell_width cell_height')
|
||||
class ScreenSize(NamedTuple):
|
||||
rows: int
|
||||
cols: int
|
||||
width: int
|
||||
height: int
|
||||
cell_width: int
|
||||
cell_height: int
|
||||
|
||||
|
||||
class ScreenSizeGetter:
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user