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