feat: terminal backend abstraction — kitty, tmux, and plain backends
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,31 @@
|
||||
"""Terminal backend abstraction — detect and adapt to kitty, tmux, or plain terminal."""
|
||||
from __future__ import annotations
|
||||
import os
|
||||
from abc import ABC, abstractmethod
|
||||
from typing import Union
|
||||
|
||||
class Backend(ABC):
|
||||
@abstractmethod
|
||||
def launch_pane(self, command: list[str], title: str) -> Union[int, str]:
|
||||
"""Split the terminal and run command in new pane. Returns pane ID."""
|
||||
@abstractmethod
|
||||
def close_pane(self, pane_id: Union[int, str]) -> None:
|
||||
"""Close the pane."""
|
||||
@abstractmethod
|
||||
def image_protocol(self) -> str:
|
||||
"""Return supported image protocol: 'kitty', 'sixel', or 'none'."""
|
||||
@property
|
||||
@abstractmethod
|
||||
def name(self) -> str:
|
||||
"""Backend name for display/logging."""
|
||||
|
||||
def detect_backend() -> Backend:
|
||||
if os.environ.get("KITTY_PID"):
|
||||
from kitty_workbench.backends.kitty import KittyBackend
|
||||
return KittyBackend()
|
||||
elif os.environ.get("TMUX"):
|
||||
from kitty_workbench.backends.tmux import TmuxBackend
|
||||
return TmuxBackend()
|
||||
else:
|
||||
from kitty_workbench.backends.plain import PlainBackend
|
||||
return PlainBackend()
|
||||
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
@@ -0,0 +1,18 @@
|
||||
"""Kitty terminal backend — native splits via kitty @ remote control."""
|
||||
from __future__ import annotations
|
||||
import subprocess
|
||||
from typing import Union
|
||||
from kitty_workbench.backends import Backend
|
||||
|
||||
class KittyBackend(Backend):
|
||||
name = "kitty"
|
||||
def launch_pane(self, command: list[str], title: str) -> int:
|
||||
result = subprocess.run(
|
||||
["kitty", "@", "launch", "--location=vsplit", "--title", title] + command,
|
||||
capture_output=True, text=True,
|
||||
)
|
||||
return int(result.stdout.strip())
|
||||
def close_pane(self, pane_id: Union[int, str]) -> None:
|
||||
subprocess.run(["kitty", "@", "close-window", f"--match=id:{pane_id}"], capture_output=True)
|
||||
def image_protocol(self) -> str:
|
||||
return "kitty"
|
||||
@@ -0,0 +1,38 @@
|
||||
"""Plain terminal backend — opens a new terminal window."""
|
||||
from __future__ import annotations
|
||||
import os
|
||||
import platform
|
||||
import shutil
|
||||
import subprocess
|
||||
from typing import Union
|
||||
from kitty_workbench.backends import Backend
|
||||
|
||||
class PlainBackend(Backend):
|
||||
name = "plain"
|
||||
def launch_pane(self, command: list[str], title: str) -> str:
|
||||
terminal = os.environ.get("TERMINAL")
|
||||
if terminal and shutil.which(terminal):
|
||||
proc = subprocess.Popen([terminal, "-e"] + command)
|
||||
return str(proc.pid)
|
||||
system = platform.system()
|
||||
if system == "Darwin":
|
||||
proc = subprocess.Popen(["open", "-a", "Terminal.app", command[0], "--args"] + command[1:])
|
||||
return str(proc.pid)
|
||||
for term_cmd in ["x-terminal-emulator", "gnome-terminal", "konsole", "xterm"]:
|
||||
if shutil.which(term_cmd):
|
||||
if term_cmd == "gnome-terminal":
|
||||
proc = subprocess.Popen([term_cmd, "--", *command])
|
||||
else:
|
||||
proc = subprocess.Popen([term_cmd, "-e", *command])
|
||||
return str(proc.pid)
|
||||
raise RuntimeError(f"Could not find a terminal emulator. Please run manually:\n {' '.join(command)}")
|
||||
def close_pane(self, pane_id: Union[int, str]) -> None:
|
||||
pass
|
||||
def image_protocol(self) -> str:
|
||||
term = os.environ.get("TERM_PROGRAM", "")
|
||||
if term in {"iTerm.app", "iTerm2", "WezTerm", "foot", "contour", "xterm", "mlterm", "mintty"}:
|
||||
return "sixel"
|
||||
vte = os.environ.get("VTE_VERSION", "")
|
||||
if vte and int(vte) >= 7600:
|
||||
return "sixel"
|
||||
return "none"
|
||||
@@ -0,0 +1,27 @@
|
||||
"""Tmux backend — pane splits via tmux split-window."""
|
||||
from __future__ import annotations
|
||||
import os
|
||||
import subprocess
|
||||
from typing import Union
|
||||
from kitty_workbench.backends import Backend
|
||||
|
||||
SIXEL_TERMINALS = {"iTerm.app", "iTerm2", "WezTerm", "foot", "contour", "xterm", "mlterm", "mintty"}
|
||||
|
||||
class TmuxBackend(Backend):
|
||||
name = "tmux"
|
||||
def launch_pane(self, command: list[str], title: str) -> str:
|
||||
result = subprocess.run(
|
||||
["tmux", "split-window", "-h", "-d", "-P", "-F", "#{pane_id}"] + command,
|
||||
capture_output=True, text=True,
|
||||
)
|
||||
return result.stdout.strip()
|
||||
def close_pane(self, pane_id: Union[int, str]) -> None:
|
||||
subprocess.run(["tmux", "kill-pane", "-t", str(pane_id)], capture_output=True)
|
||||
def image_protocol(self) -> str:
|
||||
term = os.environ.get("TERM_PROGRAM", "")
|
||||
if term in SIXEL_TERMINALS:
|
||||
return "sixel"
|
||||
vte = os.environ.get("VTE_VERSION", "")
|
||||
if vte and int(vte) >= 7600:
|
||||
return "sixel"
|
||||
return "none"
|
||||
@@ -0,0 +1,61 @@
|
||||
import os
|
||||
from unittest.mock import patch, MagicMock
|
||||
from kitty_workbench.backends import detect_backend, Backend
|
||||
from kitty_workbench.backends.kitty import KittyBackend
|
||||
from kitty_workbench.backends.tmux import TmuxBackend
|
||||
from kitty_workbench.backends.plain import PlainBackend
|
||||
|
||||
|
||||
def test_detect_kitty():
|
||||
with patch.dict(os.environ, {"KITTY_PID": "12345"}, clear=False):
|
||||
backend = detect_backend()
|
||||
assert isinstance(backend, KittyBackend)
|
||||
|
||||
def test_detect_tmux():
|
||||
env = os.environ.copy()
|
||||
env.pop("KITTY_PID", None)
|
||||
env["TMUX"] = "/tmp/tmux-1000/default,12345,0"
|
||||
with patch.dict(os.environ, env, clear=True):
|
||||
backend = detect_backend()
|
||||
assert isinstance(backend, TmuxBackend)
|
||||
|
||||
def test_detect_plain():
|
||||
env = os.environ.copy()
|
||||
env.pop("KITTY_PID", None)
|
||||
env.pop("TMUX", None)
|
||||
with patch.dict(os.environ, env, clear=True):
|
||||
backend = detect_backend()
|
||||
assert isinstance(backend, PlainBackend)
|
||||
|
||||
def test_kitty_image_protocol():
|
||||
b = KittyBackend()
|
||||
assert b.image_protocol() == "kitty"
|
||||
|
||||
def test_tmux_image_protocol():
|
||||
b = TmuxBackend()
|
||||
with patch.dict(os.environ, {"TERM_PROGRAM": "unknown"}, clear=False):
|
||||
assert b.image_protocol() in ("sixel", "none")
|
||||
|
||||
def test_plain_image_protocol():
|
||||
b = PlainBackend()
|
||||
assert b.image_protocol() in ("sixel", "none")
|
||||
|
||||
def test_kitty_launch_pane():
|
||||
b = KittyBackend()
|
||||
with patch("kitty_workbench.backends.kitty.subprocess") as mock_sp:
|
||||
mock_sp.run.return_value = MagicMock(stdout="42", returncode=0)
|
||||
pane_id = b.launch_pane(["kitty-workbench", "tui", "test"], "Test")
|
||||
assert pane_id == 42
|
||||
call_args = mock_sp.run.call_args[0][0]
|
||||
assert "kitty" in call_args[0]
|
||||
assert "--location=vsplit" in call_args
|
||||
|
||||
def test_tmux_launch_pane():
|
||||
b = TmuxBackend()
|
||||
with patch("kitty_workbench.backends.tmux.subprocess") as mock_sp:
|
||||
mock_sp.run.return_value = MagicMock(stdout="%5", returncode=0)
|
||||
pane_id = b.launch_pane(["kitty-workbench", "tui", "test"], "Test")
|
||||
assert pane_id == "%5"
|
||||
call_args = mock_sp.run.call_args[0][0]
|
||||
assert "tmux" in call_args[0]
|
||||
assert "split-window" in call_args
|
||||
Reference in New Issue
Block a user