tmux-amq/src/tamq/control.py

46 lines
1.3 KiB
Python
Raw Normal View History

from __future__ import annotations
import subprocess
from dataclasses import dataclass
class ControlModeError(RuntimeError):
pass
@dataclass
class ControlModeClient:
session: str
process: subprocess.Popen[str] | None = None
def start(self) -> None:
if self.process is not None:
return
self.process = subprocess.Popen(
["tmux", "-C", "attach-session", "-t", self.session],
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
bufsize=1,
)
def command(self, value: str) -> None:
if self.process is None or self.process.stdin is None:
raise ControlModeError("control-mode client is not started")
self.process.stdin.write(value.rstrip("\n") + "\n")
self.process.stdin.flush()
def inject(self, window: str, text: str) -> None:
# send-keys -l preserves punctuation and whitespace in the message.
self.command(f"send-keys -t {window} -l -- {shell_quote(text)}")
def close(self) -> None:
if self.process is not None:
self.process.terminate()
self.process = None
def shell_quote(value: str) -> str:
return "'" + value.replace("'", "'\\''") + "'"