whispypy-daemon.py

#!/usr/bin/env python3

import argparse
import configparser
from contextlib import contextmanager
import importlib.util
import json
import logging
import os
from pathlib import Path
import shutil
import signal
import struct
import subprocess
import sys
import tempfile
import time
from typing import Any, Generator, Optional, Union
import wave

import numpy as np
import whisper

# Audio file constants
BEEP_START_FILENAME = "BEEPTimer_Montre_numerique_bip_2_ID_2255_LS.wav"
BEEP_COMPLETE_FILENAME = "BEEPTimer_Montre_numerique_bip_1_ID_2254_LS.wav"

# Audio recording constants
SAMPLE_RATE = 16000  # Hz - Whisper's expected sample rate
CHANNELS = 1  # Mono audio
AUDIO_FORMAT = "f32"  # 32-bit float format for PipeWire
FLOAT32_BYTE_SIZE = 4  # Size of f32 in bytes

# Timing and validation constants
DEVICE_TEST_DURATION = 1.0  # seconds - Duration for device validation test

# File paths (will be replaced with proper temp files)
TEMP_AUDIO_FILENAME = "whispy_recording"  # Base filename for temporary audio (extension will be added based on engine)

# Terminal detection constants
TERMINAL_KEYWORDS = [
    "term",
    "konsole",
    "kitty",
    "alacritty",
    "ghostty",
    "wezterm",
    "foot",
    "gnome-terminal",
    "xterm",
    "urxvt",
    "st",
]  # Common terminal identifiers for window class/title matching

# State files for external indicators (e.g., Waybar)
RECORDING_STATE_FILE = Path("/tmp/whispypy_recording")
READY_STATE_FILE = Path("/tmp/whispypy_ready")

DEFAULT_SHERPA_ONNX_PARAKEET_INT8_MODEL = (
    "sherpa-onnx-nemo-parakeet-tdt-0.6b-v3-int8"
)


def _auto_onnx_threads() -> int:
    cpu_count = os.cpu_count() or 1
    return max(1, min(4, cpu_count // 2))


def _whispypy_cache_dir() -> Path:
    xdg_cache_home = os.environ.get("XDG_CACHE_HOME")
    base = Path(xdg_cache_home) if xdg_cache_home else (Path.home() / ".cache")
    return base / "whispypy"


def _is_valid_parakeet_onnx_dir(model_dir: Path) -> bool:
    return all(
        (model_dir / name).is_file()
        for name in ("encoder.int8.onnx", "decoder.int8.onnx", "joiner.int8.onnx", "tokens.txt")
    )


def ensure_sherpa_onnx_parakeet_model_dir(
    model_id: str,
    cache_dir: Optional[Union[str, Path]] = None,
) -> Path:
    """Ensure the sherpa-onnx model bundle exists locally; download if missing."""
    models_root = (
        Path(cache_dir) if cache_dir is not None else _whispypy_cache_dir()
    ) / "models"
    models_root.mkdir(parents=True, exist_ok=True)

    expected_dir = models_root / model_id
    if _is_valid_parakeet_onnx_dir(expected_dir):
        return expected_dir

    url = (
        "https://github.com/k2-fsa/sherpa-onnx/releases/download/asr-models/"
        f"{model_id}.tar.bz2"
    )
    logging.info("Downloading sherpa-onnx model bundle from %s", url)

    tmp_path: Optional[str] = None
    try:
        with tempfile.NamedTemporaryFile(suffix=".tar.bz2", delete=False) as tmp:
            tmp_path = tmp.name

        if shutil.which("curl"):
            download_cmd = [
                "curl",
                "-L",
                "-f",
                "-o",
                tmp_path,
                url,
            ]
        elif shutil.which("wget"):
            download_cmd = [
                "wget",
                "-O",
                tmp_path,
                url,
            ]
        else:
            raise RuntimeError(
                "Auto-download requires either 'curl' or 'wget' to be installed. "
                "Install one of them, or pass --parakeet-onnx-dir to point to a pre-downloaded bundle."
            )

        subprocess.run(
            download_cmd,
            check=True,
        )

        subprocess.run(
            [
                "tar",
                "-xjf",
                tmp_path,
                "-C",
                str(models_root),
            ],
            check=True,
        )
    finally:
        if tmp_path:
            Path(tmp_path).unlink(missing_ok=True)

    if _is_valid_parakeet_onnx_dir(expected_dir):
        return expected_dir

    # Some archives may not extract to the expected directory name.
    candidates = [
        p for p in models_root.iterdir() if p.is_dir() and _is_valid_parakeet_onnx_dir(p)
    ]
    if len(candidates) == 1:
        return candidates[0]

    raise FileNotFoundError(
        f"Downloaded model bundle but could not find required files under {models_root}. "
        f"Expected {expected_dir} with encoder/decoder/joiner/tokens."
    )


class SherpaOnnxParakeetInt8Transcriber:
    def __init__(
        self,
        model_dir: Union[str, Path],
        provider: str = "cpu",
        num_threads: Optional[int] = None,
    ):
        try:
            import sherpa_onnx
        except ImportError as e:
            raise ImportError(
                "sherpa-onnx is required for engine 'parakeet_onnx_int8'. "
                "Install with the optional extra (to be added): whispypy[parakeet-onnx]"
            ) from e

        self._sherpa_onnx = sherpa_onnx
        self.model_dir = Path(model_dir)

        encoder = self.model_dir / "encoder.int8.onnx"
        decoder = self.model_dir / "decoder.int8.onnx"
        joiner = self.model_dir / "joiner.int8.onnx"
        tokens = self.model_dir / "tokens.txt"

        missing = [
            str(p)
            for p in (encoder, decoder, joiner, tokens)
            if not p.is_file()
        ]
        if missing:
            raise FileNotFoundError(
                "Missing required Parakeet INT8 files in model dir. Missing: "
                + ", ".join(missing)
            )

        self.num_threads = num_threads if num_threads is not None else _auto_onnx_threads()
        self.provider = provider

        model_load_start = time.time()

        kwargs: dict[str, Any] = dict(
            encoder=str(encoder),
            decoder=str(decoder),
            joiner=str(joiner),
            tokens=str(tokens),
            num_threads=self.num_threads,
            sample_rate=SAMPLE_RATE,
            feature_dim=80,
            decoding_method="greedy_search",
            model_type="nemo_transducer",
            debug=False,
        )

        # Provider support varies by sherpa-onnx version; try best-effort.
        if self.provider in {"cpu", "cuda"}:
            kwargs["provider"] = self.provider

        try:
            try:
                self.recognizer = self._sherpa_onnx.OfflineRecognizer.from_transducer(
                    **kwargs
                )
            except TypeError:
                # Older sherpa-onnx may not accept provider/model_type kwargs.
                kwargs.pop("provider", None)
                kwargs.pop("model_type", None)
                self.recognizer = self._sherpa_onnx.OfflineRecognizer.from_transducer(
                    **kwargs
                )
        except Exception as e:
            if self.provider == "cuda":
                logging.warning(
                    "Failed to initialize sherpa-onnx with provider=cuda (%s); falling back to cpu",
                    e,
                )
                self.provider = "cpu"
                kwargs["provider"] = "cpu"
                try:
                    self.recognizer = self._sherpa_onnx.OfflineRecognizer.from_transducer(
                        **kwargs
                    )
                except TypeError:
                    kwargs.pop("provider", None)
                    kwargs.pop("model_type", None)
                    self.recognizer = self._sherpa_onnx.OfflineRecognizer.from_transducer(
                        **kwargs
                    )
            else:
                raise

        model_load_time = time.time() - model_load_start
        logging.info(
            "Sherpa-ONNX Parakeet INT8 model loaded in %.2f seconds (provider=%s, threads=%s)",
            model_load_time,
            self.provider,
            self.num_threads,
        )

    def transcribe_wav(self, wav_path: Union[str, Path]) -> str:
        stream = self.recognizer.create_stream()

        with wave.open(str(wav_path)) as wf:
            if wf.getnchannels() != 1:
                raise ValueError(f"Expected mono wav, got channels={wf.getnchannels()}")
            if wf.getsampwidth() != 2:
                raise ValueError(
                    f"Expected 16-bit PCM wav, got sampwidth={wf.getsampwidth()} bytes"
                )
            num_frames = wf.getnframes()
            pcm = wf.readframes(num_frames)
            samples_i16 = np.frombuffer(pcm, dtype=np.int16)
            samples_f32 = samples_i16.astype(np.float32) / 32768.0
            sample_rate = wf.getframerate()

        stream.accept_waveform(sample_rate, samples_f32)
        self.recognizer.decode_streams([stream])
        return str(stream.result.text).strip()


def get_config_file() -> Path:
    """Get the configuration file path following XDG Base Directory specification."""
    xdg_config_home = os.environ.get("XDG_CONFIG_HOME", Path.home() / ".config")
    config_dir = Path(xdg_config_home) / "whispypy"
    config_dir.mkdir(parents=True, exist_ok=True)
    return config_dir / "config.conf"


# Configuration
CONFIG_FILE = get_config_file()


class ConfigManager:
    """Manages configuration file operations with cached config parsing."""

    def __init__(self, config_file: Path = CONFIG_FILE):
        self.config_file = config_file
        self._config: Optional[configparser.ConfigParser] = None
        self._config_mtime: Optional[float] = None

    def _get_config(self) -> Optional[configparser.ConfigParser]:
        """Get cached config, reloading if file changed."""
        if not self.config_file.exists():
            self._config = None
            self._config_mtime = None
            return None

        current_mtime = self.config_file.stat().st_mtime
        if self._config is None or self._config_mtime != current_mtime:
            self._config = configparser.ConfigParser()
            self._config.read(self.config_file)
            self._config_mtime = current_mtime

        return self._config

    def _invalidate_cache(self) -> None:
        """Invalidate cached config after writes."""
        self._config = None
        self._config_mtime = None

    def _load_config_value(
        self, key: str, log_msg: Optional[str] = None, log_level: str = "info"
    ) -> Optional[str]:
        """Generic method to load a config value."""
        config = self._get_config()
        if config is None:
            return None

        try:
            value = config.get("DEFAULT", key, fallback=None)
            if value and log_msg:
                getattr(logging, log_level)(log_msg.format(value=value))
            return value
        except Exception as e:
            logging.debug(f"Error reading {key} from config: {e}")
            return None

    def save_device(self, device: str) -> None:
        """Save device configuration to config file."""
        config = self._get_config() or configparser.ConfigParser()

        if "DEFAULT" not in config:
            config.add_section("DEFAULT")

        config.set("DEFAULT", "device", device)

        with open(self.config_file, "w") as f:
            config.write(f)

        self._invalidate_cache()
        logging.info(f"Device '{device}' saved to {self.config_file}")

    def load_device(self) -> Optional[str]:
        """Load device configuration from config file."""
        return self._load_config_value(
            "device", "Using device from config: {value}", "info"
        )

    def load_dotool_layout(self) -> Optional[str]:
        """Load DOTOOL_XKB_LAYOUT configuration from config file."""
        return self._load_config_value(
            "dotool_xkb_layout", "Using dotool XKB layout from config: {value}", "debug"
        )

    def load_dotool_variant(self) -> Optional[str]:
        """Load DOTOOL_XKB_VARIANT configuration from config file."""
        return self._load_config_value(
            "dotool_xkb_variant", "Using dotool XKB variant from config: {value}", "debug"
        )

    def validate_config(self) -> bool:
        """Validate configuration file format and values."""
        config = self._get_config()
        if config is None:
            return True  # No config file is valid (will use defaults)

        try:

            # Check if DEFAULT section exists
            if "DEFAULT" not in config:
                logging.warning("Configuration file missing DEFAULT section")
                return False

            # Validate device if present
            device = config.get("DEFAULT", "device", fallback=None)
            if device:
                # Basic validation - device name should not be empty
                if not device.strip():
                    logging.warning("Device name in config is empty")
                    return False

                # Validate device name format (basic checks)
                if len(device.strip()) < 3:
                    logging.warning("Device name appears too short, may be invalid")
                    return False

            # Validate other audio-related settings if present
            # Validate sample_rate if present
            sample_rate_value = config.get("DEFAULT", "sample_rate", fallback=None)
            if sample_rate_value is not None:
                try:
                    sample_rate = int(sample_rate_value)
                    valid_sample_rates = [8000, 16000, 22050, 44100, 48000]
                    if sample_rate not in valid_sample_rates:
                        logging.warning(
                            f"Invalid sample_rate value '{sample_rate}'. "
                            f"Valid values: {valid_sample_rates}"
                        )
                        return False
                except ValueError:
                    logging.warning(
                        f"Invalid sample_rate value '{sample_rate_value}' - not an integer"
                    )
                    return False

            # Validate channels if present
            channels_value = config.get("DEFAULT", "channels", fallback=None)
            if channels_value is not None:
                try:
                    channels = int(channels_value)
                    valid_channels = [1, 2]
                    if channels not in valid_channels:
                        logging.warning(
                            f"Invalid channels value '{channels}'. "
                            f"Valid values: {valid_channels}"
                        )
                        return False
                except ValueError:
                    logging.warning(
                        f"Invalid channels value '{channels_value}' - not an integer"
                    )
                    return False

            # Validate audio_format if present
            audio_format_value = config.get("DEFAULT", "audio_format", fallback=None)
            if audio_format_value is not None:
                audio_format = audio_format_value.strip()
                valid_formats = ["f32", "s16", "s24", "s32"]
                if audio_format not in valid_formats:
                    logging.warning(
                        f"Invalid audio_format value '{audio_format}'. "
                        f"Valid values: {valid_formats}"
                    )
                    return False

            # Validate dotool_xkb_layout if present
            dotool_layout_value = config.get(
                "DEFAULT", "dotool_xkb_layout", fallback=None
            )
            if dotool_layout_value is not None:
                dotool_layout = dotool_layout_value.strip()
                if not dotool_layout:
                    logging.warning("dotool_xkb_layout value is empty")
                    return False

            # Validate dotool_xkb_variant if present
            dotool_variant_value = config.get(
                "DEFAULT", "dotool_xkb_variant", fallback=None
            )
            if dotool_variant_value is not None:
                dotool_variant = dotool_variant_value.strip()
                if not dotool_variant:
                    logging.warning("dotool_xkb_variant value is empty")
                    return False

            logging.debug("Configuration validation successful")
            return True

        except Exception as e:
            logging.error(f"Configuration validation failed: {e}")
            return False


def load_audio_f32(filepath: Union[str, Path]) -> np.ndarray:
    """Load audio file as float32 samples."""
    with open(filepath, "rb") as f:
        data = f.read()

    # Convert bytes to float32
    num_floats = len(data) // FLOAT32_BYTE_SIZE
    floats = struct.unpack(f"<{num_floats}f", data)
    return np.array(floats, dtype=np.float32)


def load_audio_s16_as_f32(filepath: Union[str, Path]) -> np.ndarray:
    """Load 16-bit PCM audio and convert to float32 samples in [-1, 1]."""
    with open(filepath, "rb") as f:
        data = f.read()

    num_samples = len(data) // 2
    if num_samples == 0:
        return np.array([], dtype=np.float32)

    ints = struct.unpack(f"<{num_samples}h", data)
    floats = np.asarray(ints, dtype=np.float32) / 32768.0
    return floats


def _play_beep_file(filename: str, beep_type: str) -> None:
    """Play a beep sound file with fallback options.

    Args:
        filename: The beep sound filename (from the assets directory)
        beep_type: Description of the beep type for logging (e.g., "start", "completion")
    """
    # Construct path to beep sound file
    beep_file = Path(__file__).parent / "assets" / filename

    if not beep_file.exists():
        logging.debug(f"Beep file not found: {beep_file}")
        _try_terminal_beep_fallback(beep_type)
        return

    # Audio players to try, in order of preference
    audio_players = [
        "aplay",  # ALSA player
        "paplay",  # PulseAudio player
        "pw-play",  # PipeWire player
    ]

    for player in audio_players:
        if _try_audio_player(player, str(beep_file), beep_type):
            return

    # All audio players failed, try terminal beep
    _try_terminal_beep_fallback(beep_type)


def _try_audio_player(player: str, audio_file: str, beep_type: str) -> bool:
    """Try to play audio with a specific player.

    Returns:
        bool: True if successful, False otherwise
    """
    try:
        result = subprocess.run(
            [player, audio_file],
            capture_output=True,
            text=True,
            timeout=10,  # 10 second timeout
            check=False,  # Don't raise exception on non-zero exit
        )

        if result.returncode == 0:
            logging.debug(f"{beep_type.capitalize()} beep played via {player}")
            return True
        else:
            # Log the actual error for debugging
            error_parts = [f"exit code {result.returncode}"]
            if result.stderr.strip():
                error_parts.append(f"stderr: {result.stderr.strip()}")
            logging.debug(f"{player} failed: {', '.join(error_parts)}")
            return False

    except subprocess.TimeoutExpired:
        logging.debug(f"{player} timed out after 10 seconds")
        return False
    except FileNotFoundError:
        logging.debug(f"{player} command not found")
        return False
    except PermissionError:
        logging.debug(f"{player} permission denied")
        return False
    except Exception as e:
        logging.debug(f"{player} failed with exception: {e}")
        return False


def _try_terminal_beep_fallback(beep_type: str) -> None:
    """Try to play terminal beep as fallback."""
    try:
        result = subprocess.run(
            ["printf", "\\a"], capture_output=True, text=True, timeout=5, check=False
        )

        if result.returncode == 0:
            logging.debug(f"{beep_type.capitalize()} beep played via printf (fallback)")
        else:
            logging.debug(f"Terminal beep failed with exit code {result.returncode}")
            logging.debug(f"Could not play {beep_type} beep")

    except Exception as e:
        logging.debug(f"Terminal beep failed with exception: {e}")
        logging.debug(f"Could not play {beep_type} beep")


def play_start_beep() -> None:
    """Play a start beep sound to indicate recording is starting."""
    _play_beep_file(BEEP_START_FILENAME, "start")


def play_completion_beep() -> None:
    """Play a completion beep sound to indicate transcription is ready."""
    _play_beep_file(BEEP_COMPLETE_FILENAME, "completion")


def copy_to_clipboard(text: str) -> bool:
    """Copy text to clipboard using the appropriate tool for the current display server."""
    # Check if we're on Wayland
    if os.getenv("WAYLAND_DISPLAY"):
        # Use wl-copy for Wayland
        try:
            subprocess.run(["wl-copy", text], check=True)
            logging.info("Text copied to clipboard (wl-copy)")
            return True
        except (subprocess.CalledProcessError, FileNotFoundError) as e:
            logging.warning(f"wl-copy failed: {e}")

    # Check if we're on X11
    if os.getenv("DISPLAY"):
        # Try xclip first (more common)
        try:
            subprocess.run(
                ["xclip", "-selection", "clipboard"], input=text, text=True, check=True
            )
            logging.info("Text copied to clipboard (xclip)")
            return True
        except (subprocess.CalledProcessError, FileNotFoundError):
            # Fallback to xsel
            try:
                subprocess.run(
                    ["xsel", "--clipboard", "--input"],
                    input=text,
                    text=True,
                    check=True,
                )
                logging.info("Text copied to clipboard (xsel)")
                return True
            except (subprocess.CalledProcessError, FileNotFoundError) as e:
                logging.warning(f"X11 clipboard tools failed: {e}")

    # Fallback: try to detect clipboard tools
    clipboard_tools = [
        (["xclip", "-selection", "clipboard"], True),  # input via stdin
        (["xsel", "--clipboard", "--input"], True),  # input via stdin
        (["wl-copy"], False),  # text as argument
    ]

    for cmd, use_stdin in clipboard_tools:
        try:
            if use_stdin:
                subprocess.run(cmd, input=text, text=True, check=True)
            else:
                subprocess.run(cmd + [text], check=True)
            tool_name = cmd[0]
            logging.info(f"Text copied to clipboard ({tool_name})")
            return True
        except (subprocess.CalledProcessError, FileNotFoundError):
            continue

    logging.error("Failed to copy to clipboard: no suitable clipboard tool found")
    return False


def _detect_terminal_window() -> bool:
    """Detect if the focused window is a terminal application.

    Returns True if terminal detected, False otherwise (including on detection failure).
    """
    # Try Wayland (Hyprland) detection first
    if os.getenv("WAYLAND_DISPLAY"):
        try:
            result = subprocess.run(
                ["hyprctl", "activewindow", "-j"],
                capture_output=True,
                text=True,
                check=True
            )
            window_info = json.loads(result.stdout)
            window_class = window_info.get("class", "").lower()
            window_title = window_info.get("title", "").lower()
            is_terminal = any(
                keyword in window_class or keyword in window_title
                for keyword in TERMINAL_KEYWORDS
            )
            logging.debug(f"Window class: {window_class}, is_terminal: {is_terminal}")
            return is_terminal
        except (subprocess.CalledProcessError, FileNotFoundError, json.JSONDecodeError):
            pass

    # Try X11 detection
    if os.getenv("DISPLAY"):
        try:
            result = subprocess.run(
                ["xdotool", "getactivewindow", "getwindowclassname"],
                capture_output=True,
                text=True,
                check=True,
            )
            window_class = result.stdout.strip().lower()
            window_title = ""
            try:
                title_result = subprocess.run(
                    ["xdotool", "getactivewindow", "getwindowname"],
                    capture_output=True,
                    text=True,
                    check=True,
                )
                window_title = title_result.stdout.strip().lower()
            except (subprocess.CalledProcessError, FileNotFoundError):
                pass

            is_terminal = any(
                keyword in window_class or keyword in window_title
                for keyword in TERMINAL_KEYWORDS
            )
            logging.debug(f"Window class: {window_class}, is_terminal: {is_terminal}")
            return is_terminal
        except (subprocess.CalledProcessError, FileNotFoundError):
            pass

    logging.debug("Could not detect window type, defaulting to GUI paste")
    return False


def paste_from_clipboard() -> bool:
    """Paste text from clipboard using the appropriate tool for the current display server."""
    # Check if we're on Wayland
    if os.getenv("WAYLAND_DISPLAY"):
        logging.debug("Detected Wayland display server for pasting")
        is_terminal = _detect_terminal_window()

        # Use wtype for Wayland (simulates typing)
        # Use Ctrl+Shift+V for terminals, Ctrl+V for GUI apps
        try:
            if is_terminal:
                logging.debug("Attempting to paste using wtype with Ctrl+Shift+V (terminal)")
                subprocess.run(["wtype", "-M", "ctrl", "-M", "shift", "v"], check=True)
                logging.info("Pasted from clipboard (wtype with Ctrl+Shift+V)")
            else:
                logging.debug("Attempting to paste using wtype with Ctrl+V (GUI)")
                subprocess.run(["wtype", "-M", "ctrl", "v"], check=True)
                logging.info("Pasted from clipboard (wtype with Ctrl+V)")
            return True
        except (subprocess.CalledProcessError, FileNotFoundError):
            pass

        # If wtype failed, try ydotool
        try:
            # Add small delay before paste to let window manager settle
            time.sleep(0.1)
            # Use key codes: 29 is left ctrl, 42 is left shift, 47 is v
            # Format: "keycode:state" where :1 = key down, :0 = key up
            if is_terminal:
                logging.debug("Attempting to paste using ydotool with Ctrl+Shift+V (terminal)")
                subprocess.run(["ydotool", "key", "29:1", "42:1", "47:1", "47:0", "42:0", "29:0"], check=True)
                logging.info("Pasted from clipboard (ydotool with Ctrl+Shift+V)")
            else:
                logging.debug("Attempting to paste using ydotool with Ctrl+V (GUI)")
                subprocess.run(["ydotool", "key", "29:1", "47:1", "47:0", "29:0"], check=True)
                logging.info("Pasted from clipboard (ydotool with Ctrl+V)")
            return True
        except (subprocess.CalledProcessError, FileNotFoundError):
            # Final fallback: try wl-paste + dotool with layout settings
            try:
                logging.debug(
                    "Attempting to paste using dotool with layout settings"
                )
                # Load dotool configuration
                config_manager = ConfigManager()
                dotool_layout = config_manager.load_dotool_layout()
                dotool_variant = config_manager.load_dotool_variant()

                # Build command with optional environment variables
                env_vars = []
                if dotool_layout:
                    env_vars.append(f"DOTOOL_XKB_LAYOUT={dotool_layout}")
                if dotool_variant:
                    env_vars.append(f"DOTOOL_XKB_VARIANT={dotool_variant}")

                env_prefix = " ".join(env_vars)
                if env_prefix:
                    command = f"wl-paste | sed 's/^/type /' | {env_prefix} dotool"
                else:
                    command = "wl-paste | sed 's/^/type /' | dotool"

                subprocess.run(
                    command,
                    shell=True,
                    check=True,
                )
                logging.info("Pasted from clipboard (dotool)")
                return True
            except (subprocess.CalledProcessError, FileNotFoundError):
                pass

    # Check if we're on X11
    if os.getenv("DISPLAY"):
        is_terminal = _detect_terminal_window()

        # Use xdotool for X11 - Ctrl+Shift+V for terminals, Ctrl+V for GUI apps
        try:
            if is_terminal:
                logging.debug("Attempting to paste using xdotool with Ctrl+Shift+V (terminal)")
                subprocess.run(["xdotool", "key", "ctrl+shift+v"], check=True)
                logging.info("Pasted from clipboard (xdotool with Ctrl+Shift+V)")
            else:
                logging.debug("Attempting to paste using xdotool with Ctrl+V (GUI)")
                subprocess.run(["xdotool", "key", "ctrl+v"], check=True)
                logging.info("Pasted from clipboard (xdotool with Ctrl+V)")
            return True
        except (subprocess.CalledProcessError, FileNotFoundError) as e:
            logging.warning(f"xdotool failed: {e}")

    # Fallback: try all available paste tools
    time.sleep(0.1)
    paste_tools = [
        ["xdotool", "key", "ctrl+v"],  # X11
        ["wtype", "-M", "ctrl", "v"],  # Wayland
        ["ydotool", "key", "29:1", "47:1", "47:0", "29:0"],  # Wayland alternative (ctrl+v)
    ]

    for cmd in paste_tools:
        try:
            subprocess.run(cmd, check=True)
            tool_name = cmd[0]
            logging.info(f"Pasted from clipboard ({tool_name})")
            return True
        except (subprocess.CalledProcessError, FileNotFoundError):
            continue

    logging.error("Failed to paste from clipboard: no suitable paste tool found")
    logging.error("Available tools: xdotool (X11), wtype/ydotool (Wayland)")
    return False


@contextmanager
def managed_subprocess(
    args: list[str],
) -> Generator[subprocess.Popen[bytes], None, None]:
    """Context manager for subprocess handling with proper cleanup."""
    proc = None
    try:
        proc = subprocess.Popen(args, stderr=subprocess.PIPE, stdout=subprocess.PIPE)
        yield proc
    finally:
        if proc:
            proc.terminate()
            proc.wait()


class WhispypyDaemon:
    """Signal-controlled audio transcription daemon using OpenAI Whisper or NVIDIA Parakeet."""

    def __init__(
        self,
        model_path: str,
        device_name: str,
        engine: str = "whisper",
        parakeet_onnx_dir: Optional[str] = None,
        parakeet_onnx_model_id: str = DEFAULT_SHERPA_ONNX_PARAKEET_INT8_MODEL,
        parakeet_onnx_cache_dir: Optional[str] = None,
        onnx_provider: str = "cpu",
        onnx_threads: Optional[int] = None,
        keep_audio: bool = False,
        autopaste: bool = False,
    ):
        self.model_path = model_path
        self.device_name = device_name
        self.engine = engine
        self.parakeet_onnx_dir = parakeet_onnx_dir
        self.parakeet_onnx_model_id = parakeet_onnx_model_id
        self.parakeet_onnx_cache_dir = parakeet_onnx_cache_dir
        self.onnx_provider = onnx_provider
        self.onnx_threads = onnx_threads
        self.keep_audio = keep_audio
        self.autopaste = autopaste

        # Create temporary file for audio recording with appropriate extension
        audio_extension = ".wav" if engine in {"parakeet", "parakeet_onnx_int8"} else ".au"
        self.temp_audio_file = Path(tempfile.gettempdir()) / (
            TEMP_AUDIO_FILENAME + audio_extension
        )

        # State
        self.recording = False
        self.running = True
        self.pw_record_proc: Optional[subprocess.Popen[bytes]] = None

        # Model attribute - will be assigned in load methods
        self.model: Any = None

        # Load the appropriate model
        if self.engine == "whisper":
            self._load_whisper_model()
        elif self.engine == "parakeet":
            self._load_parakeet_model()
        elif self.engine == "parakeet_onnx_int8":
            self._load_parakeet_onnx_int8_model()
        else:
            raise ValueError(f"Unsupported engine: {self.engine}")

        # Setup signal handlers
        signal.signal(signal.SIGINT, self._handle_sigint)
        signal.signal(signal.SIGUSR2, self._handle_sigusr2)

    def _load_whisper_model(self) -> None:
        """Load Whisper model."""
        logging.info(f"Loading Whisper model from {self.model_path}...")
        model_load_start = time.time()
        self.model = whisper.load_model(self.model_path)
        model_load_time = time.time() - model_load_start
        logging.info(f"Whisper model loaded in {model_load_time:.2f} seconds")

    def _load_parakeet_model(self) -> None:
        """Load Parakeet model."""
        try:
            import nemo.collections.asr as nemo_asr
        except ImportError:
            raise ImportError(
                "Parakeet (NeMo) is not available. Please see README for installation instructions."
            )

        logging.info(f"Loading Parakeet model from {self.model_path}...")
        model_load_start = time.time()
        self.model = nemo_asr.models.ASRModel.from_pretrained(
            model_name=self.model_path
        )
        model_load_time = time.time() - model_load_start
        logging.info(f"Parakeet model loaded in {model_load_time:.2f} seconds")

    def _load_parakeet_onnx_int8_model(self) -> None:
        """Load Parakeet INT8 model via sherpa-onnx."""
        if not self.parakeet_onnx_dir:
            self.parakeet_onnx_dir = str(
                ensure_sherpa_onnx_parakeet_model_dir(
                    model_id=self.parakeet_onnx_model_id,
                    cache_dir=self.parakeet_onnx_cache_dir,
                )
            )

        self.model = SherpaOnnxParakeetInt8Transcriber(
            model_dir=self.parakeet_onnx_dir,
            provider=self.onnx_provider,
            num_threads=self.onnx_threads,
        )

    def _is_alsa_device(self) -> bool:
        """Return True if device_name looks like a raw ALSA device."""
        return self.device_name.startswith(("hw:", "plughw:"))

    def _get_alsa_device(self) -> str:
        """Get ALSA device name, converting hw: to plughw: for format conversion."""
        if self.device_name.startswith("hw:"):
            return self.device_name.replace("hw:", "plughw:", 1)
        return self.device_name

    def validate_device(self) -> bool:
        """Validate that the audio device exists and is accessible."""
        try:
            # Test device by attempting a very short recording
            audio_extension = (
                ".wav" if self.engine in {"parakeet", "parakeet_onnx_int8"} else ".au"
            )
            with tempfile.NamedTemporaryFile(
                suffix=audio_extension, delete=False
            ) as test_file:
                test_file_path = test_file.name

            if self._is_alsa_device():
                # ALSA: use arecord with the same device transformation as recording
                with managed_subprocess(
                    [
                        "arecord",
                        "-D", self._get_alsa_device(),
                        "-f", "S16_LE",
                        "-r", str(SAMPLE_RATE),
                        "-c", str(CHANNELS),
                        "-t", "raw",
                        test_file_path,
                    ]
                ) as _:
                    time.sleep(DEVICE_TEST_DURATION)
            else:
                # PipeWire: use pw-record
                pw_format = AUDIO_FORMAT
                if self.engine == "parakeet_onnx_int8":
                    pw_format = "s16"
                with managed_subprocess(
                    [
                        "pw-record",
                        f"--target={self.device_name}",
                        f"--format={pw_format}",
                        f"--rate={SAMPLE_RATE}",
                        f"--channels={CHANNELS}",
                        test_file_path,
                    ]
                ) as _:
                    # Let it record for the test duration then it will be terminated
                    time.sleep(DEVICE_TEST_DURATION)

            size = Path(test_file_path).stat().st_size
            # Clean up test file
            Path(test_file_path).unlink(missing_ok=True)

            if size == 0:
                logging.debug("Device validation produced empty audio file")
                return False

            # If we got here without exception, the device is accessible
            return True

        except Exception as e:
            logging.debug(f"Device validation failed: {e}")
            return False

    def _handle_sigint(self, signum: int, frame: Any) -> None:
        """Handle SIGINT (Ctrl+C) for clean shutdown."""
        logging.info("Received SIGINT (Ctrl+C). Shutting down...")

        # Stop any ongoing recording
        if self.recording and self.pw_record_proc:
            logging.info("Stopping ongoing recording...")
            self.pw_record_proc.terminate()
            self.pw_record_proc.wait()

        # Remove ready state file
        try:
            READY_STATE_FILE.unlink(missing_ok=True)
        except Exception:
            pass

        self.running = False

    def _handle_sigusr2(self, signum: int, frame: Any) -> None:
        """Handle SIGUSR2 signal to toggle recording state."""
        try:
            logging.info(
                f"Received SIGUSR2 signal! Current recording state: {self.recording}"
            )

            if not self.recording:
                self._start_recording()
            else:
                self._stop_recording_and_transcribe()

        except Exception as e:
            logging.error(f"Error in signal handler: {e}")
            import traceback

            traceback.print_exc()

    def _start_recording(self) -> None:
        """Start audio recording."""
        logging.info("Starting recording...")
        play_start_beep()

        if self._is_alsa_device():
            # ALSA: arecord - use raw 16-bit PCM for Whisper, WAV container for Parakeet
            alsa_container = (
                "wav" if self.engine in {"parakeet", "parakeet_onnx_int8"} else "raw"
            )
            cmd = [
                "arecord",
                "-D", self._get_alsa_device(),
                "-f", "S16_LE",
                "-r", str(SAMPLE_RATE),
                "-c", str(CHANNELS),
                "-t", alsa_container,
                str(self.temp_audio_file),
            ]
        else:
            # PipeWire: pw-record
            pw_format = AUDIO_FORMAT
            if self.engine == "parakeet_onnx_int8":
                pw_format = "s16"
            cmd = [
                "pw-record",
                f"--target={self.device_name}",
                f"--format={pw_format}",
                f"--rate={SAMPLE_RATE}",
                f"--channels={CHANNELS}",
                str(self.temp_audio_file),
            ]

        self.pw_record_proc = subprocess.Popen(
            cmd,
            stderr=subprocess.PIPE,
            stdout=subprocess.PIPE,
        )
        self.recording = True
        # Create state file for external indicators (e.g., Waybar)
        try:
            RECORDING_STATE_FILE.touch()
        except Exception as e:
            # State file creation failed, but recording is already started
            # Log warning but don't abort - recording is more important
            logging.warning(f"Failed to create recording state file: {e}", exc_info=True)
        logging.info("Recording started successfully")

    def _stop_recording_and_transcribe(self) -> None:
        """Stop recording and perform transcription."""
        logging.info("Stopping recording...")
        if self.pw_record_proc:
            self.pw_record_proc.terminate()
            self.pw_record_proc.wait()
            self.pw_record_proc = None
        self.recording = False
        # Remove state file for external indicators (e.g., Waybar)
        try:
            RECORDING_STATE_FILE.unlink(missing_ok=True)
        except Exception:
            pass
        logging.info("Recording stopped")

        # Check if audio file exists and has content
        if not self.temp_audio_file.exists():
            logging.error(f"Audio file {self.temp_audio_file} not found!")
            return

        file_size = self.temp_audio_file.stat().st_size
        logging.info(f"Audio file size: {file_size} bytes")

        if file_size == 0:
            logging.warning("Audio file is empty!")
            return

        # Transcribe with appropriate engine
        logging.info(f"Transcribing with {self.engine.capitalize()}...")
        transcription_start = time.time()

        if self.engine == "whisper":
            # Load audio samples for Whisper
            logging.info("Loading audio samples...")
            if self._is_alsa_device():
                samples = load_audio_s16_as_f32(self.temp_audio_file)
            else:
                samples = load_audio_f32(self.temp_audio_file)
            logging.info(f"Loaded {len(samples)} audio samples")

            result = self.model.transcribe(
                samples, fp16=False, language=None, task="transcribe"
            )
            text = result["text"].strip()
        elif self.engine == "parakeet":
            # Parakeet expects a list of file paths
            result = self.model.transcribe([str(self.temp_audio_file)])
            text = result[0].text.strip()
        elif self.engine == "parakeet_onnx_int8":
            text = self.model.transcribe_wav(self.temp_audio_file)
        else:
            raise ValueError(f"Unsupported engine: {self.engine}")

        transcription_time = time.time() - transcription_start

        logging.info(f"Transcription completed in {transcription_time:.2f} seconds")
        logging.info(f"Transcription result: '{text}'")

        if not self.keep_audio:
            self.temp_audio_file.unlink(missing_ok=True)

        # Copy text to clipboard
        copy_to_clipboard(text)

        # Auto-paste if enabled
        if self.autopaste:
            logging.info("Auto-pasting transcribed text...")
            # Small delay to ensure clipboard is ready
            time.sleep(0.1)
            paste_success = paste_from_clipboard()
            if not paste_success:
                logging.warning("Auto-paste failed, but text is still in clipboard")
        else:
            # Play completion beep only when autopaste is disabled
            # When autopaste is enabled, the text is automatically pasted so no beep needed
            # to know when it's ready
            play_completion_beep()

    def run(self) -> None:
        """Run the daemon main loop."""
        # Print PID for easy signal sending
        pid = os.getpid()
        logging.info(f"Script PID: {pid}")
        logging.info(
            f"To send signal start/stop from another terminal: kill -USR2 {pid}"
        )
        logging.info(f"To send signal exit from another terminal: kill -SIGINT {pid}")
        logging.info(f"Using audio device: {self.device_name}")
        logging.info(f"Using transcription engine: {self.engine}")
        if self.autopaste:
            logging.info(
                "Auto-paste is enabled - transcribed text will be pasted automatically"
            )
        else:
            logging.info(
                "Auto-paste is disabled - transcribed text will only be copied to clipboard"
            )

        # Validate device before starting
        if not self.validate_device():
            logging.error(
                f"Audio device '{self.device_name}' is not accessible or working"
            )
            logging.error("Please run test_audio_devices.py to find a working device")
            sys.exit(1)

        logging.info("Device validation successful")
        logging.info("Ready. Send SIGUSR2 to start/stop recording.")
        logging.info("Press Ctrl+C to exit.")

        # Create ready state file for external indicators (e.g., Waybar)
        try:
            READY_STATE_FILE.touch()
        except Exception as e:
            logging.warning(f"Could not create ready state file: {e}")

        # Initial beep to indicate readiness
        play_completion_beep()

        # Wait for signals
        try:
            while self.running:
                signal.pause()
        except KeyboardInterrupt:
            # This shouldn't happen since we handle SIGINT, but just in case
            self.running = False

        logging.info("Daemon stopped.")


def main() -> None:
    parser = argparse.ArgumentParser(
        description="Audio transcription with signal control using Whisper or Parakeet"
    )
    parser.add_argument(
        "model_path",
        nargs="?",
        default="base",
        help=(
            "Path to the model or model name. "
            "For Whisper: tiny, base, small, medium, large, large-v2, large-v3. "
            "For Parakeet (NeMo): nvidia/parakeet-tdt-0.6b-v3. "
            "For Parakeet INT8 (Sherpa-ONNX): optionally pass a sherpa-onnx bundle id as the positional argument (default: base)"
        ),
    )
    parser.add_argument(
        "--engine",
        "-e",
        choices=["whisper", "parakeet", "parakeet_onnx_int8"],
        default="whisper",
        help="Transcription engine to use (default: whisper)",
    )

    parser.add_argument(
        "--parakeet-onnx-dir",
        default=None,
        help="Directory containing encoder.int8.onnx/decoder.int8.onnx/joiner.int8.onnx/tokens.txt. If omitted, whispypy will auto-download the model bundle.",
    )
    parser.add_argument(
        "--parakeet-onnx-model-id",
        default=DEFAULT_SHERPA_ONNX_PARAKEET_INT8_MODEL,
        help="Sherpa-ONNX model bundle ID to download when --parakeet-onnx-dir is omitted",
    )
    parser.add_argument(
        "--parakeet-onnx-cache-dir",
        default=None,
        help="Override cache directory for auto-downloaded sherpa-onnx models (defaults to XDG cache)",
    )
    parser.add_argument(
        "--onnx-provider",
        choices=["cpu", "cuda"],
        default="cpu",
        help="Execution provider for sherpa-onnx (default: cpu). If cuda is unavailable, it will fall back to cpu.",
    )
    parser.add_argument(
        "--onnx-threads",
        type=int,
        default=None,
        help="Number of threads for sherpa-onnx (default: auto)",
    )
    parser.add_argument(
        "--check-model",
        action="store_true",
        help="Load the selected model and exit (useful for verifying parakeet_onnx_int8 setup)",
    )
    parser.add_argument(
        "--device",
        "-d",
        default=None,
        help="Audio input device name. If not provided, will try to load from XDG config (~/.config/whispypy/config.conf). Use test_audio_devices.py to find working devices.",
    )
    parser.add_argument(
        "--keep-audio",
        action="store_true",
        help="Keep the temporary audio file after transcription",
    )
    parser.add_argument(
        "--autopaste",
        action="store_true",
        help="Automatically paste transcribed text after copying to clipboard",
    )
    parser.add_argument(
        "--verbose", "-v", action="store_true", help="Enable verbose logging"
    )

    args = parser.parse_args()

    # Setup logging
    log_level = logging.DEBUG if args.verbose else logging.INFO
    logging.basicConfig(
        level=log_level,
        format="%(asctime)s - %(levelname)s - %(message)s",
        handlers=[logging.StreamHandler(sys.stderr)],
    )

    # Validate engine availability
    if args.engine == "parakeet":
        if importlib.util.find_spec("nemo.collections.asr") is None:
            logging.error("Parakeet engine selected but NeMo is not available.")
            logging.error("Please see README for installation instructions.")
            sys.exit(1)

    if args.engine == "parakeet_onnx_int8":
        if importlib.util.find_spec("sherpa_onnx") is None:
            logging.error("parakeet_onnx_int8 engine selected but sherpa-onnx is not available.")
            sys.exit(1)

        model_id = args.parakeet_onnx_model_id

        # Mimic the whisper/nemo flow: positional argument selects model.
        if args.model_path != "base":
            if not args.model_path.startswith("sherpa-onnx-"):
                logging.error(
                    "For --engine parakeet_onnx_int8, the positional model must be a sherpa-onnx model bundle id starting with 'sherpa-onnx-' (got: %s). "
                    "Either omit the positional argument to use the default, or use --parakeet-onnx-model-id, or use --parakeet-onnx-dir.",
                    args.model_path,
                )
                sys.exit(1)
            model_id = args.model_path

        # Persist selected bundle id so the daemon can auto-download when --parakeet-onnx-dir is omitted.
        args.parakeet_onnx_model_id = model_id

        if args.check_model:
            try:
                model_dir = args.parakeet_onnx_dir
                if not model_dir:
                    model_dir = str(
                        ensure_sherpa_onnx_parakeet_model_dir(
                            model_id=model_id,
                            cache_dir=args.parakeet_onnx_cache_dir,
                        )
                    )

            except Exception as e:
                logging.error("Failed to auto-download sherpa-onnx model bundle: %s", e)
                sys.exit(1)

            SherpaOnnxParakeetInt8Transcriber(
                model_dir=model_dir,
                provider=args.onnx_provider,
                num_threads=args.onnx_threads,
            )
            logging.info("Model loaded successfully")
            return

    # Handle device configuration
    config_manager = ConfigManager()

    # Validate configuration file
    if not config_manager.validate_config():
        logging.warning("Configuration file has issues, continuing with caution...")

    device_name = args.device
    if device_name:
        # Device provided via command line, save it to config
        config_manager.save_device(device_name)
    else:
        # No device provided, try to load from config
        device_name = config_manager.load_device()
        if not device_name:
            logging.error("No device specified and no saved configuration found.")
            logging.error(
                "Please run with --device option first, or use test_audio_devices.py to find a working device."
            )
            sys.exit(1)

    # Create and run daemon
    daemon = WhispypyDaemon(
        model_path=args.model_path,
        device_name=device_name,
        engine=args.engine,
        parakeet_onnx_dir=args.parakeet_onnx_dir,
        parakeet_onnx_model_id=args.parakeet_onnx_model_id,
        parakeet_onnx_cache_dir=args.parakeet_onnx_cache_dir,
        onnx_provider=args.onnx_provider,
        onnx_threads=args.onnx_threads,
        keep_audio=args.keep_audio,
        autopaste=args.autopaste,
    )

    daemon.run()


if __name__ == "__main__":
    try:
        main()
    except Exception as e:
        logging.error(f"Error: {e}")
        sys.exit(1)