319 lines
12 KiB
Python
319 lines
12 KiB
Python
import subprocess
|
|
import threading
|
|
import time
|
|
from typing import List
|
|
|
|
from fabric.utils import exec_shell_command_async
|
|
|
|
import config.data as data
|
|
import utils.icons as icons
|
|
from modules.launcher.plugin_base import PluginBase
|
|
from modules.launcher.result import Result
|
|
|
|
|
|
class TmuxPlugin(PluginBase):
|
|
"""
|
|
Plugin for managing tmux sessions through the launcher.
|
|
"""
|
|
|
|
def __init__(self):
|
|
super().__init__()
|
|
self.display_name = "Tmux Manager"
|
|
self.description = "Manage tmux sessions - create, attach, rename, and kill"
|
|
|
|
# Cache for sessions to avoid repeated subprocess calls
|
|
self._sessions_cache = []
|
|
self._last_cache_update = 0
|
|
self._cache_ttl = 2 # Cache sessions for 2 seconds
|
|
|
|
# Threading for auto-refresh
|
|
self.refresh_thread = None
|
|
self.stop_refresh = threading.Event()
|
|
|
|
def initialize(self):
|
|
"""Initialize the tmux plugin."""
|
|
self.set_triggers(["tmux"])
|
|
self._start_refresh_thread()
|
|
|
|
def cleanup(self):
|
|
"""Cleanup the tmux plugin."""
|
|
self.stop_refresh.set()
|
|
if self.refresh_thread:
|
|
self.refresh_thread.join(timeout=1)
|
|
self._sessions_cache.clear()
|
|
|
|
def _start_refresh_thread(self):
|
|
"""Start background thread to refresh session cache."""
|
|
if not self.refresh_thread or not self.refresh_thread.is_alive():
|
|
self.refresh_thread = threading.Thread(
|
|
target=self._refresh_sessions_background, daemon=True
|
|
)
|
|
self.refresh_thread.start()
|
|
|
|
def _refresh_sessions_background(self):
|
|
"""Background thread to refresh sessions cache."""
|
|
while not self.stop_refresh.is_set():
|
|
try:
|
|
current_time = time.time()
|
|
if current_time - self._last_cache_update > self._cache_ttl:
|
|
self._sessions_cache = self._get_tmux_sessions()
|
|
self._last_cache_update = current_time
|
|
|
|
# Sleep for a short interval
|
|
self.stop_refresh.wait(1)
|
|
except Exception as e:
|
|
print(f"TmuxPlugin: Error in refresh thread: {e}")
|
|
self.stop_refresh.wait(5) # Wait longer on error
|
|
|
|
def _get_tmux_sessions(self):
|
|
"""Get list of tmux sessions."""
|
|
try:
|
|
result = subprocess.run(
|
|
["tmux", "list-sessions", "-F", "#{session_name}"],
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=5,
|
|
)
|
|
if result.returncode == 0:
|
|
return [
|
|
s.strip() for s in result.stdout.strip().split("\n") if s.strip()
|
|
]
|
|
return []
|
|
except (
|
|
subprocess.TimeoutExpired,
|
|
subprocess.CalledProcessError,
|
|
FileNotFoundError,
|
|
) as e:
|
|
print(f"TmuxPlugin: Error getting tmux sessions: {e}")
|
|
return []
|
|
|
|
def query(self, query_string: str) -> List[Result]:
|
|
"""Process tmux queries."""
|
|
query = query_string.strip().lower()
|
|
results = []
|
|
|
|
# Get current sessions (use cache if recent)
|
|
current_time = time.time()
|
|
if current_time - self._last_cache_update > self._cache_ttl:
|
|
self._sessions_cache = self._get_tmux_sessions()
|
|
self._last_cache_update = current_time
|
|
|
|
sessions = self._sessions_cache
|
|
|
|
# Handle specific commands
|
|
if query.startswith("new ") or query.startswith("create "):
|
|
session_name = query.split(" ", 1)[1].strip() if " " in query else ""
|
|
results.append(self._create_new_session_result(session_name))
|
|
|
|
elif query.startswith("kill ") or query.startswith("delete "):
|
|
session_name = query.split(" ", 1)[1].strip() if " " in query else ""
|
|
if session_name:
|
|
matching_sessions = [
|
|
s for s in sessions if session_name.lower() in s.lower()
|
|
]
|
|
for session in matching_sessions:
|
|
results.append(self._create_kill_session_result(session))
|
|
|
|
elif query.startswith("rename "):
|
|
parts = query.split(" ", 2)
|
|
if len(parts) >= 3:
|
|
old_name, new_name = parts[1], parts[2]
|
|
if old_name in sessions:
|
|
results.append(
|
|
self._create_rename_session_result(old_name, new_name)
|
|
)
|
|
|
|
else:
|
|
# Show existing sessions for attachment
|
|
if sessions:
|
|
# Filter sessions based on query
|
|
if query:
|
|
filtered_sessions = [s for s in sessions if query in s.lower()]
|
|
else:
|
|
filtered_sessions = sessions
|
|
|
|
for session in filtered_sessions:
|
|
results.append(self._create_attach_session_result(session))
|
|
|
|
# Always show option to create new session
|
|
if not query or "new" in query or "create" in query:
|
|
results.append(
|
|
self._create_new_session_result(
|
|
query
|
|
if query and not any(cmd in query for cmd in ["new", "create"])
|
|
else ""
|
|
)
|
|
)
|
|
|
|
return results
|
|
|
|
def _create_attach_session_result(self, session_name: str) -> Result:
|
|
"""Create result for attaching to a session."""
|
|
return Result(
|
|
title=f"Attach to '{session_name}'",
|
|
subtitle=f"Connect to tmux session: {session_name}",
|
|
icon_markup=icons.terminal,
|
|
action=lambda: self._attach_to_session(session_name),
|
|
relevance=0.9,
|
|
data={"type": "attach", "session": session_name},
|
|
)
|
|
|
|
def _create_new_session_result(self, session_name: str = "") -> Result:
|
|
"""Create result for creating a new session."""
|
|
display_name = session_name if session_name else "new session"
|
|
return Result(
|
|
title=f"Create '{display_name}'",
|
|
subtitle=f"Create new tmux session{f': {session_name}' if session_name else ''}",
|
|
icon_markup=icons.plus,
|
|
action=lambda: self._create_session(session_name),
|
|
relevance=0.8,
|
|
data={"type": "create", "session": session_name},
|
|
)
|
|
|
|
def _create_kill_session_result(self, session_name: str) -> Result:
|
|
"""Create result for killing a session."""
|
|
return Result(
|
|
title=f"Kill '{session_name}'",
|
|
subtitle=f"Terminate tmux session: {session_name}",
|
|
icon_markup=icons.trash,
|
|
action=lambda: self._kill_session(session_name),
|
|
relevance=0.7,
|
|
data={"type": "kill", "session": session_name},
|
|
)
|
|
|
|
def _create_rename_session_result(self, old_name: str, new_name: str) -> Result:
|
|
"""Create result for renaming a session."""
|
|
return Result(
|
|
title=f"Rename '{old_name}' to '{new_name}'",
|
|
subtitle=f"Rename tmux session from {old_name} to {new_name}",
|
|
icon_markup=icons.config,
|
|
action=lambda: self._rename_session(old_name, new_name),
|
|
relevance=0.6,
|
|
data={"type": "rename", "old_session": old_name, "new_session": new_name},
|
|
)
|
|
|
|
def _attach_to_session(self, session_name: str):
|
|
"""Attach to an existing tmux session."""
|
|
try:
|
|
terminal_cmd = self._get_terminal_command(
|
|
f"tmux attach-session -t '{session_name}'"
|
|
)
|
|
exec_shell_command_async(terminal_cmd)
|
|
print(f"TmuxPlugin: Attaching to session '{session_name}'")
|
|
except Exception as e:
|
|
print(f"TmuxPlugin: Error attaching to session '{session_name}': {e}")
|
|
|
|
def _create_session(self, session_name: str = ""):
|
|
"""Create a new tmux session."""
|
|
try:
|
|
if not session_name:
|
|
# Generate a default name
|
|
existing_sessions = self._get_tmux_sessions()
|
|
counter = 0
|
|
while str(counter) in existing_sessions:
|
|
counter += 1
|
|
session_name = str(counter)
|
|
|
|
# Clean the session name
|
|
clean_name = session_name.strip().replace(" ", "_")
|
|
|
|
# Create session
|
|
subprocess.run(
|
|
["tmux", "new-session", "-d", "-s", clean_name], check=True, timeout=10
|
|
)
|
|
|
|
# Launch terminal and attach
|
|
terminal_cmd = self._get_terminal_command(
|
|
f"tmux attach-session -t '{clean_name}'"
|
|
)
|
|
exec_shell_command_async(terminal_cmd)
|
|
|
|
# Refresh cache
|
|
self._sessions_cache = self._get_tmux_sessions()
|
|
self._last_cache_update = time.time()
|
|
|
|
print(f"TmuxPlugin: Created and attached to session '{clean_name}'")
|
|
except Exception as e:
|
|
print(f"TmuxPlugin: Error creating session '{session_name}': {e}")
|
|
|
|
def _kill_session(self, session_name: str):
|
|
"""Kill a tmux session."""
|
|
try:
|
|
subprocess.run(
|
|
["tmux", "kill-session", "-t", session_name], check=True, timeout=10
|
|
)
|
|
|
|
# Refresh cache
|
|
self._sessions_cache = self._get_tmux_sessions()
|
|
self._last_cache_update = time.time()
|
|
|
|
print(f"TmuxPlugin: Killed session '{session_name}'")
|
|
except Exception as e:
|
|
print(f"TmuxPlugin: Error killing session '{session_name}': {e}")
|
|
|
|
def _rename_session(self, old_name: str, new_name: str):
|
|
"""Rename a tmux session."""
|
|
try:
|
|
clean_name = new_name.strip().replace(" ", "_")
|
|
subprocess.run(
|
|
["tmux", "rename-session", "-t", old_name, clean_name],
|
|
check=True,
|
|
timeout=10,
|
|
)
|
|
|
|
# Refresh cache
|
|
self._sessions_cache = self._get_tmux_sessions()
|
|
self._last_cache_update = time.time()
|
|
|
|
print(f"TmuxPlugin: Renamed session '{old_name}' to '{clean_name}'")
|
|
except Exception as e:
|
|
print(
|
|
f"TmuxPlugin: Error renaming session '{old_name}' to '{new_name}': {e}"
|
|
)
|
|
|
|
def _get_terminal_command(self, cmd: str) -> str:
|
|
"""Get terminal command based on configured terminal or available terminals."""
|
|
# First try to use the configured terminal command
|
|
if hasattr(data, "TERMINAL_COMMAND") and data.TERMINAL_COMMAND:
|
|
parts = data.TERMINAL_COMMAND.split()
|
|
terminal = parts[0]
|
|
|
|
try:
|
|
# Check if the configured terminal is available
|
|
subprocess.run(
|
|
["which", terminal],
|
|
check=True,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
)
|
|
return f"{data.TERMINAL_COMMAND} {cmd}"
|
|
except subprocess.CalledProcessError:
|
|
# If configured terminal is not available, fall back to defaults
|
|
pass
|
|
|
|
# Fallback to checking available terminals
|
|
terminals = [
|
|
("kitty", f"kitty -e {cmd}"),
|
|
("alacritty", f"alacritty -e {cmd}"),
|
|
("foot", f"foot {cmd}"),
|
|
("gnome-terminal", f"gnome-terminal -- {cmd}"),
|
|
("konsole", f"konsole -e {cmd}"),
|
|
("xfce4-terminal", f"xfce4-terminal -e '{cmd}'"),
|
|
]
|
|
|
|
for term, term_cmd in terminals:
|
|
try:
|
|
# Check if terminal is available
|
|
subprocess.run(
|
|
["which", term],
|
|
check=True,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
)
|
|
return term_cmd
|
|
except subprocess.CalledProcessError:
|
|
continue
|
|
|
|
# Default fallback
|
|
return f"kitty -e {cmd}"
|