STT-Assistant/src/stt_daemon/ctl.py

"""One-shot control client for hotkeys (`stt-ctl`).

Connects to the assistant control socket, sends a single command, prints the
replies it receives, and exits. Intended to be bound to global hotkeys.
"""

import argparse
import json
import socket
import sys

from .assistant.control import default_control_path
from .config import load_config

COMMAND_MAP = {
    "command-start": "command_start",
    "command-execute": "command_execute",
    "command-cancel": "command_cancel",
    "command-edit": "command_edit",
    "chat-toggle": "chat_toggle",
    "clear": "clear",
    "status": "status",
    "speaker-list": "speaker_list",
    "speaker-name": "speaker_name",
}

READ_TIMEOUT_S = 1.0
READ_CHUNK = 65536


def build_message(command: str, text: str | None = None, rest: list[str] | None = None) -> dict:
    """Map a CLI verb to a control-socket message."""
    if command not in COMMAND_MAP:
        raise ValueError(f"unknown command: {command}")
    message = {"type": COMMAND_MAP[command]}
    if text is not None:
        message["text"] = text
    if command == "speaker-name":
        args = list(rest or [])
        message["speaker_id"] = args[0] if len(args) > 0 else ""
        message["name"] = args[1] if len(args) > 1 else ""
    return message


def main() -> int:
    """Entry point for `stt-ctl`."""
    parser = argparse.ArgumentParser(description="Send one command to the running assistant")
    parser.add_argument("command", choices=sorted(COMMAND_MAP))
    parser.add_argument("rest", nargs="*", help="Extra args (speaker-name: <id> <name>)")
    parser.add_argument("--text", default=None, help="Text for command-edit")
    parser.add_argument("--config", default=None, help="Path to config.toml")
    parser.add_argument("--socket", default=None, help="Override the control socket path")
    args = parser.parse_args()

    path = args.socket
    if path is None:
        cfg = load_config(args.config)
        path = cfg.assistant.control_socket or default_control_path()

    try:
        message = build_message(args.command, args.text, args.rest)
    except ValueError as e:
        print(f"error: {e}", file=sys.stderr)
        return 2

    sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
    try:
        sock.connect(path)
    except OSError as e:
        print(f"error: cannot connect to {path}: {e}", file=sys.stderr)
        return 1

    sock.sendall((json.dumps(message) + "\n").encode("utf-8"))
    sock.settimeout(READ_TIMEOUT_S)
    buffer = bytearray()
    try:
        while True:
            try:
                chunk = sock.recv(READ_CHUNK)
            except socket.timeout:
                break
            if not chunk:
                break
            buffer.extend(chunk)
            while True:
                newline = buffer.find(b"\n")
                if newline == -1:
                    break
                line = bytes(buffer[:newline])
                del buffer[: newline + 1]
                if not line.strip():
                    continue
                try:
                    reply = json.loads(line.decode("utf-8"))
                except (UnicodeDecodeError, json.JSONDecodeError):
                    continue
                print(json.dumps(reply, ensure_ascii=False))
    finally:
        try:
            sock.shutdown(socket.SHUT_RDWR)
        except OSError:
            pass
        sock.close()
    return 0


if __name__ == "__main__":
    sys.exit(main())