clicky_linux_cli.py

#!/usr/bin/env python3
"""Linux CLI companion for Clicky using the existing Cloudflare Worker proxy."""

from __future__ import annotations

import argparse
import base64
import importlib
import json
import os
import re
import shutil
import subprocess
import sys
import tempfile
from dataclasses import dataclass
from typing import List, Optional, Sequence, Tuple

import requests

# Persistent HTTP session for connection pooling across all Worker requests.
# Reuses TCP + TLS connections, cutting ~50-150ms per request.
_cli_http_session = requests.Session()


POINT_TAG_PATTERN = re.compile(
    r"\[POINT:(?:none|(\d+)\s*,\s*(\d+)(?::([^\]:\s][^\]:]*?))?(?::screen(\d+))?)\]\s*$"
)
DEFAULT_MODEL = "claude-sonnet-4-6"
MAX_HISTORY_EXCHANGES = 10

DEFAULT_SYSTEM_PROMPT = (
    "You are Clicky, a clear and concise desktop companion helping a Linux user in real time. "
    "Prioritize practical steps they can execute immediately. Keep responses short, conversational, "
    "and easy to follow out loud. If a screenshot does not provide enough context, ask a focused "
    "clarifying question. End with exactly one [POINT:...] tag if you can identify a target; otherwise "
    "use [POINT:none]."
)


@dataclass
class CapturedScreen:
    screen_index: int
    width: int
    height: int
    png_bytes: bytes
    image_media_type: str = "image/png"


@dataclass
class PointingMetadata:
    x: Optional[int]
    y: Optional[int]
    label: Optional[str]
    screen_number: Optional[int]
    indicates_no_target: bool = False


def parse_arguments() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Run Clicky from Linux terminal with screenshot + Claude + optional TTS."
    )
    parser.add_argument(
        "--worker-url",
        default=os.getenv("CLICKY_WORKER_URL", "").strip(),
        help="Cloudflare Worker base URL, e.g. https://your-worker.workers.dev",
    )
    parser.add_argument(
        "--model",
        default=os.getenv("CLICKY_MODEL", DEFAULT_MODEL).strip() or DEFAULT_MODEL,
        help=f"Claude model name (default: {DEFAULT_MODEL})",
    )
    parser.add_argument(
        "--system-prompt",
        default=os.getenv("CLICKY_SYSTEM_PROMPT", "").strip() or DEFAULT_SYSTEM_PROMPT,
        help="System prompt sent to Claude.",
    )
    parser.add_argument(
        "--no-screenshot",
        action="store_true",
        help="Disable screenshot capture and send text-only requests.",
    )
    parser.add_argument(
        "--no-tts",
        action="store_true",
        help="Disable ElevenLabs TTS playback.",
    )
    return parser.parse_args()


def validate_worker_url(worker_url: str) -> str:
    sanitized_worker_url = worker_url.strip().rstrip("/")
    if not sanitized_worker_url:
        raise ValueError(
            "Missing Worker URL. Pass --worker-url or set CLICKY_WORKER_URL."
        )
    if not (sanitized_worker_url.startswith("http://") or sanitized_worker_url.startswith("https://")):
        raise ValueError("Worker URL must start with http:// or https://")
    return sanitized_worker_url


def capture_screenshots() -> List[CapturedScreen]:
    try:
        mss_module = importlib.import_module("mss")
        mss_tools_module = importlib.import_module("mss.tools")
    except ModuleNotFoundError:
        print(
            "[warning] python package 'mss' is not installed. Running without screenshots.",
            file=sys.stderr,
        )
        return []

    # Try to use Pillow for JPEG conversion (smaller payloads, faster upload).
    pillow_image = None
    try:
        pillow_image = importlib.import_module("PIL.Image")
    except Exception:
        pass

    captured_screens: List[CapturedScreen] = []
    with mss_module.mss() as screenshot_session:
        for monitor_index, monitor in enumerate(screenshot_session.monitors[1:], start=1):
            monitor_capture = screenshot_session.grab(monitor)

            image_bytes: bytes
            media_type: str
            if pillow_image is not None:
                try:
                    import io
                    pil_img = pillow_image.frombytes(
                        "RGB",
                        (monitor_capture.width, monitor_capture.height),
                        monitor_capture.rgb,
                    )
                    jpeg_buffer = io.BytesIO()
                    pil_img.save(jpeg_buffer, format="JPEG", quality=80)
                    image_bytes = jpeg_buffer.getvalue()
                    media_type = "image/jpeg"
                except Exception:
                    image_bytes = mss_tools_module.to_png(
                        monitor_capture.rgb, monitor_capture.size
                    )
                    media_type = "image/png"
            else:
                image_bytes = mss_tools_module.to_png(
                    monitor_capture.rgb, monitor_capture.size
                )
                media_type = "image/png"

            captured_screens.append(
                CapturedScreen(
                    screen_index=monitor_index,
                    width=monitor_capture.width,
                    height=monitor_capture.height,
                    png_bytes=image_bytes,
                    image_media_type=media_type,
                )
            )
    return captured_screens


def build_messages_payload(
    conversation_history: Sequence[Tuple[str, str]],
    user_prompt: str,
    captured_screens: Sequence[CapturedScreen],
) -> List[dict]:
    messages: List[dict] = []

    for previous_user_prompt, previous_assistant_response in conversation_history:
        messages.append({"role": "user", "content": previous_user_prompt})
        messages.append({"role": "assistant", "content": previous_assistant_response})

    current_message_content: List[dict] = []
    for captured_screen in captured_screens:
        image_base64_data = base64.b64encode(captured_screen.png_bytes).decode("ascii")
        current_message_content.append(
            {
                "type": "image",
                "source": {
                    "type": "base64",
                    "media_type": captured_screen.image_media_type,
                    "data": image_base64_data,
                },
            }
        )
        current_message_content.append(
            {
                "type": "text",
                "text": (
                    f"Screen {captured_screen.screen_index} "
                    f"(image{captured_screen.screen_index}: "
                    f"{captured_screen.width}x{captured_screen.height} pixels)"
                ),
            }
        )

    current_message_content.append({"type": "text", "text": user_prompt})
    messages.append({"role": "user", "content": current_message_content})
    return messages


def stream_claude_response(
    worker_base_url: str,
    model: str,
    system_prompt: str,
    messages: Sequence[dict],
) -> str:
    chat_endpoint_url = f"{worker_base_url}/chat"
    request_body = {
        "model": model,
        "max_tokens": 1024,
        "stream": True,
        "system": system_prompt,
        "messages": list(messages),
    }

    print("Clicky> ", end="", flush=True)
    streamed_response_chunks: List[str] = []

    with _cli_http_session.post(
        chat_endpoint_url,
        json=request_body,
        stream=True,
        timeout=300,
    ) as streaming_response:
        if not streaming_response.ok:
            error_body = streaming_response.text
            raise RuntimeError(
                f"/chat failed with status {streaming_response.status_code}: {error_body}"
            )

        for streamed_line in streaming_response.iter_lines(decode_unicode=True):
            if not streamed_line:
                continue
            if not streamed_line.startswith("data: "):
                continue

            event_json_text = streamed_line[6:]
            if event_json_text == "[DONE]":
                break

            try:
                event_payload = json.loads(event_json_text)
            except json.JSONDecodeError:
                continue

            if event_payload.get("type") != "content_block_delta":
                continue

            delta_payload = event_payload.get("delta") or {}
            if delta_payload.get("type") != "text_delta":
                continue

            text_chunk = delta_payload.get("text", "")
            if not text_chunk:
                continue

            streamed_response_chunks.append(text_chunk)
            print(text_chunk, end="", flush=True)

    print()
    return "".join(streamed_response_chunks)


def parse_pointing_metadata(full_response_text: str) -> Tuple[str, Optional[PointingMetadata]]:
    point_match = POINT_TAG_PATTERN.search(full_response_text)
    if not point_match:
        return full_response_text, None

    x_value, y_value, label_value, screen_number_value = point_match.groups()
    spoken_response_text = full_response_text[: point_match.start()].rstrip()

    if x_value is None or y_value is None:
        return spoken_response_text, PointingMetadata(
            x=None,
            y=None,
            label=None,
            screen_number=None,
            indicates_no_target=True,
        )

    trimmed_label = label_value.strip() if label_value else None
    return spoken_response_text, PointingMetadata(
        x=int(x_value),
        y=int(y_value),
        label=trimmed_label,
        screen_number=int(screen_number_value) if screen_number_value else None,
    )


def request_tts_audio(worker_base_url: str, text_to_speak: str) -> bytes:
    tts_endpoint_url = f"{worker_base_url}/tts"
    request_body = {
        "text": text_to_speak,
        "model_id": "eleven_flash_v2_5",
        "voice_settings": {
            "stability": 0.5,
            "similarity_boost": 0.75,
        },
    }

    tts_response = _cli_http_session.post(
        tts_endpoint_url,
        json=request_body,
        headers={"Accept": "audio/mpeg"},
        timeout=120,
    )

    if not tts_response.ok:
        raise RuntimeError(
            f"/tts failed with status {tts_response.status_code}: {tts_response.text}"
        )

    return tts_response.content


def play_mp3_audio_bytes(tts_audio_bytes: bytes) -> None:
    with tempfile.NamedTemporaryFile(prefix="clicky-linux-", suffix=".mp3", delete=False) as temporary_file:
        temporary_file.write(tts_audio_bytes)
        temp_audio_file_path = temporary_file.name

    available_player_commands = [
        ["ffplay", "-nodisp", "-autoexit", "-loglevel", "error",
         "-fflags", "nobuffer", "-flags", "low_delay",
         "-analyzeduration", "0", "-probesize", "32",
         temp_audio_file_path],
        ["mpg123", "-q", temp_audio_file_path],
        ["mpv", "--no-video", "--really-quiet",
         "--demuxer-readahead-secs=0", "--audio-buffer=0",
         temp_audio_file_path],
        ["cvlc", "--quiet", "--play-and-exit", temp_audio_file_path],
    ]

    for player_command in available_player_commands:
        if shutil.which(player_command[0]) is None:
            continue

        subprocess.run(
            player_command,
            check=False,
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
        )
        os.unlink(temp_audio_file_path)
        return

    if shutil.which("xdg-open") is not None:
        subprocess.Popen(
            ["xdg-open", temp_audio_file_path],
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
        )
        print(
            f"[info] Opened audio using system default app: {temp_audio_file_path}",
            file=sys.stderr,
        )
        return

    fallback_audio_file_path = os.path.abspath("clicky-last-response.mp3")
    os.replace(temp_audio_file_path, fallback_audio_file_path)
    print(
        f"[warning] No supported audio player found. Saved TTS audio to {fallback_audio_file_path}",
        file=sys.stderr,
    )


def print_startup_help() -> None:
    print("Linux Clicky CLI ready.")
    print("Commands: /help, /clear, /exit")
    print("Type your prompt and press Enter.")


def handle_builtin_command(command_text: str, conversation_history: List[Tuple[str, str]]) -> bool:
    if command_text == "/help":
        print("/help  Show available commands")
        print("/clear Clear local conversation history")
        print("/exit  Exit CLI")
        return True

    if command_text == "/clear":
        conversation_history.clear()
        print("[info] Conversation history cleared.")
        return True

    return False


def run_cli() -> int:
    parsed_arguments = parse_arguments()

    try:
        worker_base_url = validate_worker_url(parsed_arguments.worker_url)
    except ValueError as validation_error:
        print(f"error: {validation_error}", file=sys.stderr)
        return 2

    conversation_history: List[Tuple[str, str]] = []
    print_startup_help()

    while True:
        try:
            user_prompt = input("You> ").strip()
        except (EOFError, KeyboardInterrupt):
            print("\nExiting.")
            return 0

        if not user_prompt:
            continue

        if user_prompt in {"/exit", "/quit"}:
            print("Exiting.")
            return 0

        if user_prompt.startswith("/"):
            command_handled = handle_builtin_command(user_prompt, conversation_history)
            if command_handled:
                continue
            print(f"[warning] Unknown command: {user_prompt}")
            continue

        if parsed_arguments.no_screenshot:
            captured_screens: List[CapturedScreen] = []
        else:
            captured_screens = capture_screenshots()
            if captured_screens:
                print(
                    f"[info] Captured {len(captured_screens)} screen(s) for context.",
                    file=sys.stderr,
                )
            else:
                print(
                    "[info] No screenshots captured. Sending text-only request.",
                    file=sys.stderr,
                )

        request_messages = build_messages_payload(
            conversation_history=conversation_history,
            user_prompt=user_prompt,
            captured_screens=captured_screens,
        )

        try:
            full_response_text = stream_claude_response(
                worker_base_url=worker_base_url,
                model=parsed_arguments.model,
                system_prompt=parsed_arguments.system_prompt,
                messages=request_messages,
            )
        except requests.RequestException as request_error:
            print(f"[error] Network error: {request_error}", file=sys.stderr)
            continue
        except RuntimeError as runtime_error:
            print(f"[error] {runtime_error}", file=sys.stderr)
            continue

        spoken_response_text, pointing_metadata = parse_pointing_metadata(full_response_text)

        if pointing_metadata is not None and not pointing_metadata.indicates_no_target:
            target_label_text = (
                f" label='{pointing_metadata.label}'" if pointing_metadata.label else ""
            )
            target_screen_text = (
                f" screen={pointing_metadata.screen_number}" if pointing_metadata.screen_number else ""
            )
            print(
                (
                    "[point]"
                    f" x={pointing_metadata.x}"
                    f" y={pointing_metadata.y}"
                    f"{target_screen_text}"
                    f"{target_label_text}"
                ),
                file=sys.stderr,
            )

        cleaned_response_text = spoken_response_text.strip()
        if not cleaned_response_text:
            cleaned_response_text = "(empty response)"

        conversation_history.append((user_prompt, cleaned_response_text))
        if len(conversation_history) > MAX_HISTORY_EXCHANGES:
            conversation_history[:] = conversation_history[-MAX_HISTORY_EXCHANGES:]

        if parsed_arguments.no_tts:
            continue

        if not cleaned_response_text or cleaned_response_text == "(empty response)":
            continue

        try:
            tts_audio_bytes = request_tts_audio(
                worker_base_url=worker_base_url,
                text_to_speak=cleaned_response_text,
            )
            play_mp3_audio_bytes(tts_audio_bytes)
        except requests.RequestException as request_error:
            print(f"[warning] TTS network error: {request_error}", file=sys.stderr)
        except RuntimeError as runtime_error:
            print(f"[warning] {runtime_error}", file=sys.stderr)


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