VoiceAgent/global_hotkey.py

"""Watch a modifier key system-wide, so push-to-talk works from any app.

Reading stdin only works while the terminal has focus, which defeats the point
of a voice assistant you talk to while doing something else. A Quartz event tap
sees key events no matter what is frontmost, and — unlike stdin — reports
releases as well as presses, so this can be true hold-to-talk.

The tap runs on its own thread with its own CFRunLoop and hands state changes
back to asyncio via ``call_soon_threadsafe``.

Requires Input Monitoring permission for whichever app runs this (your
terminal), granted under System Settings > Privacy & Security > Input
Monitoring.
"""

import asyncio
import threading
from collections.abc import Callable

from loguru import logger

import Quartz
from CoreFoundation import (
    CFMachPortCreateRunLoopSource,
    CFRunLoopAddSource,
    CFRunLoopGetCurrent,
    CFRunLoopRun,
    CFRunLoopStop,
    kCFRunLoopCommonModes,
)


def disable_app_nap(reason: str = "VoiceAgent Hold-to-Talk Event Tap"):
    """Prevent macOS App Nap from throttling thread execution and timing out event taps."""
    try:
        from Foundation import (
            NSActivityIdleSystemSleepDisabled,
            NSActivityLatencyCritical,
            NSActivityUserInitiated,
            NSProcessInfo,
        )

        options = (
            NSActivityUserInitiated
            | NSActivityIdleSystemSleepDisabled
            | NSActivityLatencyCritical
        )
        token = NSProcessInfo.processInfo().beginActivityWithOptions_reason_(
            options, reason
        )
        logger.info("Disabled macOS App Nap for low-latency hotkey monitoring.")
        return token
    except Exception as e:
        logger.warning(f"Could not disable App Nap via Foundation: {e}")
        return None


def enable_app_nap(token):
    """Restore normal App Nap power management for the given activity token."""
    if token is None:
        return
    try:
        from Foundation import NSProcessInfo

        NSProcessInfo.processInfo().endActivity_(token)
    except Exception as e:
        logger.debug(f"Error ending App Nap activity: {e}")


# (keycode, modifier mask) for the keys worth holding. Modifier keycodes arrive
# on flagsChanged events, so one handler covers all of them.
HOLD_KEYS: dict[str, tuple[int, int]] = {
    "fn": (63, Quartz.kCGEventFlagMaskSecondaryFn),
    "right-option": (61, Quartz.kCGEventFlagMaskAlternate),
    "right-command": (54, Quartz.kCGEventFlagMaskCommand),
    "right-control": (62, Quartz.kCGEventFlagMaskControl),
    "right-shift": (60, Quartz.kCGEventFlagMaskShift),
}


def permission_granted() -> bool:
    """Whether this process may observe keyboard events."""
    return bool(Quartz.CGPreflightListenEventAccess())


def request_permission() -> bool:
    """Ask for Input Monitoring, which prompts once and then opens Settings."""
    return bool(Quartz.CGRequestListenEventAccess())


class HoldKeyMonitor:
    """Report press and release of one modifier key, from anywhere in the OS.

    Args:
        key: A name from ``HOLD_KEYS``.
        on_change: Called with True on press and False on release, on the
            asyncio loop.
        loop: The loop to deliver callbacks on.
    """

    def __init__(self, *, key: str, on_change: Callable[[bool], None], loop):
        if key not in HOLD_KEYS:
            raise ValueError(f"Unsupported hold key {key!r}. Choose from {list(HOLD_KEYS)}.")
        self._key = key
        self._keycode, self._mask = HOLD_KEYS[key]
        self._on_change = on_change
        self._loop = loop
        self._thread: threading.Thread | None = None
        self._runloop = None
        self._tap = None
        self._down = False
        self._ready = threading.Event()
        self._started_ok = False
        self._activity_token = None

    def start(self) -> bool:
        """Begin watching. Returns False if the tap could not be created."""
        self._activity_token = disable_app_nap("VoiceAgent HoldKeyMonitor")
        self._thread = threading.Thread(target=self._run, name="hold-key-tap", daemon=True)
        self._thread.start()
        self._ready.wait(timeout=5)
        return self._started_ok

    def stop(self):
        if self._activity_token is not None:
            enable_app_nap(self._activity_token)
            self._activity_token = None
        if self._runloop is not None:
            CFRunLoopStop(self._runloop)
            self._runloop = None

    def _run(self):
        tap = Quartz.CGEventTapCreate(
            Quartz.kCGSessionEventTap,
            Quartz.kCGHeadInsertEventTap,
            # Listen only: we observe the key without swallowing it, so we never
            # break whatever else the user has bound to it.
            Quartz.kCGEventTapOptionListenOnly,
            Quartz.CGEventMaskBit(Quartz.kCGEventFlagsChanged),
            self._callback,
            None,
        )
        if tap is None:
            self._started_ok = False
            self._ready.set()
            return

        self._tap = tap
        source = CFMachPortCreateRunLoopSource(None, tap, 0)
        self._runloop = CFRunLoopGetCurrent()
        CFRunLoopAddSource(self._runloop, source, kCFRunLoopCommonModes)
        Quartz.CGEventTapEnable(tap, True)

        self._started_ok = True
        self._ready.set()
        CFRunLoopRun()

    def _callback(self, proxy, event_type, event, refcon):
        # macOS disables a tap that takes too long; turn it back on.
        if event_type in (
            Quartz.kCGEventTapDisabledByTimeout,
            Quartz.kCGEventTapDisabledByUserInput,
        ):
            logger.warning(
                f"Quartz event tap disabled by OS (type={event_type}); re-enabling tap."
            )
            if self._tap is not None:
                Quartz.CGEventTapEnable(self._tap, True)
            return event

        try:
            keycode = Quartz.CGEventGetIntegerValueField(
                event, Quartz.kCGKeyboardEventKeycode
            )
            if keycode == self._keycode:
                down = bool(Quartz.CGEventGetFlags(event) & self._mask)
                if down != self._down:
                    self._down = down
                    self._loop.call_soon_threadsafe(self._on_change, down)
        except Exception as e:  # never let an exception cross back into C
            logger.debug(f"Hold-key tap callback error: {e}")
        return event