Pipecat quickstart

import os
from pathlib import Path

from loguru import logger
from pipecat.audio.filters.aic_filter import AICFilter
from pipecat.frames.frames import Frame, InputAudioRawFrame, OutputAudioRawFrame
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.worker import PipelineParams, PipelineWorker
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.runner.types import RunnerArguments
from pipecat.runner.utils import create_transport
from pipecat.transports.base_transport import TransportParams
from pipecat.workers.runner import WorkerRunner


class AudioFrameConverter(FrameProcessor):
    def __init__(self):
        super().__init__()
        self.audio_frames = 0
        self.audio_samples = 0

    async def process_frame(self, frame: Frame, direction: FrameDirection):
        await super().process_frame(frame, direction)
        if isinstance(frame, InputAudioRawFrame):
            if frame.sample_rate != 16000 or frame.num_channels != 1:
                raise ValueError("Expected 16 kHz mono PCM from the input transport")
            if len(frame.audio) % 2:
                raise ValueError("Expected complete signed 16-bit PCM samples")
            self.audio_frames += 1
            self.audio_samples += len(frame.audio) // 2
            if self.audio_frames == 1 or self.audio_frames % 100 == 0:
                logger.info(
                    "Loopback: frames={} samples={} rate={} channels={}",
                    self.audio_frames, self.audio_samples,
                    frame.sample_rate, frame.num_channels,
                )
            await self.push_frame(
                OutputAudioRawFrame(
                    audio=frame.audio,
                    sample_rate=frame.sample_rate,
                    num_channels=frame.num_channels,
                ),
                direction,
            )
        else:
            await self.push_frame(frame, direction)


async def bot(runner_args: RunnerArguments):
    key = os.environ.get("AIC_SDK_LICENSE", "").strip()
    if not key:
        raise RuntimeError("Set AIC_SDK_LICENSE in the terminal that starts bot.py")

    aic_filter = AICFilter(
        license_key=key,
        model_id="quail-vf-2.2-l-16khz",
        model_download_dir=Path("models"),
    )
    transport = await create_transport(
        runner_args,
        {
            "webrtc": lambda: TransportParams(
                audio_in_enabled=True,
                audio_out_enabled=True,
                audio_in_channels=1,
                audio_out_channels=1,
                audio_in_filter=aic_filter,
            ),
        },
    )
    pipeline = Pipeline([
        transport.input(),
        AudioFrameConverter(),
        transport.output(),
    ])
    worker = PipelineWorker(
        pipeline,
        params=PipelineParams(
            audio_in_sample_rate=16000,
            audio_out_sample_rate=16000,
        ),
    )
    runner = WorkerRunner(handle_sigint=runner_args.handle_sigint)
    await runner.add_workers(worker)

    @transport.event_handler("on_client_connected")
    async def on_connected(transport, client):
        logger.info("Client connected; check the AIC initialization log")

    @transport.event_handler("on_client_disconnected")
    async def on_disconnected(transport, client):
        await runner.cancel()

    await runner.run()


if __name__ == "__main__":
    from pipecat.runner.run import main
    main()