1use std::collections::VecDeque;
20
21use rff_codec::{Codec, CodecParams, CodecRegistry, Decoder, Encoder};
22use rff_core::{AudioFrame, Dictionary, Error, Frame, MediaType, Packet, Result, SampleFormat};
23use rusty_opus::parallel::{encode_parallel, ParallelConfig};
24use rusty_opus::{Application, OpusDecoder, OpusEncoder};
25
26const FRAME_MS: usize = 20;
28const RATES: [u32; 5] = [8000, 12000, 16000, 24000, 48000];
30const MAX_PACKET_MS: usize = 120;
32
33pub fn register(registry: &mut CodecRegistry) {
35 registry.register(Codec {
36 id: rff_core::CodecId::Opus,
37 name: "opus",
38 long_name: "Opus (Opus Interactive Audio Codec)",
39 media_type: MediaType::Audio,
40 decoder: Some(|| Box::new(OpusDec::default())),
41 encoder: Some(|| Box::new(OpusEnc::default())),
42 });
43}
44
45fn check_rate(rate: u32) -> Result<()> {
46 if RATES.contains(&rate) {
47 Ok(())
48 } else {
49 Err(Error::unsupported(format!(
50 "opus: sample rate {rate} (must be 8000/12000/16000/24000/48000)"
51 )))
52 }
53}
54
55fn frame_to_f32(af: &AudioFrame) -> Result<Vec<f32>> {
57 match af.format {
58 SampleFormat::F32 => Ok(af.planes[0]
59 .chunks_exact(4)
60 .map(|b| f32::from_le_bytes([b[0], b[1], b[2], b[3]]))
61 .collect()),
62 SampleFormat::S16 => Ok(af.planes[0]
63 .chunks_exact(2)
64 .map(|b| i16::from_le_bytes([b[0], b[1]]) as f32 / 32768.0)
65 .collect()),
66 other => Err(Error::unsupported(format!(
67 "opus encode: sample format `{}` (need interleaved s16/f32)",
68 other.name()
69 ))),
70 }
71}
72
73#[derive(Debug, Clone, PartialEq)]
80struct OpusHead {
81 channels: u16,
82 pre_skip: u16,
84 gain_q8: i16,
86 mapping_family: u8,
87}
88
89impl OpusHead {
90 fn parse(data: &[u8]) -> Option<OpusHead> {
92 if data.len() < 19 || &data[0..8] != b"OpusHead" {
93 return None;
94 }
95 Some(OpusHead {
96 channels: data[9] as u16,
97 pre_skip: u16::from_le_bytes([data[10], data[11]]),
98 gain_q8: i16::from_le_bytes([data[16], data[17]]),
99 mapping_family: data[18],
100 })
101 }
102}
103
104#[derive(Default)]
105struct OpusDec {
106 dec: Option<OpusDecoder>,
107 sample_rate: u32,
108 channels: u16,
109 gain: f32,
111 skip: usize,
113 position: i64,
117 queue: VecDeque<Frame>,
118 eof: bool,
119}
120
121fn keep_range(skip: &mut usize, position: i64, packet: &Packet, decoded: usize) -> (usize, usize) {
128 let pad = match packet.end_position {
129 Some(end) if packet.flags.end_of_stream => position
130 .saturating_add(decoded as i64)
131 .saturating_sub(end)
132 .clamp(0, decoded as i64) as usize,
133 _ => 0,
134 };
135 if *skip >= decoded {
136 *skip -= decoded;
137 return (0, 0);
138 }
139 let start = std::mem::take(skip);
140 let end = if pad > 0 && pad <= decoded - start {
141 decoded - pad
142 } else {
143 decoded
144 };
145 (start, end)
146}
147
148impl Decoder for OpusDec {
149 fn configure(&mut self, params: &CodecParams) -> Result<()> {
150 let rate = if params.sample_rate == 0 {
152 48_000
153 } else {
154 params.sample_rate
155 };
156 check_rate(rate)?;
157 let head = OpusHead::parse(¶ms.extradata);
158 let channels = match &head {
159 Some(h) if h.channels > 0 => h.channels,
160 _ => params.channels.max(1),
161 };
162 if let Some(family @ 1..) = head.as_ref().map(|h| h.mapping_family) {
165 return Err(Error::unsupported(format!(
166 "opus decode: channel mapping family {family} (multichannel Opus)"
167 )));
168 }
169 let dec = OpusDecoder::new(rate as i32, channels as usize)
170 .map_err(|e| Error::invalid(format!("opus decode init: {e}")))?;
171 self.skip = head
173 .as_ref()
174 .map_or(0, |h| (h.pre_skip as u64 * rate as u64 / 48_000) as usize);
175 self.gain = head.as_ref().map_or(1.0, |h| {
177 10f64.powf(h.gain_q8 as f64 / (20.0 * 256.0)) as f32
178 });
179 self.position = 0;
180 self.dec = Some(dec);
181 self.sample_rate = rate;
182 self.channels = channels;
183 Ok(())
184 }
185
186 fn send_packet(&mut self, packet: &Packet) -> Result<()> {
187 let channels = self.channels as usize;
188 let frame_size = self.sample_rate as usize / 1000 * MAX_PACKET_MS;
190 let dec = self
191 .dec
192 .as_mut()
193 .ok_or_else(|| Error::invalid("opus decode: not configured"))?;
194
195 let mut out = vec![0f32; frame_size * channels];
196 let decoded = dec
197 .decode(&packet.data, frame_size, &mut out)
198 .map_err(|e| Error::invalid(format!("opus decode: {e}")))?;
199 let (start, end) = keep_range(&mut self.skip, self.position, packet, decoded);
200 self.position = packet
201 .end_position
202 .unwrap_or(self.position.saturating_add(decoded as i64));
203 if start == end {
204 return Ok(());
205 }
206
207 let mut bytes = Vec::with_capacity((end - start) * channels * 4);
208 for &s in &out[start * channels..end * channels] {
209 bytes.extend_from_slice(&(s * self.gain).to_le_bytes());
210 }
211 self.queue.push_back(Frame::Audio(AudioFrame {
212 sample_rate: self.sample_rate,
213 channels: self.channels,
214 format: SampleFormat::F32,
215 planes: vec![bytes],
216 samples: end - start,
217 pts: packet.pts,
218 }));
219 Ok(())
220 }
221
222 fn receive_frame(&mut self) -> Result<Frame> {
223 if let Some(frame) = self.queue.pop_front() {
224 return Ok(frame);
225 }
226 if self.eof {
227 Err(Error::Eof)
228 } else {
229 Err(Error::Again)
230 }
231 }
232
233 fn flush(&mut self) {
234 self.eof = true;
235 }
236}
237
238struct OpusEnc {
243 enc: Option<OpusEncoder>,
244 sample_rate: u32,
245 channels: u16,
246 frame_size: usize,
248 buffer: Vec<f32>,
250 next_pts: i64,
252 queue: VecDeque<Packet>,
253 eof: bool,
254 bitrate: i32,
257 complexity: i32,
259 use_cbr: bool,
261 voip: bool,
263 parallel: bool,
268 warmup: usize,
271 threads: usize,
273}
274
275impl Default for OpusEnc {
276 fn default() -> Self {
277 OpusEnc {
278 enc: None,
279 sample_rate: 0,
280 channels: 0,
281 frame_size: 0,
282 buffer: Vec::new(),
283 next_pts: 0,
284 queue: VecDeque::new(),
285 eof: false,
286 bitrate: 64_000,
287 complexity: 9,
288 use_cbr: false,
289 voip: false,
290 parallel: true,
291 warmup: 8,
292 threads: 0,
293 }
294 }
295}
296
297impl OpusEnc {
298 fn init(&mut self, af: &AudioFrame) -> Result<()> {
299 check_rate(af.sample_rate)?;
300 let channels = af.channels.max(1);
301 let app = if self.voip {
302 Application::Voip
303 } else {
304 Application::Audio
305 };
306 let mut enc = OpusEncoder::new(af.sample_rate as i32, channels as usize, app)
307 .map_err(|e| Error::invalid(format!("opus encode init: {e}")))?;
308 enc.bitrate_bps = self.bitrate;
309 enc.complexity = self.complexity.clamp(0, 10);
310 enc.use_cbr = self.use_cbr;
311 self.enc = Some(enc);
312 self.sample_rate = af.sample_rate;
313 self.channels = channels;
314 self.frame_size = af.sample_rate as usize / 1000 * FRAME_MS;
315 Ok(())
316 }
317
318 fn drain_frames(&mut self) -> Result<()> {
320 let frame_samples = self.frame_size * self.channels as usize;
321 if frame_samples == 0 || self.buffer.len() < frame_samples {
322 return Ok(());
323 }
324 let OpusEnc {
331 enc,
332 buffer,
333 queue,
334 next_pts,
335 frame_size,
336 ..
337 } = self;
338 let enc = enc.as_mut().expect("encoder initialized");
339 let frame_size = *frame_size;
340 let n_full = buffer.len() / frame_samples;
341 for i in 0..n_full {
342 let chunk = &buffer[i * frame_samples..(i + 1) * frame_samples];
343 let mut out = vec![0u8; 4000]; let n = enc
345 .encode(chunk, frame_size, &mut out)
346 .map_err(|e| Error::invalid(format!("opus encode: {e}")))?;
347 out.truncate(n);
348 let mut packet = Packet::from_data(0, out);
350 packet.pts = Some(*next_pts);
351 *next_pts += frame_size as i64;
352 queue.push_back(packet);
353 }
354 buffer.drain(..n_full * frame_samples);
355 Ok(())
356 }
357
358 fn drain_parallel(&mut self) {
361 let frame_samples = self.frame_size * self.channels as usize;
362 if frame_samples == 0 || self.buffer.is_empty() {
363 return;
364 }
365 let rem = self.buffer.len() % frame_samples;
366 if rem != 0 {
367 self.buffer
368 .extend(std::iter::repeat_n(0.0, frame_samples - rem));
369 }
370 let app = if self.voip {
371 Application::Voip
372 } else {
373 Application::Audio
374 };
375 let mut cfg = ParallelConfig::new(self.sample_rate as i32, self.channels as usize, app);
376 cfg.bitrate_bps = self.bitrate;
377 cfg.complexity = self.complexity;
378 cfg.use_cbr = self.use_cbr;
379 cfg.warmup = self.warmup;
380 cfg.threads = self.threads;
381 let packets = encode_parallel(&cfg, &self.buffer, self.frame_size);
382 self.buffer.clear();
383 for data in packets {
384 let mut packet = Packet::from_data(0, data);
385 packet.pts = Some(self.next_pts);
386 self.next_pts += self.frame_size as i64;
387 self.queue.push_back(packet);
388 }
389 }
390}
391
392impl Encoder for OpusEnc {
393 fn configure(&mut self, options: &Dictionary) -> Result<()> {
394 if let Some(b) = options.get_bitrate("b") {
395 if b > 0 {
396 self.bitrate = b as i32;
397 }
398 }
399 if let Some(c) = options.get_int("compression_level") {
401 self.complexity = (c as i32).clamp(0, 10);
402 }
403 if let Some(app) = options.get("application") {
404 self.voip = app.eq_ignore_ascii_case("voip") || app.eq_ignore_ascii_case("lowdelay");
405 }
406 if let Some(v) = options.get("vbr") {
408 self.use_cbr = v == "off" || v == "0" || v.eq_ignore_ascii_case("false");
409 }
410 if let Some(p) = options.get("opus_parallel") {
412 self.parallel = !(p == "0" || p == "off" || p.eq_ignore_ascii_case("false"));
413 }
414 if let Some(w) = options.get_int("opus_warmup") {
415 self.warmup = w.max(0) as usize;
416 }
417 if let Some(t) = options.get_int("threads") {
418 self.threads = t.max(0) as usize;
419 }
420 Ok(())
421 }
422
423 fn accepted_sample_rates(&self) -> Option<Vec<u32>> {
424 Some(RATES.to_vec())
425 }
426
427 fn send_frame(&mut self, frame: &Frame) -> Result<()> {
428 let af = match frame {
429 Frame::Audio(a) => a,
430 Frame::Video(_) => {
431 return Err(Error::unsupported(
432 "opus encode: video frame on an audio codec",
433 ))
434 }
435 };
436 if self.enc.is_none() {
437 self.init(af)?;
438 self.next_pts = af.pts.unwrap_or(0);
440 } else if af.sample_rate != self.sample_rate || af.channels != self.channels {
441 return Err(Error::unsupported(
442 "opus encode: stream layout changed mid-stream",
443 ));
444 }
445 self.buffer.extend(frame_to_f32(af)?);
446 if self.parallel {
449 Ok(())
450 } else {
451 self.drain_frames()
452 }
453 }
454
455 fn receive_packet(&mut self) -> Result<Packet> {
456 if let Some(packet) = self.queue.pop_front() {
457 return Ok(packet);
458 }
459 if self.eof {
460 Err(Error::Eof)
461 } else {
462 Err(Error::Again)
463 }
464 }
465
466 fn flush(&mut self) {
467 if self.frame_size != 0 && !self.buffer.is_empty() {
469 if self.parallel {
470 self.drain_parallel();
471 } else {
472 let frame_samples = self.frame_size * self.channels as usize;
474 let rem = self.buffer.len() % frame_samples;
475 if rem != 0 {
476 self.buffer
477 .extend(std::iter::repeat_n(0.0, frame_samples - rem));
478 }
479 let _ = self.drain_frames();
480 }
481 }
482 self.eof = true;
483 }
484}
485
486#[cfg(test)]
487mod tests {
488 use super::*;
489
490 #[test]
493 fn opus_encode_then_decode() {
494 let (rate, channels) = (48_000u32, 1u16);
495 let frame = rate as usize / 1000 * FRAME_MS; let mut samples = Vec::new();
497 for i in 0..frame * 10 {
498 samples.push((i as f32 * 0.05).sin() * 0.3);
499 }
500 let bytes: Vec<u8> = samples.iter().flat_map(|s| s.to_le_bytes()).collect();
501
502 let mut enc = OpusEnc::default();
503 enc.send_frame(&Frame::Audio(AudioFrame {
504 sample_rate: rate,
505 channels,
506 format: SampleFormat::F32,
507 planes: vec![bytes],
508 samples: samples.len(),
509 pts: Some(0),
510 }))
511 .unwrap();
512 enc.flush();
513
514 let mut packets = Vec::new();
515 loop {
516 match enc.receive_packet() {
517 Ok(p) => {
518 assert!(!p.data.is_empty());
519 packets.push(p);
520 }
521 Err(Error::Eof) | Err(Error::Again) => break,
522 Err(e) => panic!("encode: {e}"),
523 }
524 }
525 assert!(
526 packets.len() >= 8,
527 "expected ~10 frames, got {}",
528 packets.len()
529 );
530
531 let mut dec = OpusDec::default();
532 dec.configure(&CodecParams {
533 sample_rate: rate,
534 channels,
535 ..Default::default()
536 })
537 .unwrap();
538 let mut decoded = 0usize;
539 let mut energy = 0f64;
540 for p in &packets {
541 dec.send_packet(p).unwrap();
542 if let Ok(Frame::Audio(af)) = dec.receive_frame() {
543 assert_eq!(af.format, SampleFormat::F32);
544 decoded += af.samples;
545 for b in af.planes[0].chunks_exact(4) {
546 let s = f32::from_le_bytes([b[0], b[1], b[2], b[3]]) as f64;
547 energy += s * s;
548 }
549 }
550 }
551 assert!(decoded >= frame * 8, "decoded too few samples: {decoded}");
552 let rms = (energy / decoded.max(1) as f64).sqrt();
553 assert!(rms > 0.02, "decoded audio is silent: rms {rms:.4}");
554 }
555
556 fn head(channels: u8, pre_skip: u16, gain_q8: i16, family: u8) -> Vec<u8> {
558 let mut v = b"OpusHead".to_vec();
559 v.push(1);
560 v.push(channels);
561 v.extend_from_slice(&pre_skip.to_le_bytes());
562 v.extend_from_slice(&48_000u32.to_le_bytes());
563 v.extend_from_slice(&gain_q8.to_le_bytes());
564 v.push(family);
565 if family != 0 {
566 v.extend_from_slice(&[4, 2, 0, 4, 1, 2, 3, 5]); }
568 v
569 }
570
571 fn encoded(n: usize) -> Vec<Packet> {
573 let mut enc = OpusEnc {
574 parallel: false,
575 ..Default::default()
576 };
577 let samples: Vec<f32> = (0..960 * n * 2)
578 .map(|i| ((i / 2) as f32 * 0.05).sin() * 0.3)
579 .collect();
580 enc.send_frame(&Frame::Audio(AudioFrame {
581 sample_rate: 48_000,
582 channels: 2,
583 format: SampleFormat::F32,
584 planes: vec![samples.iter().flat_map(|s| s.to_le_bytes()).collect()],
585 samples: 960 * n,
586 pts: Some(0),
587 }))
588 .unwrap();
589 enc.flush();
590 let mut packets = Vec::new();
591 while let Ok(p) = enc.receive_packet() {
592 packets.push(p);
593 }
594 assert_eq!(packets.len(), n);
595 packets
596 }
597
598 fn decode(extradata: Vec<u8>, packets: &[Packet]) -> Vec<f32> {
600 let mut dec = OpusDec::default();
601 dec.configure(&CodecParams {
602 sample_rate: 48_000,
603 channels: 2,
604 extradata,
605 ..Default::default()
606 })
607 .unwrap();
608 let mut out = Vec::new();
609 for p in packets {
610 dec.send_packet(p).unwrap();
611 while let Ok(Frame::Audio(af)) = dec.receive_frame() {
612 assert_eq!(af.planes[0].len(), af.samples * 2 * 4);
613 out.extend(
614 af.planes[0]
615 .chunks_exact(4)
616 .map(|b| f32::from_le_bytes([b[0], b[1], b[2], b[3]])),
617 );
618 }
619 }
620 out
621 }
622
623 #[test]
624 fn opus_head_parses() {
625 let h = OpusHead::parse(&head(2, 312, -256, 0)).unwrap();
626 assert_eq!(
627 h,
628 OpusHead {
629 channels: 2,
630 pre_skip: 312,
631 gain_q8: -256,
632 mapping_family: 0,
633 }
634 );
635 assert_eq!(OpusHead::parse(b"OpusTags........."), None);
636 assert_eq!(OpusHead::parse(&head(2, 312, 0, 0)[..18]), None);
637 }
638
639 #[test]
640 fn keep_range_mirrors_ffmpeg() {
641 let mut p = Packet::from_data(0, Vec::new());
642 let mut skip = 1000;
644 assert_eq!(keep_range(&mut skip, 0, &p, 960), (0, 0));
645 assert_eq!(skip, 40);
646 assert_eq!(keep_range(&mut skip, 960, &p, 960), (40, 960));
647 assert_eq!(skip, 0);
648 p.end_position = Some(5000);
650 assert_eq!(keep_range(&mut skip, 4800, &p, 960), (0, 960));
651 p.flags.end_of_stream = true;
652 assert_eq!(keep_range(&mut skip, 4800, &p, 960), (0, 200));
653 assert_eq!(keep_range(&mut skip, 3000, &p, 960), (0, 960), "no overrun");
654 assert_eq!(keep_range(&mut skip, 6000, &p, 960), (0, 0), "capped: all");
655 let mut skip = 100;
657 assert_eq!(keep_range(&mut skip, 4800, &p, 960), (100, 200));
658 let mut skip = 900;
659 assert_eq!(keep_range(&mut skip, 4800, &p, 960), (900, 960));
660 }
661
662 #[test]
665 fn pre_skip_and_final_granule_trim_the_stream() {
666 let mut packets = encoded(10);
667 let raw = decode(head(2, 0, 0, 0), &packets);
668 assert_eq!(raw.len(), 960 * 10 * 2);
669 let granule_end = 960 * 10 - 500;
671 for (i, p) in packets.iter_mut().enumerate() {
672 if i == 9 {
673 p.end_position = Some(granule_end);
674 p.flags.end_of_stream = true;
675 } else if i % 4 == 3 {
676 p.end_position = Some(960 * (i as i64 + 1));
677 }
678 }
679 let got = decode(head(2, 312, 0, 0), &packets);
680 let want = &raw[312 * 2..granule_end as usize * 2];
681 assert_eq!(got.len(), want.len());
682 assert_eq!(got, want);
683 }
684
685 #[test]
687 fn output_gain_is_applied() {
688 let packets = encoded(3);
689 let raw = decode(head(2, 0, 0, 0), &packets);
690 let loud = decode(head(2, 0, 6 * 256, 0), &packets); let g = 10f64.powf(6.0 / 20.0) as f32;
692 assert_eq!(loud.len(), raw.len());
693 for (a, b) in raw.iter().zip(&loud) {
694 assert_eq!(a * g, *b);
695 }
696 }
697
698 #[test]
701 fn multichannel_mapping_is_refused() {
702 let mut dec = OpusDec::default();
703 let err = dec
704 .configure(&CodecParams {
705 sample_rate: 48_000,
706 channels: 6,
707 extradata: head(6, 312, 0, 1),
708 ..Default::default()
709 })
710 .unwrap_err();
711 assert!(matches!(err, Error::Unsupported(_)), "{err}");
712 }
713}