VoiceAgent/push_to_talk.py

"""Keyboard-driven turn taking: nothing is heard until you say so.

Voice activity detection has to guess when you've finished a sentence, and it
guesses slowly — Silero waits 0.2s of silence, then the turn strategy waits
another 0.6s in case you say more. A key removes both the waiting and the
guessing: the turn ends the moment you say it does.

It also makes echo structurally impossible. Transcription only runs over audio
captured while the key is engaged, so the agent cannot hear its own voice
however loud the speakers are.

Two input modes:

- **hold** (default): a Quartz event tap watches a modifier key system-wide, so
  it works whatever app has focus, and reports releases as well as presses —
  real hold-to-talk. Needs Input Monitoring permission.
- **toggle**: reads stdin. Works with no permissions, but only while the
  terminal has focus, and a terminal never sees key releases — so it's press to
  start, press again to send.
"""

import asyncio
import atexit
import sys
import termios
import tty

from loguru import logger

from pipecat.frames.frames import (
    CancelFrame,
    EndFrame,
    Frame,
    StartFrame,
    UserStartedSpeakingFrame,
    UserStoppedSpeakingFrame,
    VADUserStartedSpeakingFrame,
    VADUserStoppedSpeakingFrame,
)
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor


class PushToTalk(FrameProcessor):
    """Turn a key into the microphone's on-air switch.

    Emits the VAD frames the segmented STT uses to cut audio into utterances,
    and the user-turn frames the external turn strategies use to open and close
    a turn — so one key drives both.

    Args:
        hold_key: Modifier to hold, from ``global_hotkey.HOLD_KEYS``. None
            selects the stdin toggle instead.
        toggle_key: Character that toggles talking in stdin mode.
    """

    def __init__(self, *, hold_key: str | None = "fn", toggle_key: str = " ", **kwargs):
        super().__init__(**kwargs)
        self._hold_key = hold_key
        self._toggle_key = toggle_key
        self._talking = False
        self._monitor = None
        self._fd: int | None = None
        self._saved_term: list | None = None

    async def process_frame(self, frame: Frame, direction: FrameDirection):
        await super().process_frame(frame, direction)

        if isinstance(frame, StartFrame):
            await self.push_frame(frame, direction)
            self._start_input()
        elif isinstance(frame, (EndFrame, CancelFrame)):
            self._stop_input()
            await self.push_frame(frame, direction)
        else:
            await self.push_frame(frame, direction)

    # ------------------------------------------------------------------ input

    def _start_input(self):
        holding = bool(self._hold_key) and self._start_hold_mode()
        # Always arm the toggle as well. Without Input Monitoring, macOS still
        # hands back a valid-looking event tap and simply never delivers events
        # to it — so hold mode can fail silently, and this guarantees there is
        # always some way to talk.
        self._start_toggle_mode(primary=not holding)

    def _start_hold_mode(self) -> bool:
        from global_hotkey import HoldKeyMonitor, permission_granted, request_permission

        if not permission_granted():
            logger.warning("Input Monitoring permission is not granted; requesting it.")
            request_permission()
            if not permission_granted():
                logger.warning(
                    "Still not granted. Add your terminal under System Settings > "
                    "Privacy & Security > Input Monitoring, then restart it. "
                    "Falling back to the spacebar toggle for now."
                )
                return False

        monitor = HoldKeyMonitor(
            key=self._hold_key,
            on_change=self._on_hold_change,
            loop=self.get_event_loop(),
        )
        if not monitor.start():
            logger.warning("Could not create the keyboard event tap; using the toggle instead.")
            return False

        self._monitor = monitor
        label = "🌐 fn" if self._hold_key == "fn" else self._hold_key
        logger.info(f"Hold {label} to talk — works in any app, release to send.")
        if self._hold_key == "fn":
            logger.info(
                "If fn also opens the emoji picker, set System Settings > Keyboard > "
                "'Press 🌐 to' → Do Nothing."
            )
        return True

    def _start_toggle_mode(self, *, primary: bool = True):
        if not sys.stdin.isatty():
            if primary:
                logger.warning("stdin is not a terminal; push-to-talk is disabled.")
            return

        self._fd = sys.stdin.fileno()
        self._saved_term = termios.tcgetattr(self._fd)
        # Restore the terminal even on an unhandled exception, or the shell is
        # left with echo off and no line editing.
        atexit.register(self._restore_terminal)
        tty.setcbreak(self._fd)
        self.get_event_loop().add_reader(self._fd, self._on_stdin_readable)

        label = "SPACE" if self._toggle_key == " " else repr(self._toggle_key)
        if primary:
            logger.info(f"Push to talk: press {label} to start speaking, {label} again to send.")
        else:
            logger.info(f"({label} also works as a toggle, if the hold key goes quiet.)")

    def _stop_input(self):
        if self._monitor:
            self._monitor.stop()
            self._monitor = None
        if self._fd is not None:
            try:
                self.get_event_loop().remove_reader(self._fd)
            except Exception:
                pass
            self._restore_terminal()

    def _restore_terminal(self):
        if self._fd is not None and self._saved_term is not None:
            termios.tcsetattr(self._fd, termios.TCSADRAIN, self._saved_term)
            self._saved_term = None

    def _on_hold_change(self, down: bool):
        self.create_task(self._start_talking() if down else self._stop_talking())

    def _on_stdin_readable(self):
        try:
            char = sys.stdin.read(1)
        except (OSError, ValueError):
            return
        if not char:
            return
        # Ctrl-C doesn't raise KeyboardInterrupt in cbreak mode; deliver it.
        if char == "\x03":
            self._restore_terminal()
            raise KeyboardInterrupt
        if char == self._toggle_key:
            self.create_task(self._toggle())

    # ------------------------------------------------------------------ turns

    async def _start_talking(self):
        if self._talking:
            return
        self._talking = True
        logger.info("🎤 listening")
        # ExternalUserTurnStartStrategy hardcodes enable_interruptions=False, on
        # the assumption that whatever drives it externally handles this. That's
        # us: without this, talking over Claude doesn't stop him.
        await self.broadcast_interruption()
        await self.push_frame(VADUserStartedSpeakingFrame())
        await self.push_frame(UserStartedSpeakingFrame())

    async def _stop_talking(self):
        if not self._talking:
            return
        self._talking = False
        logger.info("… sent")
        await self.push_frame(VADUserStoppedSpeakingFrame())
        await self.push_frame(UserStoppedSpeakingFrame())

    async def _toggle(self):
        if self._talking:
            await self._stop_talking()
        else:
            await self._start_talking()