Add Kitty window and tab chooser with previews
This commit is contained in:
Petr Nyc
@@ -0,0 +1,121 @@
|
||||
"""Choose a tab grouped by native Kitty window, and focus its window."""
|
||||
|
||||
import json
|
||||
|
||||
from kittens.tui.handler import result_handler
|
||||
from kitty.fast_data_types import get_os_window_title
|
||||
|
||||
|
||||
def main(args):
|
||||
pass
|
||||
|
||||
|
||||
def snapshot_tab(tab, boss):
|
||||
"""Capture before opening the chooser, without focusing or relaying out tabs."""
|
||||
panes = []
|
||||
active = tab.active_window
|
||||
try:
|
||||
groups = list(tab.windows.iter_all_layoutable_groups(only_visible=True))
|
||||
except Exception:
|
||||
groups = []
|
||||
for group in groups:
|
||||
window = boss.window_id_map.get(group.active_window_id)
|
||||
if window is None:
|
||||
continue
|
||||
try:
|
||||
geometry = window.geometry
|
||||
pane = dict(id=window.id, title=window.title,
|
||||
active=window is active,
|
||||
rect=[geometry.left, geometry.top, geometry.right, geometry.bottom],
|
||||
lines=None)
|
||||
except Exception:
|
||||
continue
|
||||
try:
|
||||
pane['lines'] = [window.screen.visual_line(i).as_ansi()
|
||||
for i in range(window.screen.lines)]
|
||||
except Exception:
|
||||
pass
|
||||
panes.append(pane)
|
||||
return dict(title=tab.name or tab.title, panes=panes)
|
||||
|
||||
|
||||
@result_handler(no_ui=True)
|
||||
def handle_result(args, answer, target_window_id, boss):
|
||||
source = boss.window_id_map.get(target_window_id)
|
||||
current = source.tabref() if source is not None else boss.active_tab
|
||||
|
||||
def chosen(tab_id):
|
||||
if not isinstance(tab_id, int):
|
||||
return
|
||||
for tab in boss.all_tabs:
|
||||
if tab.id == tab_id:
|
||||
window = tab.active_window
|
||||
if window is not None:
|
||||
boss.set_active_window(window, switch_os_window_if_needed=True)
|
||||
return
|
||||
|
||||
entries = []
|
||||
snapshots = {}
|
||||
for index, (os_window_id, manager) in enumerate(boss.os_window_map.items(), 1):
|
||||
tabs = list(manager)
|
||||
active = manager.active_tab
|
||||
if not tabs or active is None:
|
||||
continue
|
||||
count = len(tabs)
|
||||
title = get_os_window_title(os_window_id) or f"Window {index}"
|
||||
label = f"- {title}: {count} {'tab' if count == 1 else 'tabs'}"
|
||||
if current in tabs:
|
||||
label += " (current)"
|
||||
tab_ids = [tab.id for tab in tabs]
|
||||
entries.append(dict(kind='window', label=label, tab_id=active.id,
|
||||
tabs=tab_ids, title=title))
|
||||
for tab_index, tab in enumerate(tabs, 1):
|
||||
branch = "└─>" if tab_index == count else "├─>"
|
||||
marker = "*" if tab is active else ""
|
||||
label = f"{branch} {tab_index}: {tab.name or tab.title}{marker}"
|
||||
if tab.num_window_groups > 1:
|
||||
label += f" [{tab.num_window_groups} panes]"
|
||||
entries.append(dict(kind='tab', label=label, tab_id=tab.id,
|
||||
title=tab.name or tab.title))
|
||||
snapshots[str(tab.id)] = snapshot_tab(tab, boss)
|
||||
selected = None
|
||||
|
||||
def done(answer, target_window_id, boss):
|
||||
nonlocal selected
|
||||
selected = answer
|
||||
|
||||
# Kittens overlay one pane. Temporarily use stack to fill the tab, keeping
|
||||
# the original layout object (including its split tree and size biases).
|
||||
layout_state = None
|
||||
chooser_layout = None
|
||||
if current is not None and current.current_layout.name != 'stack':
|
||||
layout_state = (current.current_layout, current._current_layout_name,
|
||||
current._last_used_layout)
|
||||
|
||||
def restore_layout():
|
||||
if (layout_state is not None and current in boss.all_tabs
|
||||
and current.current_layout is chooser_layout):
|
||||
current.current_layout, current._current_layout_name, current._last_used_layout = layout_state
|
||||
current.mark_tab_bar_dirty()
|
||||
current.relayout()
|
||||
|
||||
def removed(target_window_id, boss):
|
||||
restore_layout()
|
||||
chosen(selected)
|
||||
|
||||
try:
|
||||
if layout_state is not None:
|
||||
current._set_current_layout('stack')
|
||||
chooser_layout = current.current_layout
|
||||
current.relayout()
|
||||
overlay = boss.run_kitten_with_metadata(
|
||||
"window_picker.py", args=tuple(args[1:]),
|
||||
input_data=json.dumps(dict(entries=entries, snapshots=snapshots)),
|
||||
window=source,
|
||||
custom_callback=done, action_on_removal=removed,
|
||||
)
|
||||
if overlay is None:
|
||||
restore_layout()
|
||||
except Exception:
|
||||
restore_layout()
|
||||
raise
|
||||
@@ -3356,6 +3356,8 @@ map ctrl+a>ctrl+a send_text all \x01
|
||||
map ctrl+a>q focus_visible_window
|
||||
|
||||
# Tabs — equivalent to tmux windows
|
||||
map ctrl+a>s kitten choose_tab.py --collapsed
|
||||
map ctrl+a>w kitten choose_tab.py
|
||||
map ctrl+a>c new_tab
|
||||
map ctrl+a>n next_tab
|
||||
map --allow-fallback=shifted,ascii alt+n next_tab
|
||||
@@ -3375,7 +3377,6 @@ map ctrl+a>, set_tab_title
|
||||
|
||||
# Panes
|
||||
map ctrl+a>| combine : goto_layout splits : launch --location=vsplit
|
||||
map ctrl+a>s combine : goto_layout splits : launch --location=hsplit
|
||||
map ctrl+a>- combine : goto_layout splits : launch --location=hsplit
|
||||
map ctrl+a>x close_window
|
||||
map ctrl+a>h neighboring_window left
|
||||
|
||||
@@ -0,0 +1,3 @@
|
||||
hostname dabel.us.oracle.com re-userland-build-x01.us.oracle.com
|
||||
env PATH=/usr/gnu/bin:/usr/bin:/bin:/usr/sbin:/sbin
|
||||
interpreter /usr/bin/python3
|
||||
@@ -0,0 +1,327 @@
|
||||
"""Run with: kitty +runpy 'import runpy; runpy.run_path("test_window_picker.py", run_name="__main__")'."""
|
||||
|
||||
import io
|
||||
import json
|
||||
from pathlib import Path
|
||||
import runpy
|
||||
from types import SimpleNamespace as NS
|
||||
import unittest
|
||||
from unittest.mock import Mock, patch
|
||||
|
||||
from kitty.config import parse_config
|
||||
from kitty.fast_data_types import Screen, wcswidth
|
||||
from kitty.key_encoding import KeyEvent
|
||||
|
||||
ROOT = Path(__file__).resolve().parent
|
||||
ui = runpy.run_path(str(ROOT / 'window_picker.py'))
|
||||
launcher = runpy.run_path(str(ROOT / 'choose_tab.py'))
|
||||
|
||||
|
||||
def parse_bytes(screen, data):
|
||||
while data:
|
||||
buffer = screen.test_create_write_buffer()
|
||||
size = screen.test_commit_write_buffer(data, buffer)
|
||||
screen.test_parse_written_data()
|
||||
data = data[size:]
|
||||
|
||||
|
||||
def example_payload():
|
||||
entries = []
|
||||
snapshots = {}
|
||||
for group in range(2):
|
||||
ids = list(range(group * 6 + 1, group * 6 + 7))
|
||||
entries.append(dict(kind='window', label=f'- Project {group + 1}: 6 tabs',
|
||||
title=f'Project {group + 1}', tab_id=ids[0], tabs=ids))
|
||||
for index, tab_id in enumerate(ids):
|
||||
title = f'project-{tab_id}'
|
||||
entries.append(dict(kind='tab', label=f'└─> {index + 1}: {title}',
|
||||
title=title, tab_id=tab_id))
|
||||
panes = [dict(id=tab_id * 10, title='shell', active=True,
|
||||
rect=[0, 0, 100, 100],
|
||||
lines=['\x1b[32m$ build project\x1b[0m', 'result: OK', '❯'])]
|
||||
if index == 0:
|
||||
panes[0]['rect'] = [0, 0, 50, 100]
|
||||
panes.append(dict(id=tab_id * 10 + 1, title='logs', active=False,
|
||||
rect=[50, 0, 100, 100], lines=['\x1b[33mwarning\x1b[0m']))
|
||||
snapshots[str(tab_id)] = dict(title=title, panes=panes)
|
||||
return dict(entries=entries, snapshots=snapshots)
|
||||
|
||||
|
||||
class PickerTests(unittest.TestCase):
|
||||
def picker(self, collapsed=False, rows=30, cols=100):
|
||||
payload = example_payload()
|
||||
picker = ui['WindowPicker'](payload['entries'], collapsed, payload['snapshots'])
|
||||
picker.screen_size = NS(rows=rows, cols=cols)
|
||||
picker.write = Mock()
|
||||
picker.quit_loop = Mock()
|
||||
return picker
|
||||
|
||||
def rendered(self, picker):
|
||||
picker.draw()
|
||||
screen = Screen(None, picker.screen_size.rows, picker.screen_size.cols)
|
||||
parse_bytes(screen, picker.write.call_args.args[0].encode())
|
||||
return screen, '\n'.join(str(screen.line(i)) for i in range(screen.lines))
|
||||
|
||||
def test_window_vs_tab_preview(self):
|
||||
picker = self.picker()
|
||||
screen, text = self.rendered(picker)
|
||||
self.assertIn('Project 1 — snapshot', text)
|
||||
self.assertIn('1:project-1', text)
|
||||
self.assertIn('4:project-4', text)
|
||||
self.assertIn('warning', text)
|
||||
preview_text = '\n'.join(text.splitlines()[15:-1])
|
||||
self.assertEqual(preview_text.count('┌'), 5) # Frame + four tab badges.
|
||||
self.assertNotIn('││', preview_text)
|
||||
self.assertIn('[] page', text)
|
||||
self.assertIn('\x1b[32m', ''.join(screen.line(i).as_ansi() for i in range(screen.lines)))
|
||||
picker.move(1)
|
||||
_, text = self.rendered(picker)
|
||||
self.assertIn('project-1 — snapshot', text)
|
||||
preview_text = '\n'.join(text.splitlines()[15:])
|
||||
self.assertNotIn('4:project-4', preview_text)
|
||||
self.assertIn('│ 1 │', preview_text)
|
||||
self.assertIn('│ 2 │', preview_text)
|
||||
self.assertEqual(preview_text.count('┌'), 3) # Frame + two small badges.
|
||||
picker.move(1)
|
||||
_, text = self.rendered(picker)
|
||||
preview_text = '\n'.join(text.splitlines()[15:-1])
|
||||
self.assertEqual(preview_text.count('┌'), 2) # Single pane: frame + number badge.
|
||||
|
||||
def test_paging_does_not_change_selection(self):
|
||||
picker = self.picker()
|
||||
picker.on_text(']')
|
||||
self.assertEqual(picker.selected, 0)
|
||||
_, text = self.rendered(picker)
|
||||
self.assertIn('tabs 5–6/6 [2/2]', text)
|
||||
self.assertIn('5:project-5', text)
|
||||
picker.on_text(']')
|
||||
self.assertEqual(picker.preview_pages[id(picker.entries[0])], 1)
|
||||
picker.on_text('[')
|
||||
self.assertEqual(picker.preview_pages[id(picker.entries[0])], 0)
|
||||
picker.entries[0]['tab_id'] = 6
|
||||
picker.preview_pages.clear()
|
||||
_, text = self.rendered(picker)
|
||||
self.assertIn('6:project-6', text)
|
||||
|
||||
def test_resize_and_small_terminal(self):
|
||||
picker = self.picker()
|
||||
picker.on_text(']')
|
||||
picker.on_resize(NS(rows=24, cols=160))
|
||||
_, text = self.rendered(picker)
|
||||
self.assertIn('Project 1 — snapshot', text)
|
||||
self.assertNotIn('[2/2]', text)
|
||||
for rows, cols in ((17, 100), (30, 23), (1, 1)):
|
||||
picker.on_resize(NS(rows=rows, cols=cols))
|
||||
self.assertEqual(picker.preview_height(), 0)
|
||||
_, text = self.rendered(picker)
|
||||
self.assertNotIn('snapshot', text)
|
||||
|
||||
def test_folding_search_and_selection(self):
|
||||
picker = self.picker(collapsed=True)
|
||||
self.assertEqual(picker.visible_indices(), [0, 7])
|
||||
picker.on_text('j')
|
||||
self.assertEqual(picker.entries[picker.selected]['title'], 'Project 2')
|
||||
picker.on_key(KeyEvent(key='RIGHT'))
|
||||
self.assertEqual(len(picker.entries), 8)
|
||||
picker.on_text('/PROJECT-12')
|
||||
self.assertEqual(picker.entries[picker.selected]['tab_id'], 12)
|
||||
picker.on_key(KeyEvent(key='ENTER'))
|
||||
picker.quit_loop.assert_not_called()
|
||||
picker.on_key(KeyEvent(key='ENTER'))
|
||||
self.assertEqual(picker.answer, 12)
|
||||
picker.on_text('/not found')
|
||||
self.assertIn('no matches', picker.write.call_args.args[0])
|
||||
picker.on_key(KeyEvent(key='ESCAPE'))
|
||||
self.assertEqual(picker.query, 'PROJECT-12')
|
||||
picker.on_text('q')
|
||||
picker.quit_loop.assert_called_with(0)
|
||||
|
||||
def test_search_next_previous_and_arrow_navigation(self):
|
||||
picker = self.picker()
|
||||
picker.on_text('/project-')
|
||||
picker.on_key(KeyEvent(key='ENTER'))
|
||||
selected = picker.selected
|
||||
picker.on_text('n')
|
||||
self.assertGreater(picker.selected, selected)
|
||||
picker.on_text('N')
|
||||
self.assertEqual(picker.selected, selected)
|
||||
picker.on_key(KeyEvent(key='DOWN'))
|
||||
picker.on_key(KeyEvent(key='UP'))
|
||||
self.assertEqual(picker.selected, selected)
|
||||
|
||||
def test_unavailable_empty_and_tiny_panes(self):
|
||||
picker = self.picker()
|
||||
picker.selected = 1
|
||||
picker.snapshots['1']['panes'][0]['lines'] = None
|
||||
_, text = self.rendered(picker)
|
||||
self.assertIn('Preview unavailable', text)
|
||||
picker.snapshots.clear()
|
||||
_, text = self.rendered(picker)
|
||||
self.assertIn('Preview unavailable', text)
|
||||
empty = ui['WindowPicker']([])
|
||||
empty.screen_size = NS(rows=24, cols=80)
|
||||
empty.write = Mock()
|
||||
empty.draw()
|
||||
picker = self.picker(rows=18, cols=24)
|
||||
picker.snapshots['1']['panes'][0]['rect'] = [0, 0, 1, 100]
|
||||
picker.snapshots['1']['panes'][1]['rect'] = [1, 0, 100, 100]
|
||||
_, text = self.rendered(picker)
|
||||
self.assertIn('too small', text)
|
||||
|
||||
def test_nested_layout_and_clipping(self):
|
||||
preview = ui['Preview'](1, 80, 24)
|
||||
panes = [dict(id=i, title=str(i), active=i == 0, rect=rect,
|
||||
lines=['\x1b[31m' + '界' * 100] * 50)
|
||||
for i, rect in enumerate(([0, 0, 50, 50], [0, 50, 50, 100], [50, 0, 100, 100]))]
|
||||
preview.box(0, 0, 80, 24, 'Nested')
|
||||
preview.tab(dict(panes=panes), 1, 1, 78, 22)
|
||||
screen = Screen(None, 24, 80)
|
||||
parse_bytes(screen, ''.join(preview.output).encode())
|
||||
# Overlong colored lines cannot overwrite the outer border.
|
||||
self.assertTrue(str(screen.line(5)).endswith('│'))
|
||||
self.assertTrue(str(screen.line(23)).startswith('└'))
|
||||
|
||||
def test_safe_ansi_and_unicode_width(self):
|
||||
clip = ui['clip_ansi']
|
||||
result = clip('\x1b[32m界界abc\x1b[0m', 5)
|
||||
self.assertEqual(ui['clean_text'](result), '界界a')
|
||||
self.assertEqual(wcswidth(ui['clean_text'](result)), 5)
|
||||
self.assertIn('\x1b[32m', result)
|
||||
malicious = '\x1b]52;c;secret\x07\x1b[2J\x1bPpayload\x1b\\safe\x07'
|
||||
self.assertEqual(ui['clean_text'](clip(malicious, 80)), 'safe')
|
||||
self.assertNotIn('secret', clip(malicious, 80))
|
||||
self.assertNotIn('\x1b[2J', clip(malicious, 80))
|
||||
self.assertEqual(ui['clean_text'](clip('e\u0301界', 1)), 'e\u0301')
|
||||
|
||||
def test_stdin_payload_and_collapsed_flag(self):
|
||||
payload = example_payload()
|
||||
with patch.object(ui['Loop'], 'loop') as loop, patch('sys.stdin', io.StringIO(json.dumps(payload))):
|
||||
result = ui['main'](['window_picker.py', '--collapsed'])
|
||||
picker = loop.call_args.args[0]
|
||||
self.assertEqual(picker.visible_indices(), [0, 7])
|
||||
self.assertIsNone(result)
|
||||
|
||||
|
||||
class CaptureTests(unittest.TestCase):
|
||||
def fixture(self):
|
||||
events = []
|
||||
|
||||
def capture():
|
||||
events.append('capture')
|
||||
return '\x1b[32m$ hello'
|
||||
|
||||
window = NS(id=10, title='shell', geometry=NS(left=0, top=0, right=100, bottom=100),
|
||||
screen=NS(lines=2, visual_line=lambda i: NS(as_ansi=capture)))
|
||||
tab = NS(id=1, name='Tab', title='Tab', active_window=window, num_window_groups=1,
|
||||
windows=NS(iter_all_layoutable_groups=lambda **kw: [NS(active_window_id=10)]),
|
||||
current_layout=NS(name='splits', tree='original tree', biases=[0.3, 0.7]),
|
||||
_current_layout_name='splits', _last_used_layout='tall',
|
||||
relayout=Mock(), mark_tab_bar_dirty=Mock())
|
||||
|
||||
def set_layout(name):
|
||||
tab._last_used_layout = tab._current_layout_name
|
||||
tab._current_layout_name = name
|
||||
tab.current_layout = NS(name=name)
|
||||
|
||||
tab._set_current_layout = Mock(side_effect=set_layout)
|
||||
window.tabref = lambda: tab
|
||||
|
||||
class Manager:
|
||||
active_tab = tab
|
||||
|
||||
def __iter__(self):
|
||||
return iter([tab])
|
||||
|
||||
boss = NS(window_id_map={10: window}, active_tab=tab, all_tabs=[tab],
|
||||
os_window_map={99: Manager()}, set_active_window=Mock(),
|
||||
run_kitten_with_metadata=Mock(side_effect=lambda *a, **k: (events.append('launch'), NS())[1]))
|
||||
return events, window, tab, boss
|
||||
|
||||
def test_capture_before_launch_and_activation_after_removal(self):
|
||||
events, window, tab, boss = self.fixture()
|
||||
original = tab.current_layout
|
||||
with patch('kitty.fast_data_types.get_os_window_title', return_value='OS title'):
|
||||
handler = runpy.run_path(str(ROOT / 'choose_tab.py'))['handle_result']
|
||||
handler(['choose_tab.py', '--collapsed'], None, 10, boss)
|
||||
self.assertEqual(events, ['capture', 'capture', 'launch'])
|
||||
self.assertEqual(tab.current_layout.name, 'stack')
|
||||
call = boss.run_kitten_with_metadata.call_args
|
||||
self.assertEqual(call.kwargs['args'], ('--collapsed',))
|
||||
payload = json.loads(call.kwargs['input_data'])
|
||||
self.assertEqual(payload['entries'][0]['title'], 'OS title')
|
||||
self.assertEqual(payload['entries'][0]['kind'], 'window')
|
||||
self.assertEqual(payload['snapshots']['1']['panes'][0]['lines'], ['\x1b[32m$ hello'] * 2)
|
||||
call.kwargs['custom_callback'](1, 10, boss)
|
||||
boss.set_active_window.assert_not_called()
|
||||
call.kwargs['action_on_removal'](10, boss)
|
||||
self.assertIs(tab.current_layout, original)
|
||||
self.assertEqual((tab._current_layout_name, tab._last_used_layout), ('splits', 'tall'))
|
||||
boss.set_active_window.assert_called_once_with(window, switch_os_window_if_needed=True)
|
||||
boss.set_active_window.reset_mock()
|
||||
boss.all_tabs = []
|
||||
call.kwargs['action_on_removal'](10, boss)
|
||||
boss.set_active_window.assert_not_called()
|
||||
|
||||
def test_full_tab_cancel_and_existing_stack(self):
|
||||
events, window, tab, boss = self.fixture()
|
||||
original = tab.current_layout
|
||||
launcher['handle_result'](['choose_tab.py'], None, 10, boss)
|
||||
call = boss.run_kitten_with_metadata.call_args
|
||||
self.assertIs(call.kwargs['window'], window)
|
||||
self.assertEqual(tab.current_layout.name, 'stack')
|
||||
call.kwargs['custom_callback'](None, 10, boss)
|
||||
call.kwargs['action_on_removal'](10, boss)
|
||||
self.assertIs(tab.current_layout, original)
|
||||
boss.set_active_window.assert_not_called()
|
||||
tab.current_layout = NS(name='stack')
|
||||
tab._set_current_layout.reset_mock()
|
||||
launcher['handle_result'](['choose_tab.py'], None, 10, boss)
|
||||
tab._set_current_layout.assert_not_called()
|
||||
boss.run_kitten_with_metadata.call_args.kwargs['action_on_removal'](10, boss)
|
||||
self.assertEqual(tab.current_layout.name, 'stack')
|
||||
|
||||
def test_full_tab_launch_failure_restores_splits(self):
|
||||
for failed in (False, True):
|
||||
events, window, tab, boss = self.fixture()
|
||||
original = tab.current_layout
|
||||
if failed:
|
||||
boss.run_kitten_with_metadata.side_effect = RuntimeError('launch failed')
|
||||
with self.assertRaises(RuntimeError):
|
||||
launcher['handle_result'](['choose_tab.py'], None, 10, boss)
|
||||
else:
|
||||
boss.run_kitten_with_metadata.side_effect = None
|
||||
boss.run_kitten_with_metadata.return_value = None
|
||||
launcher['handle_result'](['choose_tab.py'], None, 10, boss)
|
||||
self.assertIs(tab.current_layout, original)
|
||||
self.assertEqual((tab._current_layout_name, tab._last_used_layout), ('splits', 'tall'))
|
||||
|
||||
def test_capture_failure_does_not_disable_chooser(self):
|
||||
events, window, tab, boss = self.fixture()
|
||||
window.screen.visual_line = Mock(side_effect=RuntimeError('closed'))
|
||||
snapshot = launcher['snapshot_tab'](tab, boss)
|
||||
self.assertIsNone(snapshot['panes'][0]['lines'])
|
||||
tab.windows.iter_all_layoutable_groups = Mock(side_effect=RuntimeError('closed'))
|
||||
self.assertEqual(launcher['snapshot_tab'](tab, boss)['panes'], [])
|
||||
|
||||
def test_capture_real_screen_colors(self):
|
||||
events, window, tab, boss = self.fixture()
|
||||
window.screen = Screen(None, 3, 40)
|
||||
parse_bytes(window.screen, b'\x1b[32mgreen prompt\x1b[0m\r\nnext row')
|
||||
snapshot = launcher['snapshot_tab'](tab, boss)
|
||||
lines = snapshot['panes'][0]['lines']
|
||||
self.assertIn('\x1b[32m', lines[0])
|
||||
self.assertEqual(ui['clean_text'](lines[0]), 'green prompt')
|
||||
self.assertEqual(ui['clean_text'](lines[1]), 'next row')
|
||||
|
||||
def test_config(self):
|
||||
errors = []
|
||||
with (ROOT / 'kitty.conf').open() as config:
|
||||
parsed = parse_config(config, accumulate_bad_lines=errors)
|
||||
self.assertEqual(errors, [])
|
||||
definitions = [m.definition for m in parsed['map']]
|
||||
self.assertIn('kitten choose_tab.py --collapsed', definitions)
|
||||
self.assertIn('kitten choose_tab.py', definitions)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main(argv=['test_window_picker.py'])
|
||||
@@ -0,0 +1,384 @@
|
||||
"""Keyboard-driven tree picker used by choose_tab.py."""
|
||||
|
||||
import json
|
||||
import re
|
||||
import sys
|
||||
import unicodedata
|
||||
|
||||
from kittens.tui.handler import Handler, result_handler
|
||||
from kittens.tui.loop import Loop
|
||||
from kitty.fast_data_types import truncate_point_for_length, wcswidth
|
||||
|
||||
|
||||
RESET = '\x1b[0m'
|
||||
CONTROLS = re.compile(
|
||||
r'\x1b(?:\][^\x07\x1b]*(?:\x07|\x1b\\)|[PX^_].*?\x1b\\|'
|
||||
r'\[[0-?]*[ -/]*[@-~]|[@-_])', re.DOTALL)
|
||||
SGR = re.compile(r'\x1b\[[0-9;:]*m\Z')
|
||||
|
||||
|
||||
def clean_text(text):
|
||||
text = CONTROLS.sub('', text)
|
||||
return ''.join(c for c in text if not unicodedata.category(c).startswith('C'))
|
||||
|
||||
|
||||
def clip_text(text, width):
|
||||
end = truncate_point_for_length(text, width)
|
||||
while end < len(text) and unicodedata.category(text[end]) in ('Mn', 'Me'):
|
||||
end += 1
|
||||
return text[:end]
|
||||
|
||||
|
||||
def clip_ansi(text, width):
|
||||
"""Retain only SGR formatting; never let a snapshot control the terminal."""
|
||||
result = []
|
||||
used = 0
|
||||
end = 0
|
||||
for match in CONTROLS.finditer(text):
|
||||
plain = clean_text(text[end:match.start()])
|
||||
piece = clip_text(plain, max(0, width - used))
|
||||
result.append(piece)
|
||||
used += wcswidth(piece)
|
||||
if len(piece) != len(plain):
|
||||
return ''.join(result) + RESET
|
||||
if SGR.fullmatch(match.group()):
|
||||
result.append(match.group())
|
||||
end = match.end()
|
||||
plain = clean_text(text[end:])
|
||||
result.append(clip_text(plain, max(0, width - used)))
|
||||
return ''.join(result) + RESET
|
||||
|
||||
|
||||
class Preview:
|
||||
"""Position a colored, cell-clipped preview inside a terminal rectangle."""
|
||||
|
||||
def __init__(self, top, width, height):
|
||||
self.top, self.width, self.height = top, width, height
|
||||
self.output = []
|
||||
|
||||
def put(self, x, y, text, width):
|
||||
if 0 <= x < self.width and 0 <= y < self.height:
|
||||
width = min(width, self.width - x)
|
||||
if width > 0:
|
||||
self.output.append(f'\x1b[{self.top + y};{x + 1}H{RESET}'
|
||||
+ clip_ansi(text, width))
|
||||
|
||||
def box(self, x, y, width, height, title='', active=False):
|
||||
if width < 2 or height < 2:
|
||||
return
|
||||
color = '\x1b[33m' if active else '\x1b[90m'
|
||||
self.put(x, y, color + '┌' + '─' * (width - 2) + '┐', width)
|
||||
for row in range(y + 1, y + height - 1):
|
||||
self.put(x, row, color + '│', 1)
|
||||
self.put(x + width - 1, row, color + '│', 1)
|
||||
self.put(x, y + height - 1, color + '└' + '─' * (width - 2) + '┘', width)
|
||||
if title:
|
||||
self.put(x + 1, y, color + ' ' + clean_text(title) + ' ', width - 2)
|
||||
|
||||
def badge(self, x, y, width, height, label, active):
|
||||
if width < 5 or height < 3:
|
||||
return
|
||||
label = clip_text(clean_text(label), width - 4)
|
||||
badge_width = wcswidth(label) + 4
|
||||
bx = x + (width - badge_width) // 2
|
||||
by = y + (height - 3) // 2
|
||||
self.box(bx, by, badge_width, 3)
|
||||
color = '\x1b[33m' if active else '\x1b[34m'
|
||||
self.put(bx + 1, by + 1, color + ' ' + label + ' ', badge_width - 2)
|
||||
|
||||
def tab(self, snapshot, x, y, width, height, title=None, active=False):
|
||||
panes = snapshot.get('panes', []) if snapshot else []
|
||||
if not panes:
|
||||
self.put(x, y, 'Preview unavailable', width)
|
||||
if title is not None:
|
||||
self.badge(x, y, width, height, title, active)
|
||||
return
|
||||
left = min(p['rect'][0] for p in panes)
|
||||
top = min(p['rect'][1] for p in panes)
|
||||
right = max(p['rect'][2] for p in panes)
|
||||
bottom = max(p['rect'][3] for p in panes)
|
||||
small = 0
|
||||
dividers = []
|
||||
badges = []
|
||||
for index, pane in enumerate(panes, 1):
|
||||
l, t, r, b = pane['rect']
|
||||
px = x + round((l - left) * width / max(1, right - left))
|
||||
py = y + round((t - top) * height / max(1, bottom - top))
|
||||
pw = x + round((r - left) * width / max(1, right - left)) - px
|
||||
ph = y + round((b - top) * height / max(1, bottom - top)) - py
|
||||
if l > left:
|
||||
dividers.extend((px, row, '│', 1) for row in range(py, py + ph))
|
||||
px, pw = px + 1, pw - 1
|
||||
if t > top:
|
||||
dividers.append((px, py, '─' * max(0, pw), pw))
|
||||
py, ph = py + 1, ph - 1
|
||||
if pw < 1 or ph < 1:
|
||||
small += 1
|
||||
continue
|
||||
lines = pane.get('lines')
|
||||
for row, line in enumerate((lines if lines is not None else ['Preview unavailable'])[:ph]):
|
||||
self.put(px, py + row, line, pw)
|
||||
if title is None:
|
||||
badges.append((px, py, pw, ph, str(index), pane['active']))
|
||||
for dx, dy, text, count in dividers:
|
||||
self.put(dx, dy, '\x1b[90m' + text, count)
|
||||
for badge in badges:
|
||||
self.badge(*badge)
|
||||
if title is not None:
|
||||
self.badge(x, y, width, height, title, active)
|
||||
if small:
|
||||
self.put(x, y + height - 1, f'{small} pane(s) too small', width)
|
||||
|
||||
|
||||
class WindowPicker(Handler):
|
||||
def __init__(self, entries, collapsed=False, snapshots=None):
|
||||
self.all_entries = entries
|
||||
self.groups = {i for i, entry in enumerate(entries) if entry['kind'] == 'window'}
|
||||
self.collapsed = set(self.groups) if collapsed else set()
|
||||
self.selected = 0
|
||||
self.offset = 0
|
||||
self.answer = None
|
||||
self.query = ''
|
||||
self.searching = False
|
||||
self.search_origin = ('', 0)
|
||||
self.snapshots = snapshots or {}
|
||||
self.preview_pages = {}
|
||||
|
||||
def visible_indices(self):
|
||||
hidden = False
|
||||
visible = []
|
||||
for index in range(len(self.all_entries)):
|
||||
if index in self.groups:
|
||||
hidden = index in self.collapsed
|
||||
visible.append(index)
|
||||
elif not hidden:
|
||||
visible.append(index)
|
||||
return visible
|
||||
|
||||
@property
|
||||
def entries(self):
|
||||
return [self.all_entries[i] for i in self.visible_indices()]
|
||||
|
||||
def expand_group(self, expand):
|
||||
visible = self.visible_indices()
|
||||
if not visible:
|
||||
return
|
||||
index = visible[self.selected]
|
||||
if index not in self.groups:
|
||||
return
|
||||
if expand:
|
||||
self.collapsed.discard(index)
|
||||
else:
|
||||
self.collapsed.add(index)
|
||||
self.draw()
|
||||
|
||||
def matches(self):
|
||||
if not self.query:
|
||||
return []
|
||||
query = self.query.casefold()
|
||||
return [i for i, entry in enumerate(self.entries)
|
||||
if query in clean_text(entry['label']).casefold()]
|
||||
|
||||
def update_search(self):
|
||||
matches = self.matches()
|
||||
if matches and self.selected not in matches:
|
||||
self.selected = next((i for i in matches if i >= self.selected), matches[0])
|
||||
self.draw()
|
||||
|
||||
def next_match(self, delta):
|
||||
matches = self.matches()
|
||||
if matches:
|
||||
ordered = matches if delta > 0 else list(reversed(matches))
|
||||
self.selected = next(
|
||||
(i for i in ordered if (i - self.selected) * delta > 0), ordered[0])
|
||||
self.draw()
|
||||
|
||||
def initialize(self):
|
||||
self.draw()
|
||||
|
||||
def preview_height(self):
|
||||
if self.screen_size.rows < 18 or self.screen_size.cols < 24:
|
||||
return 0
|
||||
return (self.screen_size.rows - 1) // 2
|
||||
|
||||
def page_preview(self, delta):
|
||||
if self.entries and self.preview_height():
|
||||
entry = self.entries[self.selected]
|
||||
if entry['kind'] == 'window':
|
||||
capacity = max(1, (self.screen_size.cols - 2) // 24)
|
||||
count = len(entry['tabs'])
|
||||
default = entry['tabs'].index(entry['tab_id']) // capacity
|
||||
page = self.preview_pages.get(id(entry), default)
|
||||
self.preview_pages[id(entry)] = max(0, min((count - 1) // capacity, page + delta))
|
||||
self.draw()
|
||||
|
||||
def draw_preview(self, height):
|
||||
preview = Preview(self.screen_size.rows - height, self.screen_size.cols, height)
|
||||
if not self.entries:
|
||||
preview.box(0, 0, preview.width, height, 'Snapshot')
|
||||
return ''.join(preview.output)
|
||||
entry = self.entries[self.selected]
|
||||
tab_ids = entry.get('tabs', [entry['tab_id']])
|
||||
capacity = max(1, (preview.width - 2) // 24) if entry['kind'] == 'window' else 1
|
||||
pages = max(1, (len(tab_ids) + capacity - 1) // capacity)
|
||||
default = tab_ids.index(entry['tab_id']) // capacity
|
||||
page = min(pages - 1, self.preview_pages.get(id(entry), default))
|
||||
self.preview_pages[id(entry)] = page
|
||||
start = page * capacity
|
||||
shown = tab_ids[start:start + capacity]
|
||||
label = f'{entry["title"]} — snapshot'
|
||||
if pages > 1:
|
||||
label += f' — tabs {start + 1}–{start + len(shown)}/{len(tab_ids)} [{page + 1}/{pages}]'
|
||||
preview.box(0, 0, preview.width, height, label)
|
||||
inner_width = preview.width - 2
|
||||
for offset, tab_id in enumerate(shown):
|
||||
left = 1 + offset * inner_width // len(shown)
|
||||
right = 1 + (offset + 1) * inner_width // len(shown)
|
||||
if offset:
|
||||
for row in range(1, height - 1):
|
||||
preview.put(left, row, '\x1b[90m│', 1)
|
||||
left += 1
|
||||
snapshot = self.snapshots.get(str(tab_id))
|
||||
title = None
|
||||
if entry['kind'] == 'window':
|
||||
title = f'{start + offset + 1}:{snapshot["title"] if snapshot else "Tab"}'
|
||||
active = tab_id == entry['tab_id']
|
||||
preview.tab(snapshot, left, 1, right - left, height - 2,
|
||||
title, active)
|
||||
return ''.join(preview.output)
|
||||
|
||||
def draw(self):
|
||||
height = self.preview_height()
|
||||
rows = max(1, self.screen_size.rows - 1 - height)
|
||||
cols = max(1, self.screen_size.cols)
|
||||
self.offset = min(self.offset, self.selected)
|
||||
self.offset = max(self.offset, self.selected - rows + 1)
|
||||
output = ["\x1b[?25l\x1b[H\x1b[2J"]
|
||||
matches = self.matches()
|
||||
visible = self.visible_indices()
|
||||
for index in range(self.offset, min(len(self.entries), self.offset + rows)):
|
||||
label = self.entries[index]['label']
|
||||
if visible[index] in self.collapsed:
|
||||
label = '+' + label[1:]
|
||||
# Titles are untrusted terminal text: strip control sequences/characters.
|
||||
label = clean_text(label)
|
||||
line = f"({index}) {label}"
|
||||
line = line[:truncate_point_for_length(line, cols)]
|
||||
style = "\x1b[1;33m" if index in matches else ""
|
||||
if index == self.selected:
|
||||
style += "\x1b[7m"
|
||||
output.append(f"{style}{line}\x1b[K\x1b[0m\r\n")
|
||||
if height:
|
||||
output.append(self.draw_preview(height))
|
||||
footer = 'j/k ↑/↓ move h/l ←/→ fold / search'
|
||||
footer += ' [] page' if height else ''
|
||||
footer += ' Enter select Esc cancel'
|
||||
if not height:
|
||||
footer += ' Preview needs 24 cols × 18 rows'
|
||||
if self.searching or self.query:
|
||||
position = str(matches.index(self.selected) + 1) if self.selected in matches else '-'
|
||||
status = f"{position}/{len(matches)}" if matches else "no matches"
|
||||
help_text = "Enter: finish search Esc: undo" if self.searching else "n/N: next/previous /: search"
|
||||
footer = f"/{self.query} [{status}] {help_text}"
|
||||
footer = footer[:truncate_point_for_length(footer, cols)]
|
||||
output.append(f"\x1b[0m\x1b[{self.screen_size.rows};1H{footer}")
|
||||
self.write(''.join(output))
|
||||
|
||||
def move(self, delta):
|
||||
self.selected = max(0, min(len(self.entries) - 1, self.selected + delta))
|
||||
self.draw()
|
||||
|
||||
def on_text(self, text, in_bracketed_paste=False):
|
||||
if len(text) > 1 and not in_bracketed_paste:
|
||||
for char in text:
|
||||
self.on_text(char)
|
||||
return
|
||||
if self.searching:
|
||||
self.query += ''.join(c for c in text if not unicodedata.category(c).startswith('C'))
|
||||
self.update_search()
|
||||
elif not in_bracketed_paste:
|
||||
if text == '/':
|
||||
self.search_origin = (self.query, self.selected)
|
||||
self.query = ''
|
||||
self.searching = True
|
||||
self.draw()
|
||||
elif text == 'n':
|
||||
self.next_match(1)
|
||||
elif text == 'N':
|
||||
self.next_match(-1)
|
||||
elif text == 'j':
|
||||
self.move(1)
|
||||
elif text == 'k':
|
||||
self.move(-1)
|
||||
elif text == 'h':
|
||||
self.expand_group(False)
|
||||
elif text == 'l':
|
||||
self.expand_group(True)
|
||||
elif text == '[':
|
||||
self.page_preview(-1)
|
||||
elif text == ']':
|
||||
self.page_preview(1)
|
||||
elif text == 'q':
|
||||
self.quit_loop(0)
|
||||
|
||||
def on_key(self, key):
|
||||
if key.is_release:
|
||||
return
|
||||
if self.searching:
|
||||
if key.matches('enter'):
|
||||
self.searching = False
|
||||
self.draw()
|
||||
return
|
||||
if key.matches('escape'):
|
||||
self.query, self.selected = self.search_origin
|
||||
self.searching = False
|
||||
self.draw()
|
||||
return
|
||||
if key.matches('backspace'):
|
||||
self.query = self.query[:-1]
|
||||
self.update_search()
|
||||
return
|
||||
if key.matches('ctrl+u'):
|
||||
self.query = ''
|
||||
self.update_search()
|
||||
return
|
||||
if key.matches('left') or key.matches('right'):
|
||||
return
|
||||
if key.matches('down'):
|
||||
self.move(1)
|
||||
elif key.matches('up'):
|
||||
self.move(-1)
|
||||
elif key.matches('left'):
|
||||
self.expand_group(False)
|
||||
elif key.matches('right'):
|
||||
self.expand_group(True)
|
||||
elif key.matches('enter'):
|
||||
if self.entries:
|
||||
self.answer = self.entries[self.selected]['tab_id']
|
||||
self.quit_loop(0)
|
||||
elif key.matches('escape'):
|
||||
self.quit_loop(0)
|
||||
|
||||
def on_resize(self, size):
|
||||
super().on_resize(size)
|
||||
self.draw()
|
||||
|
||||
def on_interrupt(self):
|
||||
self.quit_loop(0)
|
||||
|
||||
def on_eot(self):
|
||||
self.quit_loop(0)
|
||||
|
||||
|
||||
def main(args):
|
||||
payload = json.load(sys.stdin)
|
||||
picker = WindowPicker(payload['entries'], collapsed='--collapsed' in args[1:],
|
||||
snapshots=payload['snapshots'])
|
||||
Loop().loop(picker)
|
||||
return picker.answer
|
||||
|
||||
|
||||
@result_handler()
|
||||
def handle_result(args, answer, target_window_id, boss):
|
||||
# choose_tab.py supplies the callback and activates after this overlay closes.
|
||||
pass
|
||||
Reference in New Issue
Block a user