Basic tmux instrumentation through libtmux
This commit is contained in:
parent
65bc6d3e8e
commit
f5987d4048
3 changed files with 135 additions and 0 deletions
74
patch2vimedit/main.py
Normal file
74
patch2vimedit/main.py
Normal file
|
@ -0,0 +1,74 @@
|
||||||
|
import argparse
|
||||||
|
import pathlib
|
||||||
|
import os
|
||||||
|
import sys
|
||||||
|
import time
|
||||||
|
|
||||||
|
import patch
|
||||||
|
from tmux import TmuxSession
|
||||||
|
|
||||||
|
|
||||||
|
def parse_args():
|
||||||
|
""" Parse command-line arguments """
|
||||||
|
parser = argparse.ArgumentParser(prog="patch2vimedit")
|
||||||
|
parser.add_argument(
|
||||||
|
"-C",
|
||||||
|
"--directory",
|
||||||
|
metavar="DIR",
|
||||||
|
help="Change to DIR before performing any operations.",
|
||||||
|
)
|
||||||
|
parser.add_argument(
|
||||||
|
"patch",
|
||||||
|
help=(
|
||||||
|
"The patch file to be applied. If '-' is supplied, reads from STDIN "
|
||||||
|
"instead."
|
||||||
|
),
|
||||||
|
)
|
||||||
|
return parser.parse_args()
|
||||||
|
|
||||||
|
|
||||||
|
def get_patch(patch_path):
|
||||||
|
if patch_path == "-":
|
||||||
|
return patch.PatchSet(sys.stdin)
|
||||||
|
with open(patch_path, "rb") as stream:
|
||||||
|
return patch.PatchSet(stream)
|
||||||
|
|
||||||
|
|
||||||
|
def main():
|
||||||
|
""" Entry-point function """
|
||||||
|
args = parse_args()
|
||||||
|
|
||||||
|
if args.directory:
|
||||||
|
chdir_to = pathlib.Path(args.directory)
|
||||||
|
if not chdir_to.is_dir():
|
||||||
|
print(
|
||||||
|
(
|
||||||
|
"The required working directory '{}' does not exist, or is not a "
|
||||||
|
"directory."
|
||||||
|
).format(str(chdir_to)),
|
||||||
|
file=sys.stderr,
|
||||||
|
)
|
||||||
|
sys.exit(1)
|
||||||
|
os.chdir(chdir_to)
|
||||||
|
|
||||||
|
patchset = get_patch(args.patch)
|
||||||
|
|
||||||
|
tmux_session = TmuxSession.create_detached()
|
||||||
|
tmux_fg_process = tmux_session.process_attach(read_only=True)
|
||||||
|
|
||||||
|
time.sleep(0.2)
|
||||||
|
tmux_session.send_keys('echo "Hello, world!"', "enter")
|
||||||
|
time.sleep(1)
|
||||||
|
tmux_session.send_keys('echo "Bye, world!"', "enter")
|
||||||
|
time.sleep(1)
|
||||||
|
tmux_session.send_keys("exit", "enter")
|
||||||
|
|
||||||
|
tmux_fg_process.join(1)
|
||||||
|
if tmux_fg_process.exitcode is None: # Did not terminate -- kill it
|
||||||
|
tmux_fg_process.kill()
|
||||||
|
if tmux_session.session_exists():
|
||||||
|
tmux_session.session.kill_session()
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
main()
|
59
patch2vimedit/tmux.py
Normal file
59
patch2vimedit/tmux.py
Normal file
|
@ -0,0 +1,59 @@
|
||||||
|
import subprocess
|
||||||
|
import multiprocessing
|
||||||
|
import sys
|
||||||
|
import random
|
||||||
|
import libtmux
|
||||||
|
|
||||||
|
|
||||||
|
class TmuxSession:
|
||||||
|
""" A tmux session """
|
||||||
|
|
||||||
|
tmux_server = libtmux.Server()
|
||||||
|
|
||||||
|
def __init__(self, session):
|
||||||
|
self.session = session
|
||||||
|
self.session_id = session.id
|
||||||
|
|
||||||
|
@classmethod
|
||||||
|
def create_detached(cls, name=None):
|
||||||
|
""" Create a new detached tmux session, returning the TmuxSession """
|
||||||
|
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`). """
|
||||||
|
# libtmux is not expressive enough to avoid `subprocess` here...
|
||||||
|
subprocess.run(["tmux", "attach", "-r", "-t", self.session.id], check=True)
|
||||||
|
|
||||||
|
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:
|
||||||
|
self.session.attached_pane.send_keys(
|
||||||
|
arg, suppress_history=False, enter=False
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
# subprocess.run(
|
||||||
|
# ["tmux", "send-keys", "-t", self.session.id] + list(args), check=True
|
||||||
|
# )
|
2
requirements.txt
Normal file
2
requirements.txt
Normal file
|
@ -0,0 +1,2 @@
|
||||||
|
patch==1.*
|
||||||
|
libtmux==0.8.*
|
Loading…
Reference in a new issue