active_call/media/
loader.rs

1use crate::media::cache;
2use anyhow::{Result, anyhow};
3use audio_codec::Resampler;
4use audio_codec::opus::OpusDecoder;
5use hound::WavReader;
6use ogg::reading::PacketReader;
7use reqwest::Client;
8use std::fs::File;
9use std::io::{BufReader, Seek, SeekFrom, Write};
10use std::time::Instant;
11use symphonia::core::codecs::audio::AudioDecoderOptions;
12use symphonia::core::codecs::CodecParameters;
13use symphonia::core::errors::Error as SymphoniaError;
14use symphonia::core::formats::probe::Hint;
15use symphonia::core::formats::FormatOptions;
16use symphonia::core::formats::TrackType;
17use symphonia::core::io::MediaSourceStream;
18use symphonia::core::meta::MetadataOptions;
19use symphonia::default::{get_codecs, get_probe};
20use tracing::{info, warn};
21use url::Url;
22
23pub async fn download_from_url(url: &str, use_cache: bool) -> Result<(File, Option<String>)> {
24    let cache_key = cache::generate_cache_key(url, 0, None, None);
25    if use_cache && cache::is_cached(&cache_key).await? {
26        match cache::get_cache_path(&cache_key) {
27            Ok(path) => return Ok((File::open(&path).map_err(|e| anyhow!(e))?, None)),
28            Err(e) => {
29                warn!("loader: Error getting cache path: {}", e);
30                return Err(e);
31            }
32        }
33    }
34
35    let start_time = Instant::now();
36    let client = Client::new();
37    let response = client.get(url).send().await?;
38    let content_type = response
39        .headers()
40        .get(reqwest::header::CONTENT_TYPE)
41        .and_then(|v| v.to_str().ok())
42        .map(|s| s.split(';').next().unwrap_or(s).trim().to_string());
43    let bytes = response.bytes().await?;
44    let data = bytes.to_vec();
45    let duration = start_time.elapsed();
46
47    info!(
48        "loader: Downloaded {} bytes in {:?} for {} (content-type: {:?})",
49        data.len(),
50        duration,
51        url,
52        content_type,
53    );
54
55    if use_cache {
56        cache::store_in_cache(&cache_key, &data).await?;
57        match cache::get_cache_path(&cache_key) {
58            Ok(path) => return Ok((File::open(path).map_err(|e| anyhow!(e))?, content_type)),
59            Err(e) => {
60                warn!("loader: Error getting cache path: {}", e);
61                return Err(e);
62            }
63        }
64    }
65
66    let mut temp_file = tempfile::tempfile()?;
67    temp_file.write_all(&data)?;
68    temp_file.seek(SeekFrom::Start(0))?;
69    Ok((temp_file, content_type))
70}
71
72fn is_ogg(extension: &str, mime_type: Option<&str>) -> bool {
73    matches!(extension, "ogg" | "opus")
74        || matches!(
75            mime_type,
76            Some("audio/ogg") | Some("audio/opus") | Some("application/ogg")
77        )
78}
79
80enum OggCodec {
81    Opus { channels: u16 },
82    Other,
83}
84
85fn detect_ogg_codec(file: &mut File) -> Result<OggCodec> {
86    let mut reader = PacketReader::new(BufReader::new(&mut *file));
87    let head = reader
88        .read_packet_expected()
89        .map_err(|e| anyhow!("loader: failed reading OGG header: {e}"))?;
90    let codec = if head.data.starts_with(b"OpusHead") {
91        let channels = if head.data.len() > 9 {
92            head.data[9] as u16
93        } else {
94            2
95        };
96        OggCodec::Opus { channels }
97    } else {
98        OggCodec::Other
99    };
100    file.seek(SeekFrom::Start(0))?;
101    Ok(codec)
102}
103
104fn decode_opus_ogg(file: File, channels: u16, target_sample_rate: u32) -> Result<Vec<i16>> {
105    let mut reader = PacketReader::new(BufReader::new(file));
106
107    // Consume OpusHead (already peeked, but file was seeked back)
108    let head = reader
109        .read_packet_expected()
110        .map_err(|e| anyhow!("loader: failed reading OGG header: {e}"))?;
111    let channels = if head.data.len() > 9 {
112        head.data[9] as u16
113    } else {
114        channels
115    };
116
117    // Skip OpusTags packet
118    reader
119        .read_packet_expected()
120        .map_err(|e| anyhow!("loader: failed reading OpusTags: {e}"))?;
121
122    // Opus always encodes at 48 kHz; decode there and resample afterwards
123    let mut decoder = OpusDecoder::new(48000, channels);
124    let mut all_samples: Vec<i16> = Vec::new();
125
126    loop {
127        let packet = match reader.read_packet() {
128            Ok(Some(p)) => p,
129            Ok(None) => break,
130            Err(e) => return Err(anyhow!("loader: failed reading OGG packet: {e}")),
131        };
132        let samples = audio_codec::Decoder::decode(&mut decoder, &packet.data);
133        all_samples.extend_from_slice(&samples);
134    }
135
136    if all_samples.is_empty() {
137        return Err(anyhow!(
138            "loader: no decodable audio samples found in Opus stream"
139        ));
140    }
141
142    info!(
143        "loader: decoded Opus stream at 48000 Hz, {} channel(s)",
144        channels
145    );
146
147    if target_sample_rate != 48000 {
148        let mut resampler = Resampler::new(48000, target_sample_rate as usize);
149        all_samples = resampler.resample(&all_samples);
150    }
151
152    Ok(all_samples)
153}
154
155pub fn decode_wav(file: File, target_sample_rate: u32) -> Result<Vec<i16>> {
156    let reader = BufReader::new(file);
157    let mut wav_reader = WavReader::new(reader)?;
158    let spec = wav_reader.spec();
159    let sample_rate = spec.sample_rate;
160    let is_stereo = spec.channels == 2;
161
162    info!(
163        "WAV file detected with sample rate: {} Hz, channels: {}, bits: {}",
164        sample_rate, spec.channels, spec.bits_per_sample
165    );
166
167    let mut all_samples = Vec::new();
168
169    // Read all samples based on format and bit depth
170    match spec.sample_format {
171        hound::SampleFormat::Int => match spec.bits_per_sample {
172            16 => {
173                for sample in wav_reader.samples::<i16>() {
174                    if let Ok(s) = sample {
175                        all_samples.push(s);
176                    } else {
177                        break;
178                    }
179                }
180            }
181            8 => {
182                for sample in wav_reader.samples::<i8>() {
183                    if let Ok(s) = sample {
184                        all_samples.push((s as i16) * 256); // Convert 8-bit to 16-bit
185                    } else {
186                        break;
187                    }
188                }
189            }
190            24 | 32 => {
191                for sample in wav_reader.samples::<i32>() {
192                    if let Ok(s) = sample {
193                        all_samples.push((s >> 16) as i16); // Convert 24/32-bit to 16-bit
194                    } else {
195                        break;
196                    }
197                }
198            }
199            _ => {
200                return Err(anyhow!(
201                    "Unsupported bits per sample: {}",
202                    spec.bits_per_sample
203                ));
204            }
205        },
206        hound::SampleFormat::Float => {
207            for sample in wav_reader.samples::<f32>() {
208                if let Ok(s) = sample {
209                    all_samples.push((s * 32767.0) as i16); // Convert float to 16-bit
210                } else {
211                    break;
212                }
213            }
214        }
215    }
216
217    // Convert stereo to mono if needed
218    if is_stereo {
219        let mono_samples = all_samples
220            .chunks(2)
221            .map(|chunk| {
222                if chunk.len() == 2 {
223                    ((chunk[0] as i32 + chunk[1] as i32) / 2) as i16
224                } else {
225                    chunk[0]
226                }
227            })
228            .collect();
229        all_samples = mono_samples;
230    }
231
232    if sample_rate != target_sample_rate && sample_rate > 0 {
233        let mut resampler = Resampler::new(sample_rate as usize, target_sample_rate as usize);
234        all_samples = resampler.resample(&all_samples);
235    }
236
237    Ok(all_samples)
238}
239
240pub fn decode_audio(
241    mut file: File,
242    extension: &str,
243    mime_type: Option<&str>,
244    target_sample_rate: u32,
245) -> Result<Vec<i16>> {
246    if matches!(extension, "wav")
247        || matches!(
248            mime_type,
249            Some("audio/wav") | Some("audio/wave") | Some("audio/x-wav")
250        )
251    {
252        return decode_wav(file, target_sample_rate);
253    }
254
255    if is_ogg(extension, mime_type) {
256        match detect_ogg_codec(&mut file)? {
257            OggCodec::Opus { channels } => {
258                return decode_opus_ogg(file, channels, target_sample_rate);
259            }
260            OggCodec::Other => {} // fall through to symphonia (e.g. Vorbis)
261        }
262    }
263
264    let mss = MediaSourceStream::new(Box::new(file), Default::default());
265    let mut hint = Hint::new();
266    if !extension.is_empty() {
267        hint.with_extension(extension);
268    }
269    if let Some(mime) = mime_type {
270        hint.mime_type(mime);
271    }
272
273    let mut format = get_probe().probe(
274        &hint,
275        mss,
276        FormatOptions::default(),
277        MetadataOptions::default(),
278    )?;
279    let (track_id, audio_params) = {
280        let track = format
281            .default_track(TrackType::Audio)
282            .ok_or_else(|| anyhow!("loader: no default audio track found"))?;
283        let codec_params = track.codec_params
284            .as_ref()
285            .ok_or_else(|| anyhow!("loader: no codec parameters"))?;
286        let params = match codec_params {
287            CodecParameters::Audio(params) => params.clone(),
288            _ => return Err(anyhow!("loader: expected audio codec")),
289        };
290        (track.id, params)
291    };
292
293    let mut decoder = get_codecs().make_audio_decoder(&audio_params, &AudioDecoderOptions::default())?;
294    let mut all_samples = Vec::new();
295    let mut sample_rate = audio_params.sample_rate.unwrap_or(0);
296
297    loop {
298        let packet = match format.next_packet() {
299            Ok(Some(packet)) => packet,
300            Ok(None) => break,
301            Err(SymphoniaError::IoError(_)) => break,
302            Err(SymphoniaError::ResetRequired) => continue,
303            Err(e) => return Err(anyhow!("loader: failed reading audio packet: {e}")),
304        };
305
306        if packet.track_id != track_id {
307            continue;
308        }
309
310        match decoder.decode(&packet) {
311            Ok(decoded) => {
312                if sample_rate == 0 {
313                    sample_rate = decoded.spec().rate();
314                    info!(
315                        "loader: detected {:?} with sample rate: {} Hz, channels: {}",
316                        audio_params.codec,
317                        sample_rate,
318                        decoded.spec().channels().count()
319                    );
320                }
321                let channels = decoded.spec().channels().count();
322
323                let mut interleaved = Vec::new();
324                decoded.copy_to_vec_interleaved(&mut interleaved);
325
326                if channels <= 1 {
327                    all_samples.extend_from_slice(&interleaved);
328                } else {
329                    for frame in interleaved.chunks(channels) {
330                        if frame.is_empty() {
331                            continue;
332                        }
333                        let sum: i32 = frame.iter().map(|s| *s as i32).sum();
334                        all_samples.push((sum / frame.len() as i32) as i16);
335                    }
336                }
337            }
338            Err(SymphoniaError::DecodeError(_)) => continue,
339            Err(SymphoniaError::IoError(_)) => break,
340            Err(SymphoniaError::ResetRequired) => continue,
341            Err(e) => return Err(anyhow!("loader: failed decoding audio packet: {e}")),
342        }
343    }
344
345    if all_samples.is_empty() {
346        return Err(anyhow!("loader: no decodable audio samples found"));
347    }
348
349    if sample_rate != target_sample_rate && sample_rate > 0 {
350        let mut resampler = Resampler::new(sample_rate as usize, target_sample_rate as usize);
351        all_samples = resampler.resample(&all_samples);
352    }
353
354    Ok(all_samples)
355}
356
357pub async fn load_audio_as_pcm(
358    path: &str,
359    target_sample_rate: u32,
360    use_cache: bool,
361) -> Result<Vec<i16>> {
362    let is_url = path.starts_with("http://") || path.starts_with("https://");
363
364    let (file, content_type) = if is_url {
365        download_from_url(path, use_cache).await?
366    } else {
367        (File::open(path).map_err(|e| anyhow!("loader: {} {}", path, e))?, None)
368    };
369
370    let extension = if is_url {
371        path.parse::<Url>()?.path().split('.').last().unwrap_or("").to_string()
372    } else {
373        path.split('.').last().unwrap_or("").to_string()
374    };
375
376    decode_audio(
377        file,
378        &extension,
379        content_type.as_deref(),
380        target_sample_rate,
381    )
382}
active_call/media/loader.rs

active_call/media/
loader.rs
