add decoder to bark-core

Cargo.lock

@@ -142,6 +142,7 @@ dependencies = [
  "derive_more",
  "heapless",
  "log",
+ "thiserror",
 ]
 
 [[package]]

Cargo.toml

@@ -10,8 +10,9 @@ members = [
 bark-core = { path = "bark-core" }
 bark-protocol = { path = "bark-protocol" }
 
-bitflags = { version = "2.4.0", features = ["bytemuck"] }
+bitflags = { version = "2.4", features = ["bytemuck"] }
 bytemuck = { version = "1.13", features = ["derive"] }
 derive_more = { version = "0.99" }
 heapless = "0.7"
 log = "0.4"
+thiserror = "1.0"

bark-core/Cargo.toml

@@ -12,3 +12,4 @@ bytemuck = { workspace = true }
 derive_more = { workspace = true }
 heapless = { workspace = true }
 log = { workspace = true }
+thiserror = { workspace = true }

bark-core/src/decode/mod.rs

@@ -0,0 +1,74 @@
+mod pcm;
+
+use core::fmt::Display;
+
+use bark_protocol::packet::Audio;
+use thiserror::Error;
+
+use bark_protocol::types::{AudioPacketHeader, AudioPacketFormat};
+use bark_protocol::SAMPLES_PER_PACKET;
+
+#[derive(Debug, Error)]
+pub enum NewDecoderError {
+    #[error("unknown format in audio header: {0:?}")]
+    UnknownFormat(AudioPacketFormat),
+}
+
+#[derive(Debug, Error)]
+pub enum DecodeError {
+    #[error("wrong length: {length}, expected: {expected}")]
+    WrongLength { length: usize, expected: usize }
+}
+
+pub struct Decoder {
+    decode: DecodeFormat,
+}
+
+pub type SampleBuffer = [f32; SAMPLES_PER_PACKET];
+
+impl Decoder {
+    pub fn new(header: &AudioPacketHeader) -> Result<Self, NewDecoderError> {
+        let decode = match header.format {
+            AudioPacketFormat::S16LE => DecodeFormat::S16LE(pcm::S16LEDecoder),
+            AudioPacketFormat::F32LE => DecodeFormat::F32LE(pcm::F32LEDecoder),
+            format => { return Err(NewDecoderError::UnknownFormat(format)) }
+        };
+
+        Ok(Decoder { decode })
+    }
+
+    pub fn describe(&self) -> impl Display + '_ {
+        &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)
+    }
+}
+
+trait Decode: Display {
+    fn decode_packet(&mut self, bytes: &[u8], out: &mut SampleBuffer) -> Result<(), DecodeError>;
+}
+
+enum DecodeFormat {
+    S16LE(pcm::S16LEDecoder),
+    F32LE(pcm::F32LEDecoder),
+}
+
+impl Decode for DecodeFormat {
+    fn decode_packet(&mut self, bytes: &[u8], out: &mut SampleBuffer) -> Result<(), DecodeError> {
+        match self {
+            DecodeFormat::S16LE(dec) => dec.decode_packet(bytes, out),
+            DecodeFormat::F32LE(dec) => dec.decode_packet(bytes, out),
+        }
+    }
+}
+
+impl Display for DecodeFormat {
+    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
+        match self {
+            DecodeFormat::S16LE(dec) => dec.fmt(f),
+            DecodeFormat::F32LE(dec) => dec.fmt(f),
+        }
+    }
+}

bark-core/src/decode/pcm.rs

@@ -0,0 +1,63 @@
+use core::fmt::{self, Display};
+
+use super::{Decode, DecodeError, SampleBuffer};
+
+pub struct S16LEDecoder;
+
+impl Display for S16LEDecoder {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "signed16 (little endian)")
+    }
+}
+
+impl Decode for S16LEDecoder {
+    fn decode_packet(&mut self, bytes: &[u8], out: &mut SampleBuffer) -> Result<(), DecodeError> {
+        decode_packed(bytes, out, |bytes| {
+            let input = i16::from_le_bytes(bytes);
+            let scale = (u16::MIN as f32).abs();
+            input as f32 / scale
+        })
+    }
+}
+
+pub struct F32LEDecoder;
+
+impl Display for F32LEDecoder {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "float32 (little endian)")
+    }
+}
+
+impl Decode for F32LEDecoder {
+    fn decode_packet(&mut self, bytes: &[u8], out: &mut SampleBuffer) -> Result<(), DecodeError> {
+        decode_packed(bytes, out, f32::from_le_bytes)
+    }
+}
+
+fn decode_packed<const N: usize>(
+    bytes: &[u8],
+    out: &mut SampleBuffer,
+    func: impl Fn([u8; N]) -> f32,
+) -> Result<(), DecodeError> {
+    check_length(bytes, out.len() * N)?;
+
+    for (input, output) in bytes.chunks_exact(N).zip(out) {
+        // when array_chunks stabilises we can use that instead
+        // but for now use try_into to turn a &[u8] (guaranteed len == width)
+        // into a [u8; width]
+        let input = input.try_into().unwrap();
+        *output = func(input);
+    }
+
+    Ok(())
+}
+
+fn check_length(bytes: &[u8], expected: usize) -> Result<(), DecodeError> {
+    let length = bytes.len();
+
+    if length == expected {
+        Ok(())
+    } else {
+        Err(DecodeError::WrongLength { length, expected })
+    }
+}

bark-core/src/lib.rs

@@ -1,2 +1,3 @@
 pub mod receive;
 pub mod consts;
+pub mod decode;

bark-protocol/src/packet.rs

@@ -89,18 +89,18 @@ pub enum PacketKind {
 pub struct Audio(Packet);
 
 impl Audio {
-    const LENGTH: usize =
-        size_of::<types::AudioPacketHeader>() +
-        size_of::<types::AudioPacketBuffer>();
+    const HEADER_LENGTH: usize =
+        size_of::<types::AudioPacketHeader>();
 
-    pub fn allocate() -> Result<Audio, AllocError> {
-        let packet = Packet::allocate(Magic::AUDIO, Self::LENGTH)?;
+    pub fn allocate(buffer_length: usize) -> Result<Audio, AllocError> {
+        let length = Self::HEADER_LENGTH + buffer_length;
+        let packet = Packet::allocate(Magic::AUDIO, length)?;
 
         Ok(Audio(packet))
     }
 
     pub fn parse(packet: Packet) -> Option<Self> {
-        if packet.len() != Self::LENGTH {
+        if packet.len() <= Self::HEADER_LENGTH {
             return None;
         }
 
@@ -115,13 +115,13 @@ impl Audio {
         &self.0
     }
 
-    pub fn buffer(&self) -> &[f32] {
+    pub fn buffer_bytes(&self) -> &[u8] {
         let header_size = size_of::<types::AudioPacketHeader>();
         let buffer_bytes = &self.0.as_bytes()[header_size..];
         bytemuck::cast_slice(buffer_bytes)
     }
 
-    pub fn buffer_mut(&mut self) -> &mut [f32] {
+    pub fn buffer_bytes_mut(&mut self) -> &mut [u8] {
         let header_size = size_of::<types::AudioPacketHeader>();
         let buffer_bytes = &mut self.0.as_bytes_mut()[header_size..];
         bytemuck::cast_slice_mut(buffer_bytes)
@@ -148,7 +148,10 @@ impl Time {
     // that time packets experience as similar delay as possible to audio
     // packets for most accurate synchronisation, so we pad this packet out
     // to the same size as the audio packet
-    const LENGTH: usize = Audio::LENGTH;
+
+    // TODO fix this
+    // const LENGTH: usize = Audio::LENGTH;
+    const LENGTH: usize = size_of::<types::TimePacket>();
 
     // time packets are padded so that they are
     // the same length as audio packets:

bark-protocol/src/types.rs

@@ -53,7 +53,7 @@ pub struct AudioPacketHeader {
 
 /// This, regrettably, has to be a u64 to fill out `AudioPacketHeader` with
 /// no hidden padding. TODO this whole protocol tier needs a big rethink
-#[derive(Debug, Clone, Copy, Zeroable, Pod)]
+#[derive(Debug, Clone, Copy, Zeroable, Pod, PartialEq, Eq)]
 #[repr(transparent)]
 pub struct AudioPacketFormat(u64);
 

bark/Cargo.toml

@@ -24,6 +24,6 @@ socket2 = "0.5.3"
 static_assertions = "1.1.0"
 structopt = "0.3.26"
 termcolor = "1.2.0"
-thiserror = "1.0.51"
+thiserror = { workspace = true }
 toml = "0.7.6"
 xdg = "2.5.2"

bark/src/receive.rs

@@ -2,6 +2,7 @@ use std::array;
 use std::sync::{Arc, Mutex};
 use std::time::Duration;
 
+use bark_core::decode::{Decoder, SampleBuffer};
 use bytemuck::Zeroable;
 use structopt::StructOpt;
 
@@ -27,25 +28,40 @@ pub struct Receiver {
 
 struct Stream {
     sid: SessionId,
-    resampler: Resampler,
-    rate_adjust: RateAdjust,
     latency: Aggregate<Duration>,
     clock_delta: Aggregate<ClockDelta>,
     queue: PacketQueue,
+    /// None indicates error creating decoder, we cannot decode this stream
+    decoder: Option<Decoder>,
+    resampler: Resampler,
+    rate_adjust: RateAdjust,
 }
 
 impl Stream {
     pub fn new(header: &AudioPacketHeader) -> Self {
-        let resampler = Resampler::new();
         let queue = PacketQueue::new(header);
 
+        let decoder = match Decoder::new(header) {
+            Ok(dec) => {
+                log::info!("instantiated decoder for new stream: {}", dec.describe());
+                Some(dec)
+            }
+            Err(err) => {
+                log::error!("error creating decoder for new stream: {err}");
+                None
+            }
+        };
+
+        let resampler = Resampler::new();
+
         Stream {
             sid: header.sid,
-            resampler,
-            rate_adjust: RateAdjust::new(),
             latency: Aggregate::new(),
             clock_delta: Aggregate::new(),
             queue,
+            decoder,
+            resampler,
+            rate_adjust: RateAdjust::new(),
         }
     }
 
@@ -169,6 +185,7 @@ impl Receiver {
                 play: stream_pts,
             });
 
+        // adjust resampler rate based on packet timing info
         if let Some(timing) = timing {
             let rate = stream.rate_adjust.sample_rate(timing);
 
@@ -183,11 +200,25 @@ impl Receiver {
             self.stats.set_audio_latency(timing.real, timing.play);
         }
 
-        let resample = stream.resampler.process_interleaved(packet.buffer(), buffer)
+        // 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) {
+                Ok(()) => {}
+                Err(e) => {
+                    log::warn!("error in decoder, skipping packet: {e}");
+                    decode_buffer.fill(0.0);
+                }
+            }
+        }
+
+        // resample decoded audio
+        let resample = stream.resampler.process_interleaved(&decode_buffer, buffer)
             .expect("resample error!");
 
-        assert_eq!(resample.input_read.as_buffer_offset(), packet.buffer().len());
+        assert_eq!(resample.input_read.as_buffer_offset(), decode_buffer.len());
 
+        // report stats and return
         self.stats.set_buffer_length(
             SampleDuration::from_frame_count(
                 (FRAMES_PER_PACKET * stream.queue.len()).try_into().unwrap()));

bark/src/stream.rs

@@ -1,6 +1,7 @@
 use std::sync::Arc;
 use std::time::Duration;
 
+use bark_protocol::SAMPLES_PER_PACKET;
 use structopt::StructOpt;
 
 use bark_protocol::time::SampleDuration;
@@ -75,11 +76,14 @@ pub fn run(opt: StreamOpt) -> Result<(), RunError> {
 
             loop {
                 // create new audio buffer
-                let mut audio = Audio::allocate()
+                let buffer_bytes_length = core::mem::size_of::<f32>() * SAMPLES_PER_PACKET;
+                let mut audio = Audio::allocate(buffer_bytes_length)
                     .expect("allocate Audio packet");
 
+                let sample_buffer = bytemuck::cast_slice_mut(audio.buffer_bytes_mut());
+
                 // read audio input
-                let timestamp = match input.read(audio.buffer_mut()) {
+                let timestamp = match input.read(sample_buffer) {
                     Ok(ts) => ts,
                     Err(e) => {
                         log::error!("error reading audio input: {e}");