feat: support streaming livekit helper

cai

2026-06-28 e59b20f60fff034a7268febd3348b15fc767edc4

 src/audio.rs

@@ -1,7 +1,11 @@
use std::{fs, io::Cursor, path::Path};

use std::{
    fs,
    io::Cursor,
    path::{Path, PathBuf},
};


use anyhow::{Context, Result, anyhow};

use hound::{SampleFormat, WavReader};

use hound::{SampleFormat, WavReader, WavSpec, WavWriter};
use minimp3::{Decoder as Mp3Decoder, Error as Mp3Error, Frame as Mp3Frame};

use reqwest::Client;



@@ -29,17 +33,63 @@
    }

}



#[derive(Clone, Debug)]
pub struct LoadedAudio {
    pub frames: Vec<PcmFrame>,
    pub diagnostics: AudioDiagnostics,
}

#[derive(Clone, Debug)]
pub struct AudioDiagnostics {
    pub source_kind: &'static str,
    pub source_format: &'static str,
    pub source_bytes: usize,
    pub source_sample_rate_hz: u32,
    pub source_num_channels: u16,
    pub decoded_sample_count: usize,
    pub decoded_duration_ms: u64,
    pub target_sample_rate_hz: u32,
    pub target_num_channels: u16,
    pub target_sample_count: usize,
    pub target_duration_ms: u64,
    pub frame_count: usize,
    pub rms: f64,
    pub peak: f64,
    pub clipped_sample_count: usize,
    pub silence_ratio: f64,
    pub mp3_skipped_data_count: usize,
    pub mp3_insufficient_data_count: usize,
    pub debug_source_path: Option<String>,
    pub debug_pcm_wav_path: Option<String>,
}

struct DecodeResult {
    samples: Vec<i16>,
    diagnostics: AudioDiagnostics,
}

pub async fn load_pre_recorded_frames(

    http: &Client,

    audio_file: Option<&str>,

    audio_url: Option<&str>,

    target_sample_rate_hz: u32,

    target_num_channels: u16,

) -> Result<Option<Vec<PcmFrame>>> {

    debug_dump_dir: Option<&str>,
    call_id: &str,
    debug_label: &str,
) -> Result<Option<LoadedAudio>> {
    if let Some(path) = audio_file.filter(|value| !value.trim().is_empty()) {

        let audio_bytes = fs::read(Path::new(path))

            .with_context(|| format!("failed to read greeting audio file {path}"))?;

        return decode_audio_frames(&audio_bytes, target_sample_rate_hz, target_num_channels)

        return decode_audio_frames(
            &audio_bytes,
            "local_file",
            target_sample_rate_hz,
            target_num_channels,
            debug_dump_dir,
            call_id,
            debug_label,
        )
            .map(Some);

    }

    if let Some(url) = audio_url.filter(|value| !value.trim().is_empty()) {

@@ -58,51 +108,137 @@
            .bytes()

            .await

            .with_context(|| format!("failed to read greeting audio body {url}"))?;

        return decode_audio_frames(&bytes, target_sample_rate_hz, target_num_channels).map(Some);

        return decode_audio_frames(
            &bytes,
            "downloaded_url",
            target_sample_rate_hz,
            target_num_channels,
            debug_dump_dir,
            call_id,
            debug_label,
        )
        .map(Some);
    }

    Ok(None)

}



fn decode_audio_frames(

pub fn decode_audio_bytes_to_frames(
    audio_bytes: &[u8],

    source_kind: &'static str,
    target_sample_rate_hz: u32,

    target_num_channels: u16,

) -> Result<Vec<PcmFrame>> {

    if is_wav(audio_bytes) {

        return decode_wav_frames(audio_bytes, target_sample_rate_hz, target_num_channels);

    }

    if is_mp3(audio_bytes) {

        return decode_mp3_frames(audio_bytes, target_sample_rate_hz, target_num_channels);

    debug_dump_dir: Option<&str>,
    call_id: &str,
    debug_label: &str,
) -> Result<LoadedAudio> {
    decode_audio_frames(
        audio_bytes,
        source_kind,
        target_sample_rate_hz,
        target_num_channels,
        debug_dump_dir,
        call_id,
        debug_label,
    )
    }



    let wav_result = decode_wav_frames(audio_bytes, target_sample_rate_hz, target_num_channels);

fn decode_audio_frames(
    audio_bytes: &[u8],
    source_kind: &'static str,
    target_sample_rate_hz: u32,
    target_num_channels: u16,
    debug_dump_dir: Option<&str>,
    call_id: &str,
    debug_label: &str,
) -> Result<LoadedAudio> {
    if is_wav(audio_bytes) {
        return finalize_decode_result(
            audio_bytes,
            decode_wav_samples(
                audio_bytes,
                source_kind,
                target_sample_rate_hz,
                target_num_channels,
            )?,
            target_sample_rate_hz,
            target_num_channels,
            debug_dump_dir,
            call_id,
            debug_label,
        );
    }
    if is_mp3(audio_bytes) {
        return finalize_decode_result(
            audio_bytes,
            decode_mp3_samples(
                audio_bytes,
                source_kind,
                target_sample_rate_hz,
                target_num_channels,
            )?,
            target_sample_rate_hz,
            target_num_channels,
            debug_dump_dir,
            call_id,
            debug_label,
        );
    }

    let wav_result = decode_wav_samples(
        audio_bytes,
        source_kind,
        target_sample_rate_hz,
        target_num_channels,
    );
    if wav_result.is_ok() {

        return wav_result;

        return finalize_decode_result(
            audio_bytes,
            wav_result?,
            target_sample_rate_hz,
            target_num_channels,
            debug_dump_dir,
            call_id,
            debug_label,
        );
    }

    let wav_error = wav_result.err();

    decode_mp3_frames(audio_bytes, target_sample_rate_hz, target_num_channels).map_err(

        |mp3_error| {

            anyhow!(

    let mp3_result = decode_mp3_samples(
        audio_bytes,
        source_kind,
        target_sample_rate_hz,
        target_num_channels,
    );
    match mp3_result {
        Ok(result) => finalize_decode_result(
            audio_bytes,
            result,
            target_sample_rate_hz,
            target_num_channels,
            debug_dump_dir,
            call_id,
            debug_label,
        ),
        Err(mp3_error) => Err(anyhow!(
                "failed to decode greeting audio as wav or mp3: wav={}, mp3={}",

                wav_error

                    .map(|error| error.to_string())

                    .unwrap_or_else(|| "unknown".to_string()),

                mp3_error

            )

        },

    )

        )),
    }
}



fn decode_wav_frames(

fn decode_wav_samples(
    wav_bytes: &[u8],

    source_kind: &'static str,
    target_sample_rate_hz: u32,

    target_num_channels: u16,

) -> Result<Vec<PcmFrame>> {

) -> Result<DecodeResult> {
    let cursor = Cursor::new(wav_bytes.to_vec());

    let mut reader = WavReader::new(cursor).context("failed to open greeting wav bytes")?;

    let spec = reader.spec();

    let src_channels = spec.channels.max(1);

    let mut samples = match (spec.sample_format, spec.bits_per_sample) {

    let decoded_samples = match (spec.sample_format, spec.bits_per_sample) {
        (SampleFormat::Int, 16) => reader

            .samples::<i16>()

            .collect::<std::result::Result<Vec<_>, _>>()

@@ -121,7 +257,8 @@
        }

    };



    samples = remap_channels(samples, src_channels, target_num_channels);

    let decoded_sample_count = decoded_samples.len();
    let mut samples = remap_channels(decoded_samples, src_channels, target_num_channels);
    samples = resample_linear(

        samples,

        spec.sample_rate,

@@ -129,21 +266,38 @@
        target_num_channels,

    );



    Ok(chunk_pcm_samples(

        samples,

    Ok(DecodeResult {
        diagnostics: build_diagnostics(
            source_kind,
            "wav",
            wav_bytes.len(),
            spec.sample_rate,
            src_channels,
            decoded_sample_count,
            &samples,
        target_sample_rate_hz,

        target_num_channels,

    ))

            0,
            0,
        ),
        samples,
    })
}



fn decode_mp3_frames(

fn decode_mp3_samples(
    mp3_bytes: &[u8],

    source_kind: &'static str,
    target_sample_rate_hz: u32,

    target_num_channels: u16,

) -> Result<Vec<PcmFrame>> {

) -> Result<DecodeResult> {
    let cursor = Cursor::new(mp3_bytes.to_vec());

    let mut decoder = Mp3Decoder::new(cursor);

    let mut samples = Vec::new();

    let mut decoded_sample_count = 0usize;
    let mut source_sample_rate_hz = 0u32;
    let mut source_num_channels = 0u16;
    let mut skipped_data_count = 0usize;
    let mut insufficient_data_count = 0usize;


    loop {

        match decoder.next_frame() {

@@ -157,6 +311,13 @@
                    .map_err(|_| anyhow!("unsupported mp3 sample rate {sample_rate}"))?;

                let src_channels = u16::try_from(channels)

                    .map_err(|_| anyhow!("unsupported mp3 channel count {channels}"))?;

                if source_sample_rate_hz == 0 {
                    source_sample_rate_hz = src_rate;
                }
                if source_num_channels == 0 {
                    source_num_channels = src_channels;
                }
                decoded_sample_count += data.len();
                let mut frame_samples = remap_channels(data, src_channels, target_num_channels);

                frame_samples = resample_linear(

                    frame_samples,

@@ -167,7 +328,14 @@
                samples.extend(frame_samples);

            }

            Err(Mp3Error::Eof) => break,

            Err(Mp3Error::SkippedData) | Err(Mp3Error::InsufficientData) => continue,

            Err(Mp3Error::SkippedData) => {
                skipped_data_count += 1;
                continue;
            }
            Err(Mp3Error::InsufficientData) => {
                insufficient_data_count += 1;
                continue;
            }
            Err(error) => return Err(anyhow!("failed to decode mp3 frame: {error:?}")),

        }

    }

@@ -176,11 +344,232 @@
        return Err(anyhow!("decoded mp3 contains no audio samples"));

    }



    Ok(chunk_pcm_samples(

        samples,

    Ok(DecodeResult {
        diagnostics: build_diagnostics(
            source_kind,
            "mp3",
            mp3_bytes.len(),
            source_sample_rate_hz,
            source_num_channels.max(1),
            decoded_sample_count,
            &samples,
        target_sample_rate_hz,

        target_num_channels,

            skipped_data_count,
            insufficient_data_count,
        ),
        samples,
    })
}

fn finalize_decode_result(
    audio_bytes: &[u8],
    mut decode_result: DecodeResult,
    target_sample_rate_hz: u32,
    target_num_channels: u16,
    debug_dump_dir: Option<&str>,
    call_id: &str,
    debug_label: &str,
) -> Result<LoadedAudio> {
    if let Some(dir) = debug_dump_dir.filter(|value| !value.trim().is_empty()) {
        let paths = dump_debug_audio(
            dir,
            call_id,
            debug_label,
            decode_result.diagnostics.source_format,
            audio_bytes,
            &decode_result.samples,
            target_sample_rate_hz,
            target_num_channels,
        )?;
        decode_result.diagnostics.debug_source_path = paths.0;
        decode_result.diagnostics.debug_pcm_wav_path = paths.1;
    }

    let frames = chunk_pcm_samples(
        decode_result.samples,
        target_sample_rate_hz,
        target_num_channels,
    );
    decode_result.diagnostics.frame_count = frames.len();

    Ok(LoadedAudio {
        frames,
        diagnostics: decode_result.diagnostics,
    })
}

fn build_diagnostics(
    source_kind: &'static str,
    source_format: &'static str,
    source_bytes: usize,
    source_sample_rate_hz: u32,
    source_num_channels: u16,
    decoded_sample_count: usize,
    target_samples: &[i16],
    target_sample_rate_hz: u32,
    target_num_channels: u16,
    mp3_skipped_data_count: usize,
    mp3_insufficient_data_count: usize,
) -> AudioDiagnostics {
    let target_sample_count = target_samples.len();
    let decoded_duration_ms = duration_ms(
        decoded_sample_count,
        source_sample_rate_hz,
        source_num_channels,
    );
    let target_duration_ms = duration_ms(
        target_sample_count,
        target_sample_rate_hz,
        target_num_channels,
    );
    let (rms, peak, clipped_sample_count, silence_ratio) = quality_stats(target_samples);
    AudioDiagnostics {
        source_kind,
        source_format,
        source_bytes,
        source_sample_rate_hz,
        source_num_channels,
        decoded_sample_count,
        decoded_duration_ms,
        target_sample_rate_hz,
        target_num_channels,
        target_sample_count,
        target_duration_ms,
        frame_count: 0,
        rms,
        peak,
        clipped_sample_count,
        silence_ratio,
        mp3_skipped_data_count,
        mp3_insufficient_data_count,
        debug_source_path: None,
        debug_pcm_wav_path: None,
    }
}

fn duration_ms(sample_count: usize, sample_rate_hz: u32, channels: u16) -> u64 {
    if sample_count == 0 || sample_rate_hz == 0 || channels == 0 {
        return 0;
    }
    let frames = sample_count as f64 / f64::from(channels.max(1));
    ((frames / f64::from(sample_rate_hz)) * 1000.0).round() as u64
}

fn quality_stats(samples: &[i16]) -> (f64, f64, usize, f64) {
    if samples.is_empty() {
        return (0.0, 0.0, 0, 0.0);
    }
    const SILENCE_THRESHOLD: i32 = 256;
    let mut sum_squares = 0f64;
    let mut peak = 0i32;
    let mut clipped = 0usize;
    let mut silence = 0usize;
    for &sample in samples {
        let abs = i32::from(sample).abs();
        peak = peak.max(abs);
        if abs >= i32::from(i16::MAX) {
            clipped += 1;
        }
        if abs <= SILENCE_THRESHOLD {
            silence += 1;
        }
        let normalized = f64::from(sample) / f64::from(i16::MAX);
        sum_squares += normalized * normalized;
    }
    let rms = (sum_squares / samples.len() as f64).sqrt();
    let peak = f64::from(peak) / f64::from(i16::MAX);
    let silence_ratio = silence as f64 / samples.len() as f64;
    (round4(rms), round4(peak), clipped, round4(silence_ratio))
}

fn round4(value: f64) -> f64 {
    (value * 10_000.0).round() / 10_000.0
}

fn dump_debug_audio(
    dir: &str,
    call_id: &str,
    debug_label: &str,
    source_format: &str,
    source_bytes: &[u8],
    target_samples: &[i16],
    target_sample_rate_hz: u32,
    target_num_channels: u16,
) -> Result<(Option<String>, Option<String>)> {
    let base_dir = PathBuf::from(dir);
    fs::create_dir_all(&base_dir)
        .with_context(|| format!("failed to create audio debug dump dir {dir}"))?;
    let safe_call_id = sanitize_file_segment(call_id);
    let safe_label = sanitize_file_segment(debug_label);
    let source_path = base_dir.join(format!(
        "{safe_call_id}-{safe_label}-source.{source_format}"
    ));
    fs::write(&source_path, source_bytes).with_context(|| {
        format!(
            "failed to write audio debug source {}",
            source_path.display()
        )
    })?;

    let pcm_path = base_dir.join(format!("{safe_call_id}-{safe_label}-target.wav"));
    write_debug_wav(
        &pcm_path,
        target_samples,
        target_sample_rate_hz,
        target_num_channels,
    )?;

    Ok((
        Some(source_path.to_string_lossy().to_string()),
        Some(pcm_path.to_string_lossy().to_string()),
    ))

}

fn write_debug_wav(path: &Path, samples: &[i16], sample_rate_hz: u32, channels: u16) -> Result<()> {
    let spec = WavSpec {
        channels,
        sample_rate: sample_rate_hz,
        bits_per_sample: 16,
        sample_format: SampleFormat::Int,
    };
    let mut writer = WavWriter::create(path, spec)
        .with_context(|| format!("failed to create audio debug wav {}", path.display()))?;
    for &sample in samples {
        writer
            .write_sample(sample)
            .with_context(|| format!("failed to write audio debug wav {}", path.display()))?;
    }
    writer
        .finalize()
        .with_context(|| format!("failed to finalize audio debug wav {}", path.display()))
}

pub fn write_pcm_wav(
    path: &Path,
    samples: &[i16],
    sample_rate_hz: u32,
    channels: u16,
) -> Result<()> {
    write_debug_wav(path, samples, sample_rate_hz, channels)
}

fn sanitize_file_segment(value: &str) -> String {
    let sanitized: String = value
        .chars()
        .map(|ch| {
            if ch.is_ascii_alphanumeric() || ch == '-' || ch == '_' || ch == '.' {
                ch
            } else {
                '_'
            }
        })
        .collect();
    if sanitized.is_empty() {
        "unknown".to_string()
    } else {
        sanitized
    }
}



fn chunk_pcm_samples(

@@ -305,13 +694,27 @@
        writer.finalize().expect("finalize wav");



        let http = reqwest::Client::new();

        let frames = load_pre_recorded_frames(&http, path.to_str(), None, 48_000, 1)

        let loaded = load_pre_recorded_frames(
            &http,
            path.to_str(),
            None,
            48_000,
            1,
            Some(dir.to_string_lossy().as_ref()),
            "test-call",
            "greeting",
        )
            .await

            .expect("load frames")

            .expect("frames");



        assert!(!frames.is_empty());

        assert_eq!(frames[0].sample_rate, 48_000);

        assert_eq!(frames[0].num_channels, 1);

        assert!(!loaded.frames.is_empty());
        assert_eq!(loaded.frames[0].sample_rate, 48_000);
        assert_eq!(loaded.frames[0].num_channels, 1);
        assert_eq!(loaded.diagnostics.source_format, "wav");
        assert_eq!(loaded.diagnostics.source_sample_rate_hz, 16_000);
        assert_eq!(loaded.diagnostics.target_sample_rate_hz, 48_000);
        assert!(loaded.diagnostics.debug_source_path.is_some());
        assert!(loaded.diagnostics.debug_pcm_wav_path.is_some());
    }

}