tmux-amq/src/tamq/control.py

from __future__ import annotations

import os
import subprocess
from collections.abc import Sequence
from dataclasses import dataclass


class ControlModeError(RuntimeError):
    pass


@dataclass
class ControlModeClient:
    session: str
    tmux_command: Sequence[str] | None = None
    process: subprocess.Popen[str] | None = None

    def start(self) -> None:
        if self.process is not None:
            return
        socket_name = os.environ.get("TAMQ_TMUX_SOCKET")
        command = tuple(
            self.tmux_command
            or (["tmux", "-L", socket_name] if socket_name else ["tmux"])
        )
        self.process = subprocess.Popen(
            [*command, "-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:
            process = self.process
            process.terminate()
            try:
                process.wait(timeout=1)
            except subprocess.TimeoutExpired:
                process.kill()
                process.wait(timeout=1)
            self.process = None


def shell_quote(value: str) -> str:
    return "'" + value.replace("'", "'\\''") + "'"