Merge branch 'opus'

Cargo.lock

@@ -101,6 +101,17 @@ dependencies = [
  "winapi",
 ]
 
+[[package]]
+name = "audiopus_sys"
+version = "0.2.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "62314a1546a2064e033665d658e88c620a62904be945f8147e6b16c3db9f8651"
+dependencies = [
+ "cmake",
+ "log",
+ "pkg-config",
+]
+
 [[package]]
 name = "autocfg"
 version = "1.1.0"
@@ -142,6 +153,7 @@ dependencies = [
  "derive_more",
  "heapless",
  "log",
+ "opus",
  "thiserror",
 ]
 
@@ -864,6 +876,16 @@ version = "1.18.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "dd8b5dd2ae5ed71462c540258bedcb51965123ad7e7ccf4b9a8cafaa4a63576d"
 
+[[package]]
+name = "opus"
+version = "0.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6526409b274a7e98e55ff59d96aafd38e6cd34d46b7dbbc32ce126dffcd75e8e"
+dependencies = [
+ "audiopus_sys",
+ "libc",
+]
+
 [[package]]
 name = "peeking_take_while"
 version = "0.1.2"

bark-core/Cargo.toml

@@ -12,4 +12,5 @@ bytemuck = { workspace = true }
 derive_more = { workspace = true }
 heapless = { workspace = true }
 log = { workspace = true }
+opus = "0.3.0"
 thiserror = { workspace = true }

bark-core/src/decode/mod.rs

@@ -1,4 +1,5 @@
-mod pcm;
+pub mod opus;
+pub mod pcm;
 
 use core::fmt::Display;
 
@@ -12,12 +13,16 @@ use bark_protocol::SAMPLES_PER_PACKET;
 pub enum NewDecoderError {
     #[error("unknown format in audio header: {0:?}")]
     UnknownFormat(AudioPacketFormat),
+    #[error("opus codec error: {0}")]
+    Opus(#[from] ::opus::Error),
 }
 
 #[derive(Debug, Error)]
 pub enum DecodeError {
     #[error("wrong length: {length}, expected: {expected}")]
-    WrongLength { length: usize, expected: usize }
+    WrongLength { length: usize, expected: usize },
+    #[error("opus codec error: {0}")]
+    Opus(#[from] ::opus::Error),
 }
 
 pub struct Decoder {
@@ -31,6 +36,7 @@ impl Decoder {
         let decode = match header.format {
             AudioPacketFormat::S16LE => DecodeFormat::S16LE(pcm::S16LEDecoder),
             AudioPacketFormat::F32LE => DecodeFormat::F32LE(pcm::F32LEDecoder),
+            AudioPacketFormat::OPUS => DecodeFormat::Opus(opus::OpusDecoder::new()?),
             format => { return Err(NewDecoderError::UnknownFormat(format)) }
         };
 
@@ -41,25 +47,28 @@ impl Decoder {
         &self.decode as &dyn Display
     }
 
-    pub fn decode(&mut self, packet: &Audio, out: &mut SampleBuffer) -> Result<(), DecodeError> {
-        self.decode.decode_packet(packet.buffer_bytes(), out)
+    pub fn decode(&mut self, packet: Option<&Audio>, out: &mut SampleBuffer) -> Result<(), DecodeError> {
+        let bytes = packet.map(|packet| packet.buffer_bytes());
+        self.decode.decode_packet(bytes, out)
     }
 }
 
 trait Decode: Display {
-    fn decode_packet(&mut self, bytes: &[u8], out: &mut SampleBuffer) -> Result<(), DecodeError>;
+    fn decode_packet(&mut self, bytes: Option<&[u8]>, out: &mut SampleBuffer) -> Result<(), DecodeError>;
 }
 
 enum DecodeFormat {
     S16LE(pcm::S16LEDecoder),
     F32LE(pcm::F32LEDecoder),
+    Opus(opus::OpusDecoder),
 }
 
 impl Decode for DecodeFormat {
-    fn decode_packet(&mut self, bytes: &[u8], out: &mut SampleBuffer) -> Result<(), DecodeError> {
+    fn decode_packet(&mut self, bytes: Option<&[u8]>, out: &mut SampleBuffer) -> Result<(), DecodeError> {
         match self {
             DecodeFormat::S16LE(dec) => dec.decode_packet(bytes, out),
             DecodeFormat::F32LE(dec) => dec.decode_packet(bytes, out),
+            DecodeFormat::Opus(dec) => dec.decode_packet(bytes, out),
         }
     }
 }
@@ -69,6 +78,7 @@ impl Display for DecodeFormat {
         match self {
             DecodeFormat::S16LE(dec) => dec.fmt(f),
             DecodeFormat::F32LE(dec) => dec.fmt(f),
+            DecodeFormat::Opus(dec) => dec.fmt(f),
         }
     }
 }

bark-core/src/decode/opus.rs

@@ -0,0 +1,43 @@
+use core::fmt::{self, Display};
+
+use bark_protocol::SAMPLE_RATE;
+
+use super::{Decode, DecodeError, SampleBuffer};
+
+pub struct OpusDecoder {
+    opus: opus::Decoder,
+}
+
+impl OpusDecoder {
+    pub fn new() -> Result<Self, opus::Error> {
+        let opus = opus::Decoder::new(
+            SAMPLE_RATE.0,
+            opus::Channels::Stereo,
+        )?;
+
+        Ok(OpusDecoder { opus })
+    }
+}
+
+impl Display for OpusDecoder {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "opus")
+    }
+}
+
+impl Decode for OpusDecoder {
+    fn decode_packet(&mut self, bytes: Option<&[u8]>, out: &mut SampleBuffer) -> Result<(), DecodeError> {
+        let expected = out.len() / 2;
+
+        let length = match bytes {
+            Some(bytes) => self.opus.decode_float(bytes, out, false)?,
+            None => self.opus.decode_float(&[], out, true)?,
+        };
+
+        if expected != length {
+            return Err(DecodeError::WrongLength { length, expected });
+        }
+
+        Ok(())
+    }
+}

bark-core/src/decode/pcm.rs

@@ -11,7 +11,7 @@ impl Display for S16LEDecoder {
 }
 
 impl Decode for S16LEDecoder {
-    fn decode_packet(&mut self, bytes: &[u8], out: &mut SampleBuffer) -> Result<(), DecodeError> {
+    fn decode_packet(&mut self, bytes: Option<&[u8]>, out: &mut SampleBuffer) -> Result<(), DecodeError> {
         decode_packed(bytes, out, |bytes| {
             let input = i16::from_le_bytes(bytes);
             let scale = i16::MAX as f32;
@@ -29,16 +29,23 @@ impl Display for F32LEDecoder {
 }
 
 impl Decode for F32LEDecoder {
-    fn decode_packet(&mut self, bytes: &[u8], out: &mut SampleBuffer) -> Result<(), DecodeError> {
+    fn decode_packet(&mut self, bytes: Option<&[u8]>, out: &mut SampleBuffer) -> Result<(), DecodeError> {
         decode_packed(bytes, out, f32::from_le_bytes)
     }
 }
 
 fn decode_packed<const N: usize>(
-    bytes: &[u8],
+    bytes: Option<&[u8]>,
     out: &mut SampleBuffer,
     func: impl Fn([u8; N]) -> f32,
 ) -> Result<(), DecodeError> {
+    let Some(bytes) = bytes else {
+        // PCM codecs have no packet loss correction
+        // just zero fill and return
+        out.fill(0.0);
+        return Ok(());
+    };
+
     check_length(bytes, out.len() * N)?;
 
     for (input, output) in bytes.chunks_exact(N).zip(out) {

bark-core/src/encode/mod.rs

@@ -1,3 +1,4 @@
+pub mod opus;
 pub mod pcm;
 
 use core::fmt::Display;
@@ -5,10 +6,18 @@ use core::fmt::Display;
 use bark_protocol::types::AudioPacketFormat;
 use thiserror::Error;
 
+#[derive(Debug, Error)]
+pub enum NewEncoderError {
+    #[error("opus codec error: {0}")]
+    Opus(#[from] ::opus::Error),
+}
+
 #[derive(Debug, Error)]
 pub enum EncodeError {
     #[error("output buffer too small, need at least {need} bytes")]
     OutputBufferTooSmall { need: usize },
+    #[error("opus codec error: {0}")]
+    Opus(#[from] ::opus::Error),
 }
 
 pub trait Encode: Display + Send {

bark-core/src/encode/opus.rs

@@ -0,0 +1,41 @@
+use core::fmt::{self, Display};
+
+use bark_protocol::{types::AudioPacketFormat, SAMPLE_RATE};
+
+use super::{Encode, EncodeError, NewEncoderError};
+
+pub struct OpusEncoder {
+    opus: opus::Encoder,
+}
+
+impl OpusEncoder {
+    pub fn new() -> Result<Self, NewEncoderError> {
+        let mut opus = opus::Encoder::new(
+            SAMPLE_RATE.0,
+            opus::Channels::Stereo,
+            opus::Application::Audio,
+        )?;
+
+        opus.set_inband_fec(true)?;
+        opus.set_packet_loss_perc(50)?;
+        opus.set_bitrate(opus::Bitrate::Max)?;
+
+        Ok(OpusEncoder { opus })
+    }
+}
+
+impl Display for OpusEncoder {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "opus")
+    }
+}
+
+impl Encode for OpusEncoder {
+    fn header_format(&self) -> AudioPacketFormat {
+        AudioPacketFormat::OPUS
+    }
+
+    fn encode_packet(&mut self, samples: &[f32], out: &mut [u8]) -> Result<usize, EncodeError> {
+        Ok(self.opus.encode_float(samples, out)?)
+    }
+}

bark-protocol/src/types.rs

@@ -60,6 +60,7 @@ pub struct AudioPacketFormat(u64);
 impl AudioPacketFormat {
     pub const F32LE: Self = Self(1);
     pub const S16LE: Self = Self(2);
+    pub const OPUS: Self = Self(3);
 }
 
 pub type AudioPacketBuffer = [f32; SAMPLES_PER_PACKET];

bark/src/config.rs

@@ -29,6 +29,7 @@ pub struct Source {
 pub enum Format {
     S16LE,
     F32LE,
+    Opus,
 }
 
 #[derive(Debug, Error)]
@@ -42,6 +43,7 @@ impl FromStr for Format {
         match s {
             "s16le" => Ok(Format::S16LE),
             "f32le" => Ok(Format::F32LE),
+            "opus" => Ok(Format::Opus),
             _ => Err(UnknownFormat),
         }
     }
@@ -52,6 +54,7 @@ impl Display for Format {
         match self {
             Format::S16LE => write!(f, "s16le"),
             Format::F32LE => write!(f, "f32le"),
+            Format::Opus => write!(f, "opus"),
         }
     }
 }

bark/src/main.rs

@@ -23,11 +23,13 @@ enum Opt {
 #[derive(Debug, Error)]
 pub enum RunError {
     #[error("opening network socket: {0}")]
-    Listen(socket::ListenError),
+    Listen(#[from] socket::ListenError),
     #[error("opening audio device: {0}")]
-    OpenAudioDevice(audio::config::OpenError),
+    OpenAudioDevice(#[from] audio::config::OpenError),
     #[error("receiving from network: {0}")]
     Receive(std::io::Error),
+    #[error("opening encoder: {0}")]
+    OpenEncoder(#[from] bark_core::encode::NewEncoderError),
 }
 
 fn main() -> Result<(), ExitCode> {

bark/src/receive.rs

@@ -171,21 +171,23 @@ impl Receiver {
             return SampleDuration::ONE_PACKET;
         };
 
-        let Some(packet) = stream.queue.pop_front() else {
-            // no packets yet
-            buffer[0..SAMPLES_PER_PACKET].fill(0f32);
-            return SampleDuration::ONE_PACKET;
-        };
+        // get next packet from queue, or None if missing (packet loss)
+        let packet = stream.queue.pop_front();
 
-        let header_pts = Timestamp::from_micros_lossy(packet.header().pts);
+        // calculate stream timing from packet timing info if present
+        let header_pts = packet.as_ref()
+            .map(|packet| packet.header().pts)
+            .map(Timestamp::from_micros_lossy);
 
-        let timing = stream.adjust_pts(header_pts)
-            .map(|stream_pts| Timing {
-                real: pts,
-                play: stream_pts,
-            });
+        let stream_pts = header_pts
+            .and_then(|header_pts| stream.adjust_pts(header_pts));
 
-        // adjust resampler rate based on packet timing info
+        let timing = stream_pts.map(|stream_pts| Timing {
+            real: pts,
+            play: stream_pts,
+        });
+
+        // adjust resampler rate based on stream timing info
         if let Some(timing) = timing {
             let rate = stream.rate_adjust.sample_rate(timing);
 
@@ -203,7 +205,7 @@ impl Receiver {
         // decode packet
         let mut decode_buffer: SampleBuffer = array::from_fn(|_| 0.0);
         if let Some(decoder) = stream.decoder.as_mut() {
-            match decoder.decode(&packet, &mut decode_buffer) {
+            match decoder.decode(packet.as_ref(), &mut decode_buffer) {
                 Ok(()) => {}
                 Err(e) => {
                     log::warn!("error in decoder, skipping packet: {e}");

bark/src/stream.rs

@@ -2,6 +2,7 @@ use std::sync::Arc;
 use std::time::Duration;
 
 use bark_core::encode::Encode;
+use bark_core::encode::opus::OpusEncoder;
 use bark_core::encode::pcm::{S16LEEncoder, F32LEEncoder};
 use bark_protocol::SAMPLES_PER_PACKET;
 use structopt::StructOpt;
@@ -57,10 +58,9 @@ pub fn run(opt: StreamOpt) -> Result<(), RunError> {
         buffer: opt.input_buffer
             .map(SampleDuration::from_frame_count)
             .unwrap_or(DEFAULT_BUFFER),
-    }).map_err(RunError::OpenAudioDevice)?;
+    })?;
 
-    let socket = Socket::open(opt.socket)
-        .map_err(RunError::Listen)?;
+    let socket = Socket::open(opt.socket)?;
 
     let protocol = Arc::new(ProtocolSocket::new(socket));
 
@@ -70,9 +70,10 @@ pub fn run(opt: StreamOpt) -> Result<(), RunError> {
     let sid = generate_session_id();
     let node = stats::node::get();
 
-    let mut encoder = match opt.format {
-        config::Format::S16LE => Box::new(S16LEEncoder) as Box<dyn Encode>,
-        config::Format::F32LE => Box::new(F32LEEncoder) as Box<dyn Encode>,
+    let mut encoder: Box<dyn Encode> = match opt.format {
+        config::Format::S16LE => Box::new(S16LEEncoder),
+        config::Format::F32LE => Box::new(F32LEEncoder),
+        config::Format::Opus => Box::new(OpusEncoder::new()?),
     };
 
     log::info!("instantiated encoder: {}", encoder);