CERBERUS/kitty
1
0
Fork 0
mirror of https://github.com/kovidgoyal/kitty.git synced 2026-05-13 16:37:27 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 3

Start work on goto_session action

Browse source
This commit is contained in:
Kovid Goyal 2025-08-14 19:43:14 +05:30
parent 1544cab96f
commit 035ce949b4
No known key found for this signature in database
GPG key ID: 06BC317B515ACE7C
10 changed files with 130 additions and 22 deletions
Download patch file Download diff file Expand all files Collapse all files

18
kitty/boss.py
View file

@ -122,7 +122,7 @@ from .notifications import NotificationManager
from .options.types import Options, nullable_colors
from .options.utils import MINIMUM_FONT_SIZE, KeyboardMode, KeyDefinition
from .os_window_size import initial_window_size_func
from .session import Session, create_sessions, get_os_window_sizing_data
from .session import Session, create_sessions, get_os_window_sizing_data, goto_session
from .shaders import load_shader_programs
from .simple_cli_definitions import grab_keyboard_docs
from .tabs import SpecialWindow, SpecialWindowInstance, Tab, TabDict, TabManager
@ -601,8 +601,14 @@ class Boss:
if tab is not self.active_tab:
tm.set_active_tab(tab, for_keep_focus=window.tabref() if for_keep_focus else None)
tab.set_active_window(w, for_keep_focus=window if for_keep_focus else None)
if activation_token or (switch_os_window_if_needed and current_focused_os_window_id() != os_window_id):
focus_os_window(os_window_id, True, activation_token)
if switch_os_window_if_needed and current_focused_os_window_id() != os_window_id:
if activation_token or not is_wayland():
focus_os_window(os_window_id, True, activation_token)
else:
def doit(token: str = '') -> None:
focus_os_window(os_window_id, True, token)
if not run_with_activation_token(doit):
doit()
return os_window_id
return None
@ -877,7 +883,7 @@ class Boss:
args.session = 'none'
else:
from .session import PreReadSession
args.session = PreReadSession(data['session_data'], data['environ'])
args.session = PreReadSession(data['session_data'], data['environ'], data['session_arg'])
else:
args.session = ''
if not os.path.isabs(args.directory):
@ -2986,6 +2992,10 @@ class Boss:
)
return q if isinstance(q, Window) else None
@ac('misc', 'Switch to the specified session, creating it if not already present.')
def goto_session(self, cmdline: list[str]) -> None:
goto_session(self, cmdline)
@ac('tab', 'Interactively select a tab to switch to')
def select_tab(self) -> None:

3
kitty/launcher/single-instance.c
View file

@ -257,6 +257,9 @@ talk_to_instance(int s, struct sockaddr_un *server_addr, int argc, char *argv[],
w("{\"cmd\":\"new_instance\",\"session_data\":");
if (session_data.used) write_json_string(&output, session_data.data, session_data.used);
else write_json_string(&output, "", 0);
w(",\"session_arg\":");
if (opts->session && opts->session[0]) write_json_string(&output, opts->session, strlen(opts->session));
else write_json_string(&output, "", 0);
w(",\"args\":"); write_json_string_array(&output, argc, argv);
char cwd[4096];
if (!getcwd(cwd, sizeof(cwd))) fail_on_errno("Failed to get cwd");

2
kitty/main.py
View file

@ -535,7 +535,7 @@ def kitty_main(called_from_panel: bool = False) -> None:
if cli_opts.detach:
if cli_opts.session == '-':
from .session import PreReadSession
cli_opts.session = PreReadSession(sys.stdin.read(), os.environ)
cli_opts.session = PreReadSession(sys.stdin.read(), os.environ, '-')
if cli_opts.replay_commands:
from kitty.client import main as client_main
client_main(cli_opts.replay_commands)

2
kitty/options/utils.py
View file

@ -77,7 +77,7 @@ class InvalidMods(ValueError):
@func_with_args(
'pass_selection_to_program', 'new_window', 'new_tab', 'new_os_window',
'new_window_with_cwd', 'new_tab_with_cwd', 'new_os_window_with_cwd',
'launch', 'mouse_handle_click', 'show_error',
'launch', 'mouse_handle_click', 'show_error', 'goto_session',
)
def shlex_parse(func: str, rest: str) -> FuncArgsType:
return func, to_cmdline(rest)

6
kitty/rc/focus_tab.py
View file

@ -33,7 +33,11 @@ using this option means that you will not be notified of failures.
def response_from_kitty(self, boss: Boss, window: Window | None, payload_get: PayloadGetType) -> ResponseType:
for tab in self.tabs_for_match_payload(boss, window, payload_get):
if tab:
boss.set_active_tab(tab)
window = tab.active_window
if window:
boss.set_active_window(window, switch_os_window_if_needed=True)
else:
boss.set_active_tab(tab)
break
return None

6
kitty/rc/focus_window.py
View file

@ -4,8 +4,6 @@
from typing import TYPE_CHECKING
from kitty.fast_data_types import focus_os_window
from .base import MATCH_WINDOW_OPTION, ArgsType, Boss, PayloadGetType, PayloadType, RCOptions, RemoteCommand, ResponseType, Window
if TYPE_CHECKING:
@ -33,9 +31,7 @@ the command will exit with a success code.
def response_from_kitty(self, boss: Boss, window: Window | None, payload_get: PayloadGetType) -> ResponseType:
for window in self.windows_for_match_payload(boss, window, payload_get):
if window:
os_window_id = boss.set_active_window(window)
if os_window_id:
focus_os_window(os_window_id, True)
boss.set_active_window(window, switch_os_window_if_needed=True)
break
return None

105
kitty/session.py
View file

@ -8,14 +8,16 @@ import sys
from collections.abc import Callable, Generator, Iterator, Mapping
from contextlib import suppress
from functools import partial
from typing import TYPE_CHECKING, Any, Optional, Union
from gettext import gettext as _
from typing import TYPE_CHECKING, Any, Optional, Sequence, Union
from .cli_stub import CLIOptions
from .fast_data_types import get_options
from .layout.interface import all_layouts
from .options.types import Options
from .options.utils import resize_window, to_layout_names, window_size
from .os_window_size import WindowSize, WindowSizeData, WindowSizes
from .typing_compat import SpecialWindowInstance
from .typing_compat import BossType, SpecialWindowInstance, WindowType
from .utils import expandvars, log_error, resolve_custom_file, resolved_shell, shlex_split
if TYPE_CHECKING:
@ -73,6 +75,9 @@ class Tab:
class Session:
session_name: str = ''
num_of_windows_in_definition: int = 0
def __init__(self, default_title: str | None = None):
self.tabs: list[Tab] = []
self.active_tab_idx = 0
@ -167,9 +172,21 @@ class Session:
self.tabs[-1].cwd = val
def parse_session(raw: str, opts: Options, environ: Mapping[str, str] | None = None) -> Generator[Session, None, None]:
def session_arg_to_name(session_arg: str) -> str:
if session_arg in ('-', '/dev/stdin', 'none'):
session_arg = ''
session_name = os.path.basename(session_arg)
if session_name.rpartition('.')[2] in ('session', 'kitty-session'):
session_name = session_name.rpartition('.')[0]
return session_name
def parse_session(raw: str, opts: Options, environ: Mapping[str, str] | None = None, session_arg: str = '') -> Generator[Session, None, None]:
session_name = session_arg_to_name(session_arg)
def finalize_session(ans: Session) -> Session:
ans.session_name = session_name
ans.num_of_windows_in_definition = sum(len(t.windows) for t in ans.tabs)
from .tabs import SpecialWindow
for t in ans.tabs:
if not t.windows:
@ -234,10 +251,13 @@ def parse_session(raw: str, opts: Options, environ: Mapping[str, str] | None = N
class PreReadSession(str):
def __new__(cls, val: str, associated_environ: Mapping[str, str]) -> 'PreReadSession':
associated_environ: Mapping[str, str]
session_arg: str
def __new__(cls, val: str, associated_environ: Mapping[str, str], session_arg: str) -> 'PreReadSession':
ans: PreReadSession = str.__new__(cls, val)
ans.pre_read = True # type: ignore
ans.associated_environ = associated_environ # type: ignore
ans.associated_environ = associated_environ
ans.session_arg = session_arg
return ans
@ -254,10 +274,12 @@ def create_sessions(
if args.session == "none":
default_session = "none"
else:
session_arg = args.session
environ: Mapping[str, str] | None = None
if isinstance(args.session, PreReadSession):
session_data = '' + str(args.session)
environ = args.session.associated_environ # type: ignore
environ = args.session.associated_environ
session_arg = args.session.session_arg
else:
if args.session == '-':
f = sys.stdin
@ -265,16 +287,17 @@ def create_sessions(
f = open(resolve_custom_file(args.session))
with f:
session_data = f.read()
yield from parse_session(session_data, opts, environ=environ)
yield from parse_session(session_data, opts, environ=environ, session_arg=session_arg)
return
if default_session and default_session != 'none' and not getattr(args, 'args', None):
session_arg = session_arg_to_name(default_session)
try:
with open(default_session) as f:
session_data = f.read()
except OSError:
log_error(f'Failed to read from session file, ignoring: {default_session}')
else:
yield from parse_session(session_data, opts)
yield from parse_session(session_data, opts, session_arg=session_arg)
return
ans = Session()
current_layout = opts.enabled_layouts[0] if opts.enabled_layouts else 'tall'
@ -290,3 +313,67 @@ def create_sessions(
special_window = SpecialWindow(cmd, cwd_from=cwd_from, cwd=cwd, env=env_when_no_session, hold=bool(args and args.hold))
ans.add_special_window(special_window)
yield ans
def window_for_session_name(boss: BossType, session_name: str) -> WindowType | None:
windows = [w for w in boss.all_windows if w.created_in_session_name == session_name]
if not windows:
tabs = (t for t in boss.all_tabs if t.created_in_session_name == session_name)
windows = [t.active_window for t in tabs if t.active_window]
if not windows:
os_windows = (tm for tm in boss.all_tab_managers if tm.created_in_session_name == session_name)
windows = [tm.active_window for tm in os_windows if tm.active_window]
if windows:
def skey(w: WindowType) -> float:
return w.last_focused_at
windows.sort(key=skey, reverse=True)
return windows[0]
return None
def create_session(boss: BossType, path: str) -> None:
for i, s in enumerate(create_sessions(get_options(), default_session=path)):
if i == 0:
if s.num_of_windows_in_definition == 0: # leading new_os_window
continue
tm = boss.active_tab_manager
if tm is None:
boss.add_os_window(s)
else:
boss.add_os_window(s, os_window_id=tm.os_window_id)
else:
boss.add_os_window(s)
def goto_session(boss: BossType, cmdline: Sequence[str]) -> None:
if not cmdline:
boss.show_error('TODO: implement interactive goto_session', 'implement me')
return
path = cmdline[0]
if len(cmdline) == 1:
try:
idx = int(path)
except Exception:
idx = 0
if idx < 0:
boss.show_error('TODO: implement goto_session prev', 'implement me')
return
else:
for x in cmdline:
if not x.startswith('-'):
path = x
break
session_name = session_arg_to_name(path)
if not session_name:
boss.show_error(_('Invalid session'), _('{} is not a valid path for a session').format(path))
return
w = window_for_session_name(boss, session_name)
if w is not None:
boss.set_active_window(w, switch_os_window_if_needed=True)
return
try:
create_session(boss, path)
except Exception:
import traceback
tb = traceback.format_exc()
boss.show_error(_('Failed to create session'), _('Could not create session from {0} with error: {1}').format(path, tb))

8
kitty/tabs.py
View file

@ -130,6 +130,7 @@ class Tab: # {{{
has_indeterminate_progress: bool = False
last_focused_window_with_progress_id: int = 0
allow_relayouts: bool = True
created_in_session_name: str = ''
def __init__(
self,
@ -257,6 +258,8 @@ class Tab: # {{{
else:
from .launch import launch
launched_window = launch(boss, spec.opts, spec.args, target_tab=target_tab, force_target_tab=True)
if launched_window is not None:
launched_window.created_in_session_name = self.created_in_session_name
if window.resize_spec is not None:
self.resize_window(*window.resize_spec)
if window.focus_matching_window_spec:
@ -975,6 +978,7 @@ class TabManager: # {{{
def __init__(self, os_window_id: int, args: CLIOptions, wm_class: str, wm_name: str, startup_session: SessionType | None = None):
self.os_window_id = os_window_id
self.wm_class = wm_class
self.created_in_session_name = startup_session.session_name if startup_session else ''
self.recent_mouse_events: Deque[TabMouseEvent] = deque()
self.wm_name = wm_name
self.args = args
@ -986,7 +990,9 @@ class TabManager: # {{{
if startup_session is not None:
for t in startup_session.tabs:
self._add_tab(Tab(self, session_tab=t))
tab = Tab(self, session_tab=t)
tab.created_in_session_name = startup_session.session_name
self._add_tab(tab)
self._set_active_tab(max(0, min(startup_session.active_tab_idx, len(self.tabs) - 1)))
@property

1
kitty/types.py
View file

@ -18,6 +18,7 @@ class SingleInstanceData(TypedDict):
cmdline_args_for_open: Sequence[str]
cwd: str
session_data: str
session_arg: str
environ: Mapping[str, str]
notify_on_os_window_death: str | None

1
kitty/window.py
View file

@ -652,6 +652,7 @@ class Window:
initial_ignore_focus_changes: bool = False
initial_ignore_focus_changes_context_manager_in_operation: bool = False
creation_spec: WindowCreationSpec | None = None
created_in_session_name: str = ''
@classmethod
@contextmanager