full_duplex.py

"""
Full-duplex audio using PlatformAudio.

This example demonstrates:
- Using PlatformAudio for microphone capture with built-in voice processing (AEC, NS, AGC)
- Automatic speaker playout for received audio (handled by PlatformAudio)
- Monitoring microphone dB levels via AudioStream
- Device enumeration and selection

With PlatformAudio:
- Microphone audio is captured with AEC, NS, and AGC applied automatically
- Received audio from remote participants is played through speakers automatically
- No manual audio routing needed

Usage:
    python full_duplex.py
    python full_duplex.py --list-devices
    python full_duplex.py --mic-id "mic-guid" --speaker-id "speaker-guid"
"""

import argparse
import asyncio
import logging
import os
import queue
import threading

try:
    from dotenv import find_dotenv, load_dotenv

    HAS_DOTENV = True
except ImportError:
    HAS_DOTENV = False

from livekit import api, rtc

from db_meter import calculate_db_level, display_single_db_meter


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Full-duplex audio using PlatformAudio")
    parser.add_argument(
        "--list-devices",
        action="store_true",
        help="List available audio devices and exit",
    )
    parser.add_argument(
        "--mic-id",
        type=str,
        help="Select microphone by device ID (from --list-devices)",
    )
    parser.add_argument(
        "--speaker-id",
        type=str,
        help="Select speaker by device ID (from --list-devices)",
    )
    return parser.parse_args()


def list_audio_devices() -> None:
    """List available audio devices using PlatformAudio."""
    try:
        platform_audio = rtc.PlatformAudio()
    except rtc.PlatformAudioError as e:
        print(f"Failed to initialize PlatformAudio: {e}")
        return

    try:
        print("\nRecording devices (microphones):")
        for device in platform_audio.recording_devices():
            print(f"  [{device.index}] {device.name}")
            print(f"      ID: {device.id}")

        print("\nPlayout devices (speakers):")
        for device in platform_audio.playout_devices():
            print(f"  [{device.index}] {device.name}")
            print(f"      ID: {device.id}")

        print()
    finally:
        platform_audio.close()


async def main(args: argparse.Namespace) -> None:
    logging.basicConfig(level=logging.INFO)

    # Load environment variables from a .env file if present
    if HAS_DOTENV:
        load_dotenv(find_dotenv())

    url = os.getenv("LIVEKIT_URL")
    api_key = os.getenv("LIVEKIT_API_KEY")
    api_secret = os.getenv("LIVEKIT_API_SECRET")
    token = os.getenv("LIVEKIT_TOKEN")
    # check for either token or api_key and api_secret
    if not url or (not token and (not api_key or not api_secret)):
        raise RuntimeError(
            "LIVEKIT_TOKEN or LIVEKIT_API_KEY and LIVEKIT_API_SECRET must be set in env"
        )

    # Initialize PlatformAudio for microphone capture and speaker playout
    # PlatformAudio provides:
    # - Built-in AEC (echo cancellation), NS (noise suppression), AGC (auto gain control)
    # - Automatic speaker playout for received audio
    try:
        platform_audio = rtc.PlatformAudio()
        logging.info("PlatformAudio initialized")
    except rtc.PlatformAudioError as e:
        logging.error(f"Failed to initialize PlatformAudio: {e}")
        return

    # Select microphone if specified
    if args.mic_id:
        try:
            platform_audio.set_recording_device(args.mic_id)
            logging.info(f"Selected microphone: {args.mic_id}")
        except rtc.PlatformAudioError as e:
            logging.warning(f"Failed to select microphone: {e}")

    # Select speaker if specified
    if args.speaker_id:
        try:
            platform_audio.set_playout_device(args.speaker_id)
            logging.info(f"Selected speaker: {args.speaker_id}")
        except rtc.PlatformAudioError as e:
            logging.warning(f"Failed to select speaker: {e}")

    room = rtc.Room()
    source = None

    # dB level monitoring (mic only)
    mic_db_queue: queue.Queue[float] = queue.Queue()

    # With PlatformAudio, received audio is automatically played through speakers
    # We just log when tracks are subscribed/unsubscribed
    def on_track_subscribed(
        track: rtc.Track,
        publication: rtc.RemoteTrackPublication,
        participant: rtc.RemoteParticipant,
    ):
        if track.kind == rtc.TrackKind.KIND_AUDIO:
            logging.info(
                "Subscribed to audio from %s (auto-playing through speaker)",
                participant.identity,
            )

    room.on("track_subscribed", on_track_subscribed)

    def on_track_unsubscribed(
        track: rtc.Track,
        publication: rtc.RemoteTrackPublication,
        participant: rtc.RemoteParticipant,
    ):
        if track.kind == rtc.TrackKind.KIND_AUDIO:
            logging.info("Unsubscribed from audio of %s", participant.identity)

    room.on("track_unsubscribed", on_track_unsubscribed)

    # generate token if not provided
    if not token:
        token = (
            api.AccessToken(api_key, api_secret)
            .with_identity("local-audio")
            .with_name("Local Audio")
            .with_grants(
                api.VideoGrants(
                    room_join=True,
                    room="local-audio",
                )
            )
            .to_jwt()
        )

    try:
        await room.connect(url, token)
        logging.info("connected to room %s", room.name)

        # Create audio source with voice processing enabled
        source = platform_audio.create_audio_source(
            rtc.PlatformAudioOptions(
                echo_cancellation=True,
                noise_suppression=True,
                auto_gain_control=True,
            )
        )
        track = rtc.LocalAudioTrack.create_audio_track("mic", source)

        pub_opts = rtc.TrackPublishOptions()
        pub_opts.source = rtc.TrackSource.SOURCE_MICROPHONE
        await room.local_participant.publish_track(track, pub_opts)
        logging.info("published local microphone with PlatformAudio")

        # Start dB meter display in a separate thread
        meter_thread = threading.Thread(
            target=display_single_db_meter,
            args=(mic_db_queue,),
            kwargs={"label": "Mic Level: "},
            daemon=True,
        )
        meter_thread.start()

        # Monitor microphone dB levels via AudioStream
        async def monitor_mic_db():
            mic_stream = rtc.AudioStream(track, sample_rate=48000, num_channels=1)
            frame_count = 0
            sample_interval = 5  # Process every 5th frame to reduce load

            try:
                async for frame_event in mic_stream:
                    # Skip frames to reduce processing load
                    frame_count += 1
                    if frame_count % sample_interval != 0:
                        continue

                    frame = frame_event.frame
                    # Convert frame data to list of samples
                    samples = list(frame.data)
                    db_level = calculate_db_level(samples)
                    # Update queue with latest value (non-blocking)
                    try:
                        mic_db_queue.put_nowait(db_level)
                    except queue.Full:
                        pass  # Drop if queue is full
            except asyncio.CancelledError:
                pass
            except Exception:
                pass
            finally:
                await mic_stream.aclose()

        asyncio.create_task(monitor_mic_db())

        # Run until Ctrl+C
        while True:
            await asyncio.sleep(1)
    except (KeyboardInterrupt, asyncio.CancelledError):
        pass
    finally:
        try:
            await room.disconnect()
        except Exception:
            pass
        # Clean up PlatformAudio resources
        if source is not None:
            source.close()
        platform_audio.close()


if __name__ == "__main__":
    args = parse_args()

    if args.list_devices:
        list_audio_devices()
    else:
        asyncio.run(main(args))