111 lines
3.6 KiB
Python
111 lines
3.6 KiB
Python
import multiprocessing
|
|
import random
|
|
import time
|
|
import logging
|
|
import libtmux
|
|
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class TmuxSession:
|
|
""" A tmux session """
|
|
|
|
tmux_server = None
|
|
tmux_socket = None
|
|
|
|
def __init__(self, session):
|
|
if self.tmux_server is None:
|
|
raise Exception("Server not initialized")
|
|
|
|
self.session = session
|
|
self.session_id = session.id
|
|
self.keyboard_speed = 0.00001
|
|
|
|
self.sendkey_log_buffer = []
|
|
|
|
@classmethod
|
|
def initialize_server(cls, socket_name=None, config_file=None):
|
|
""" Initialize the tmux server """
|
|
cls.tmux_socket = socket_name
|
|
cls.tmux_server = libtmux.Server(
|
|
socket_name=socket_name, config_file=config_file
|
|
)
|
|
|
|
@classmethod
|
|
def create_detached(cls, name=None):
|
|
""" Create a new detached tmux session, returning the TmuxSession """
|
|
if cls.tmux_server is None:
|
|
raise Exception("Server not initialized")
|
|
if not name:
|
|
name = "patch2vimsession-{:04x}".format(random.randint(0, 1 << 16 - 1))
|
|
session = cls.tmux_server.new_session(session_name=name, attach=False)
|
|
return cls(session)
|
|
|
|
def attach(self):
|
|
""" Attach the tmux session to the current terminal. Does NOT return until the
|
|
tmux session ends or is detached. """
|
|
self.session.attach_session()
|
|
|
|
def attach_ro(self):
|
|
""" Same as `attach`, but attaches a tmux session in a read-only fashion (see
|
|
`tmux attach -r`). """
|
|
self.tmux_server.cmd("attach", "-r", "-t", self.session.id)
|
|
|
|
def session_exists(self):
|
|
""" Checks that the session exists """
|
|
return self.tmux_server.has_session(self.session_id)
|
|
|
|
def process_attach(self, read_only=True):
|
|
""" Runs `self.attach` in a `multiprocessing.Process` and returns the process.
|
|
"""
|
|
target = self.attach_ro if read_only else self.attach
|
|
process = multiprocessing.Process(target=target)
|
|
process.start()
|
|
return process
|
|
|
|
def send_keys(self, *args):
|
|
""" Send keys to the tmux process. See `man tmux` and `tmux send-keys` for
|
|
documentation """
|
|
for arg in args:
|
|
if arg.endswith(";"):
|
|
# Weirdly, `tmux send-keys 'blah;'` doesn't send the semicolon; and so
|
|
# does `tmux send-keys ';'`. We must escape it with a backslash.
|
|
arg = arg[:-1] + r"\;"
|
|
|
|
if arg in ["escape", "enter"]:
|
|
logline = "".join(self.sendkey_log_buffer) + " " + arg
|
|
logger.debug(logline)
|
|
self.sendkey_log_buffer = []
|
|
else:
|
|
self.sendkey_log_buffer.append(arg)
|
|
|
|
self.session.attached_pane.send_keys(
|
|
arg, suppress_history=False, enter=False
|
|
)
|
|
|
|
def type_delay(self):
|
|
""" Introduce a small delay, to emulate a human typing """
|
|
delay = self.keyboard_speed + random.gauss(0, self.keyboard_speed * 0.1)
|
|
time.sleep(delay)
|
|
|
|
def type_keys(self, *args):
|
|
""" Same as `send_keys`, but emulates a human typing with pauses. This is
|
|
**unsafe** for special keys, although the common ones are preserved. """
|
|
|
|
if self.keyboard_speed == 0:
|
|
return self.send_keys(*args)
|
|
|
|
for arg in args:
|
|
if (
|
|
arg in ["enter", "escape"]
|
|
or arg.startswith("C-")
|
|
or arg.startswith("M-")
|
|
):
|
|
self.send_keys(arg)
|
|
self.type_delay()
|
|
else:
|
|
for char in arg:
|
|
self.send_keys(char)
|
|
self.type_delay()
|