audio_debug.py

"""Audio debug monitor — prints live volume bars and diagnostics."""

import struct
import time

from loguru import logger

from pipecat.frames.frames import (
    InputAudioRawFrame,
    OutputAudioRawFrame,
    StartFrame,
    EndFrame,
)
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor


def rms_from_bytes(audio_bytes: bytes, sample_width: int = 2) -> float:
    """Compute RMS volume from raw PCM bytes (16-bit signed LE)."""
    if not audio_bytes or len(audio_bytes) < sample_width:
        return 0.0
    n_samples = len(audio_bytes) // sample_width
    fmt = f"<{n_samples}h"
    try:
        samples = struct.unpack(fmt, audio_bytes[: n_samples * sample_width])
    except struct.error:
        return 0.0
    if not samples:
        return 0.0
    sq_sum = sum(s * s for s in samples)
    return (sq_sum / n_samples) ** 0.5


def volume_bar(rms: float, max_rms: float = 8000.0, width: int = 40) -> str:
    """Render a volume bar like Google Meet."""
    level = min(rms / max_rms, 1.0)
    filled = int(level * width)
    bar = "█" * filled + "░" * (width - filled)
    db = 20 * (max(rms, 1) / 32768)  # rough dB-ish scale
    return f"|{bar}| {rms:6.0f} rms"


class AudioInputMonitor(FrameProcessor):
    """Monitors audio input frames and prints live volume."""

    def __init__(self, interval: float = 0.5, **kwargs):
        super().__init__(**kwargs)
        self._interval = interval
        self._last_print = 0.0
        self._frame_count = 0
        self._total_bytes = 0
        self._peak_rms = 0.0
        self._silent_count = 0
        self._started = False

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

        if isinstance(frame, StartFrame):
            self._started = True
            logger.debug(
                "🎤 Audio input started — sample_rate={} channels={}",
                getattr(frame, "audio_in_sample_rate", "?"),
                "1",
            )
            await self.push_frame(frame, direction)
            return

        if isinstance(frame, InputAudioRawFrame):
            self._frame_count += 1
            self._total_bytes += len(frame.audio)
            rms = rms_from_bytes(frame.audio)

            if rms > self._peak_rms:
                self._peak_rms = rms

            if rms < 50:
                self._silent_count += 1

            now = time.monotonic()
            if now - self._last_print >= self._interval:
                bar = volume_bar(rms)
                silence_pct = (
                    (self._silent_count / self._frame_count * 100)
                    if self._frame_count
                    else 0
                )
                logger.debug(
                    "🎤 IN  {}  peak={:6.0f}  frames={}  silence={:.0f}%",
                    bar,
                    self._peak_rms,
                    self._frame_count,
                    silence_pct,
                )
                self._last_print = now

            # Log detailed every 200 frames
            if self._frame_count % 200 == 0:
                logger.debug(
                    "Audio IN: frame={} rms={:.0f} peak={:.0f} bytes={} silence={:.0f}%",
                    self._frame_count,
                    rms,
                    self._peak_rms,
                    self._total_bytes,
                    (self._silent_count / self._frame_count * 100)
                    if self._frame_count
                    else 0,
                )

        await self.push_frame(frame, direction)


class AudioOutputMonitor(FrameProcessor):
    """Monitors audio output frames and prints live volume."""

    def __init__(self, interval: float = 0.5, **kwargs):
        super().__init__(**kwargs)
        self._interval = interval
        self._last_print = 0.0
        self._frame_count = 0
        self._total_bytes = 0

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

        if isinstance(frame, OutputAudioRawFrame):
            self._frame_count += 1
            self._total_bytes += len(frame.audio)
            rms = rms_from_bytes(frame.audio)

            now = time.monotonic()
            if now - self._last_print >= self._interval:
                bar = volume_bar(rms)
                logger.debug(
                    "🔊 OUT {}  frames={}  bytes={}",
                    bar,
                    self._frame_count,
                    self._total_bytes,
                )
                self._last_print = now

        await self.push_frame(frame, direction)


def list_audio_devices():
    """No-op in WebRTC mode — audio is handled by the browser."""
    logger.debug("Audio devices managed by browser (WebRTC mode)")