From c93cdd7e1e4364d37a4b7f1adc665421e55f818c Mon Sep 17 00:00:00 2001
From: Patrick Moessler <pub@asaril.de>
Date: Fri, 21 Mar 2025 19:35:58 +0100
Subject: [PATCH] split remaining tasks into own files. add triple buffer
---
src/audio_input.rs | 73 ++++++++++++++
src/{audio.rs => audio_process.rs} | 0
src/effects/bass_sparks.rs | 2 +-
src/effects/led_effect.rs | 2 +-
src/led_output.rs | 37 +++++++
src/main.rs | 135 ++++----------------------
src/triple_buffer.rs | 149 +++++++++++++++++++++++++++++
7 files changed, 278 insertions(+), 120 deletions(-)
create mode 100644 src/audio_input.rs
rename src/{audio.rs => audio_process.rs} (100%)
create mode 100644 src/led_output.rs
create mode 100644 src/triple_buffer.rs
diff --git a/src/audio_input.rs b/src/audio_input.rs
new file mode 100644
index 0000000..f006f78
--- /dev/null
+++ b/src/audio_input.rs
@@ -0,0 +1,73 @@
+use bytemuck::bytes_of_mut;
+use embassy_time::Duration;
+use esp_idf_svc::hal::{
+ gpio::{AnyIOPin, Gpio4, Gpio5, Gpio6},
+ i2s::{self, I2sRx, I2S0},
+};
+
+use crate::{audio_process::AudioBuffer, config::AUDIO_SAMPLES_PER_BUF};
+
+#[embassy_executor::task]
+pub async fn mic_input_task(i2s_per: I2S0, sd: Gpio5, sclk: Gpio4, ws: Gpio6) {
+ let std_cfg = i2s::config::StdConfig::new(
+ i2s::config::Config::new().role(i2s::config::Role::Controller),
+ i2s::config::StdClkConfig::new(
+ 24000,
+ i2s::config::ClockSource::default(),
+ i2s::config::MclkMultiple::M256,
+ ),
+ i2s::config::StdSlotConfig::msb_slot_default(
+ i2s::config::DataBitWidth::Bits32,
+ i2s::config::SlotMode::Stereo,
+ )
+ .data_bit_width(i2s::config::DataBitWidth::Bits32)
+ .slot_bit_width(i2s::config::SlotBitWidth::Bits32)
+ .slot_mode_mask(
+ i2s::config::SlotMode::Stereo,
+ i2s::config::StdSlotMask::Both,
+ )
+ .ws_width(32)
+ .ws_polarity(false)
+ .bit_shift(true)
+ .left_align(true)
+ .big_endian(false)
+ .bit_order_lsb(false),
+ i2s::config::StdGpioConfig::new(false, false, false),
+ );
+ let mut mic_drv =
+ i2s::I2sDriver::new_std_rx(i2s_per, &std_cfg, sclk, sd, AnyIOPin::none(), ws).unwrap();
+
+ mic_drv.rx_enable().expect("rx not enabled");
+
+ async fn ignore_mic_startup(mic: &mut i2s::I2sDriver<'_, I2sRx>) {
+ let mut tmp_buf: [u8; 128] = [0; 128];
+ loop {
+ mic.read_async(tmp_buf.as_mut_slice()).await.unwrap();
+ }
+ }
+
+ embassy_time::with_timeout(Duration::from_millis(100), ignore_mic_startup(&mut mic_drv))
+ .await
+ .expect_err("ignore died early");
+
+ loop {
+ let audio_in_buf: &mut AudioBuffer = audio_sender.send().await;
+
+ let buffer = bytes_of_mut(audio_in_buf);
+ let mut total_bytes_read: usize = 0;
+ let mut remaining_bytes = AUDIO_SAMPLES_PER_BUF * 4;
+
+ while total_bytes_read < AUDIO_SAMPLES_PER_BUF * 4 {
+ let chunk = &mut buffer[total_bytes_read..total_bytes_read + remaining_bytes];
+ let num_bytes_read = mic_drv.read_async(chunk).await.unwrap();
+ total_bytes_read += num_bytes_read;
+ remaining_bytes -= num_bytes_read;
+ }
+
+ if total_bytes_read != AUDIO_SAMPLES_PER_BUF * 4 {
+ log::warn!("buffer underflow: {}", total_bytes_read);
+ }
+
+ audio_sender.send_done();
+ }
+}
diff --git a/src/audio.rs b/src/audio_process.rs
similarity index 100%
rename from src/audio.rs
rename to src/audio_process.rs
diff --git a/src/effects/bass_sparks.rs b/src/effects/bass_sparks.rs
index 157e8de..c6e2374 100644
--- a/src/effects/bass_sparks.rs
+++ b/src/effects/bass_sparks.rs
@@ -3,7 +3,7 @@ use embassy_sync::zerocopy_channel::Sender;
use embassy_sync::{blocking_mutex::raw::NoopRawMutex, zerocopy_channel::Receiver};
use embassy_time::{Duration, Ticker};
-use crate::audio::{AudioStats, DspBuffer};
+use crate::audio_process::{AudioStats, DspBuffer};
use crate::effects::led_effect::{LedColors, LedData, LedEffect, Rgbv};
use crate::helpers::random_at_most;
use crate::LED_COUNT;
diff --git a/src/effects/led_effect.rs b/src/effects/led_effect.rs
index ea51d04..03204cf 100644
--- a/src/effects/led_effect.rs
+++ b/src/effects/led_effect.rs
@@ -6,7 +6,7 @@ use embassy_sync::{
};
use embassy_time::Duration;
-use crate::{audio::DspBuffer, config::LED_COUNT, AudioStats};
+use crate::{audio_process::DspBuffer, config::LED_COUNT, AudioStats};
pub trait LedEffect {
fn render(&mut self, fft: &DspBuffer, stats: &AudioStats, leds: &mut LedColors) -> Duration;
diff --git a/src/led_output.rs b/src/led_output.rs
new file mode 100644
index 0000000..6406fbf
--- /dev/null
+++ b/src/led_output.rs
@@ -0,0 +1,37 @@
+use embassy_sync::{blocking_mutex::raw::NoopRawMutex, zerocopy_channel::Receiver};
+use esp_idf_svc::hal::{gpio::{Gpio11, Gpio12}, spi::SPI2};
+
+use crate::effects::led_effect::LedData;
+
+
+#[embassy_executor::task]
+pub async fn output_leds(
+ spi_per: SPI2,
+ sdo: Gpio11,
+ sck: Gpio12,
+ mut receiver: Receiver<'static, NoopRawMutex, LedData>,
+) {
+ let mut led_drv = spi::SpiDeviceDriver::new_single(
+ spi_per,
+ sck,
+ sdo,
+ AnyIOPin::none(),
+ AnyIOPin::none(),
+ &spi::config::DriverConfig::new(),
+ &spi::config::Config::new()
+ .baudrate(1.MHz().into())
+ .data_mode(spi::config::MODE_3),
+ )
+ .unwrap();
+
+ loop {
+ let led_data = receiver.receive().await;
+
+ let output_buffer = bytes_of(led_data);
+ led_drv
+ .write_async(output_buffer)
+ .await
+ .expect("spi write failed");
+ receiver.receive_done();
+ }
+}
\ No newline at end of file
diff --git a/src/main.rs b/src/main.rs
index 36d211a..0a47f88 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -1,21 +1,16 @@
-pub mod audio;
+pub mod audio_input;
+pub mod audio_process;
pub mod config;
pub mod effects;
pub mod helpers;
+pub mod led_output;
+pub mod triple_buffer;
-use embassy_time::Duration;
use esp_idf_svc::{
- hal::{
- gpio::{AnyIOPin, Gpio11, Gpio12},
- i2s::{self, I2sRx},
- peripherals::Peripherals,
- spi::{self, SPI2},
- units::FromValueType,
- },
+ hal::peripherals::Peripherals,
sys::{esp_dsp, esp_nofail},
};
-use bytemuck::{bytes_of, bytes_of_mut};
use embassy_executor::Spawner;
use embassy_sync::{
blocking_mutex::raw::NoopRawMutex,
@@ -23,12 +18,14 @@ use embassy_sync::{
};
use static_cell::StaticCell;
-use audio::{process_audio, AudioBuffer, AudioStats, DspBuffer};
+use audio_input::mic_input_task;
+use audio_process::{process_audio, AudioBuffer, AudioStats, DspBuffer};
use config::{AUDIO_BUFFERS, AUDIO_SAMPLES_PER_BUF, LED_COUNT};
use effects::{
bass_sparks::LedEffectBassSparks,
led_effect::{LedData, LedEffect},
};
+use led_output::output_leds;
#[embassy_executor::task]
async fn effect_task(
@@ -39,38 +36,6 @@ async fn effect_task(
led_effect.process_led_effect(input, output).await;
}
-#[embassy_executor::task]
-async fn output_leds(
- spi_per: SPI2,
- sdo: Gpio11,
- sck: Gpio12,
- mut receiver: Receiver<'static, NoopRawMutex, LedData>,
-) {
- let mut led_drv = spi::SpiDeviceDriver::new_single(
- spi_per,
- sck,
- sdo,
- AnyIOPin::none(),
- AnyIOPin::none(),
- &spi::config::DriverConfig::new(),
- &spi::config::Config::new()
- .baudrate(1.MHz().into())
- .data_mode(spi::config::MODE_3),
- )
- .unwrap();
-
- loop {
- let led_data = receiver.receive().await;
-
- let output_buffer = bytes_of(led_data);
- led_drv
- .write_async(output_buffer)
- .await
- .expect("spi write failed");
- receiver.receive_done();
- }
-}
-
#[embassy_executor::main]
async fn main(spawner: Spawner) {
// It is necessary to call this function once. Otherwise some patches to the runtime
@@ -111,47 +76,6 @@ async fn main(spawner: Spawner) {
let led_channel = LED_CHANNEL.init(Channel::new(leds));
let (led_sender, led_receiver) = led_channel.split();
- // i2s config
- let mic_i2s_per = peripherals.i2s0;
- let mic_i2s_sd = peripherals.pins.gpio5;
- let mic_i2s_sclk = peripherals.pins.gpio4;
- let mic_i2s_ws = peripherals.pins.gpio6;
-
- let mic_i2s_std_cfg = i2s::config::StdConfig::new(
- i2s::config::Config::new().role(i2s::config::Role::Controller),
- i2s::config::StdClkConfig::new(
- 24000,
- i2s::config::ClockSource::default(),
- i2s::config::MclkMultiple::M256,
- ),
- i2s::config::StdSlotConfig::msb_slot_default(
- i2s::config::DataBitWidth::Bits32,
- i2s::config::SlotMode::Stereo,
- )
- .data_bit_width(i2s::config::DataBitWidth::Bits32)
- .slot_bit_width(i2s::config::SlotBitWidth::Bits32)
- .slot_mode_mask(
- i2s::config::SlotMode::Stereo,
- i2s::config::StdSlotMask::Both,
- )
- .ws_width(32)
- .ws_polarity(false)
- .bit_shift(true)
- .left_align(true)
- .big_endian(false)
- .bit_order_lsb(false),
- i2s::config::StdGpioConfig::new(false, false, false),
- );
- let mut mic_drv = i2s::I2sDriver::new_std_rx(
- mic_i2s_per,
- &mic_i2s_std_cfg,
- mic_i2s_sclk,
- mic_i2s_sd,
- AnyIOPin::none(),
- mic_i2s_ws,
- )
- .unwrap();
-
unsafe {
esp_nofail!(esp_dsp::dsps_fft2r_init_fc32(
std::ptr::null_mut(),
@@ -163,6 +87,15 @@ async fn main(spawner: Spawner) {
));
}
+ spawner
+ .spawn(mic_input_task(
+ peripherals.i2s0,
+ peripherals.pins.gpio5,
+ peripherals.pins.gpio4,
+ peripherals.pins.gpio6,
+ ))
+ .expect("spawn failed");
+
spawner
.spawn(process_audio(audio_receiver, fft_sender))
.expect("spawn failed");
@@ -183,38 +116,4 @@ async fn main(spawner: Spawner) {
led_receiver,
))
.expect("spawn failed");
-
- mic_drv.rx_enable().expect("rx not enabled");
-
- async fn ignore_mic_startup(mic: &mut i2s::I2sDriver<'_, I2sRx>) {
- let mut tmp_buf: [u8; 128] = [0; 128];
- loop {
- mic.read_async(tmp_buf.as_mut_slice()).await.unwrap();
- }
- }
-
- embassy_time::with_timeout(Duration::from_millis(100), ignore_mic_startup(&mut mic_drv))
- .await
- .expect_err("ignore died early");
-
- loop {
- let audio_in_buf: &mut AudioBuffer = audio_sender.send().await;
-
- let buffer = bytes_of_mut(audio_in_buf);
- let mut total_bytes_read: usize = 0;
- let mut remaining_bytes = AUDIO_SAMPLES_PER_BUF * 4;
-
- while total_bytes_read < AUDIO_SAMPLES_PER_BUF * 4 {
- let chunk = &mut buffer[total_bytes_read..total_bytes_read + remaining_bytes];
- let num_bytes_read = mic_drv.read_async(chunk).await.unwrap();
- total_bytes_read += num_bytes_read;
- remaining_bytes -= num_bytes_read;
- }
-
- if total_bytes_read != AUDIO_SAMPLES_PER_BUF * 4 {
- log::warn!("buffer underflow: {}", total_bytes_read);
- }
-
- audio_sender.send_done();
- }
}
diff --git a/src/triple_buffer.rs b/src/triple_buffer.rs
new file mode 100644
index 0000000..295c64d
--- /dev/null
+++ b/src/triple_buffer.rs
@@ -0,0 +1,149 @@
+use core::cell::RefCell;
+use core::future::{poll_fn, Future};
+use core::marker::PhantomData;
+use core::task::Poll;
+
+use embassy_sync::blocking_mutex::{raw::RawMutex, Mutex};
+use embassy_sync::waitqueue::WakerRegistration;
+
+/// Triple-buffered zero-copy Channel.
+///
+/// Sender will flip the input buffer when done, which will in turn
+/// flip the output buffer at the next Receiver call.
+pub struct TripleBuffer<'a, M: RawMutex, T> {
+ buf: *mut T,
+ phantom: PhantomData<&'a mut T>,
+ state: Mutex<M, RefCell<State>>,
+}
+
+impl<'a, M: RawMutex, T> TripleBuffer<'a, M, T> {
+ pub fn new(buffers: &'a mut [T; 3]) -> Self {
+ Self {
+ buf: buffers.as_mut_ptr(),
+ phantom: PhantomData,
+ state: Mutex::new(RefCell::new(State {
+ input: 0,
+ internal: 1,
+ output: None,
+ new_data: false,
+ send_waker: WakerRegistration::new(),
+ })),
+ }
+ }
+
+ /// Creates a [`Sender`] and [`Receiver`] from an existing buffer.
+ ///
+ // / Further Senders and Receivers can be created through [`Sender::borrow`] and
+ // / [`Receiver::borrow`] respectively.
+ pub fn split(&mut self) -> (Sender<'_, M, T>, Receiver<'_, M, T>) {
+ (Sender { buffer: self }, Receiver { buffer: self })
+ }
+}
+
+/// Send-only access to a [`TripleBuffer`].
+pub struct Sender<'a, M: RawMutex, T> {
+ buffer: &'a TripleBuffer<'a, M, T>,
+}
+
+impl<'a, M: RawMutex, T> Sender<'a, M, T> {
+ /// Asynchronously send a value into the buffer.
+ pub fn send(&mut self) -> impl Future<Output = &mut T> {
+ poll_fn(|_| {
+ self.buffer.state.lock(|s| {
+ let s = &mut *s.borrow_mut();
+ let r = unsafe { &mut *self.buffer.buf.add(s.input) };
+ Poll::Ready(r)
+ })
+ })
+ }
+
+ /// Notify the buffer that the sending of the value has been finalized.
+ pub fn send_done(&mut self) {
+ self.buffer.state.lock(|s| s.borrow_mut().flip_input());
+ }
+}
+
+/// Receive-only access to a [`Channel`].
+pub struct Receiver<'a, M: RawMutex, T> {
+ buffer: &'a TripleBuffer<'a, M, T>,
+}
+
+impl<'a, M: RawMutex, T> Receiver<'a, M, T> {
+ /// Checks if the buffer has data available
+ pub fn is_ready(&self) -> bool {
+ self.buffer.state.lock(|s| s.borrow_mut().output.is_some())
+ }
+
+ /// Asynchronously receive, wait for a new value.
+ pub fn receive_new(&mut self) -> impl Future<Output = &mut T> {
+ poll_fn(|cx| {
+ self.buffer.state.lock(|s| {
+ let s = &mut *s.borrow_mut();
+
+ if s.output.is_some() && s.new_data {
+ s.flip_output();
+ let r = unsafe { &mut *self.buffer.buf.add(s.output.unwrap()) };
+ Poll::Ready(r)
+ } else {
+ s.send_waker.register(cx.waker());
+ Poll::Pending
+ }
+ })
+ })
+ }
+
+ /// Asynchronously receive, return with old data if possible.
+ pub fn receive_cached(&mut self) -> impl Future<Output = (&mut T, bool)> {
+ poll_fn(|cx| {
+ self.buffer.state.lock(|s| {
+ let s = &mut *s.borrow_mut();
+
+ if s.output.is_some() {
+ let had_new_data = s.new_data;
+ if s.new_data {
+ s.flip_output();
+ }
+ let r = unsafe { &mut *self.buffer.buf.add(s.output.unwrap()) };
+ Poll::Ready((r, had_new_data))
+ } else {
+ s.send_waker.register(cx.waker());
+ Poll::Pending
+ }
+ })
+ })
+ }
+}
+
+struct State {
+ input: usize,
+ internal: usize,
+ output: Option<usize>,
+ new_data: bool,
+ send_waker: WakerRegistration, // notifies when data was flipped from input to internal
+}
+
+impl State {
+ fn flip_input(&mut self) {
+ (self.input, self.internal) = (self.internal, self.input);
+ self.new_data = true;
+ if self.output.is_none() {
+ self.output = match (self.input, self.internal) {
+ (0, 1) => Some(2),
+ (1, 1) => Some(2),
+ (0, 2) => Some(1),
+ (2, 0) => Some(1),
+ (_, _) => None,
+ }
+ }
+ self.send_waker.wake();
+ }
+ fn flip_output(&mut self) {
+ match self.output {
+ Some(o) => {
+ (self.output, self.internal) = (Some(self.internal), o);
+ self.new_data = false;
+ }
+ None => {}
+ }
+ }
+}