bl/djdeck
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

internal process dsp, skipping ladspa pipewire attempt, and reworked waveforms

Browse Source
This commit is contained in:
Bendik Aagaard Lynghaug
2026-05-27 22:04:08 +02:00
parent f81608734a
commit 39095f3a4a
11 changed files with 1606 additions and 838 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/audio.rs
+443 -171
View File
@@ -3,11 +3,12 @@
/// Responsibilities:
/// - Decode audio files (symphonia) into interleaved stereo f32 PCM
/// - Drive one PipeWire playback stream per deck
/// - Each stream targets the deck's filter-chain sink node
/// (djdeck.chain.<x> input side) so all DSP happens in the graph
/// - Report position, VU, and load status back to the app via channel
///
/// No DSP happens here. Pitch/tempo/EQ all live in the PipeWire graph.
/// - Run the per-deck DSP graph (`crate::dsp::DeckDsp`) inside the realtime
/// process callback: pitch + tempo + cutoff + reverb + fader, all in-process
/// - Stream connects with AUTOCONNECT to the default sink (no filter chains,
/// no `node.target`)
/// - Effect parameters flow lock-free via `Arc<DeckParams>` atomics shared
/// with the app
use std::{
path::PathBuf,
sync::{Arc, Mutex},
@@ -21,13 +22,16 @@ use symphonia::core::{
codecs::DecoderOptions,
formats::FormatOptions,
io::MediaSourceStream,
meta::MetadataOptions,
meta::{MetadataOptions, StandardTagKey},
probe::Hint,
};
use crate::{deck::DeckId, effect_chain::EffectChain};
use crate::{
deck::DeckId,
dsp::{DeckDsp, DeckParams},
};
// ─── Commands ───────────────────────────────────────────────────────────────
// ─── Commands ────────────────────────────────────────────────────────────────
#[derive(Debug)]
pub enum AudioCmd {
@@ -38,124 +42,170 @@ pub enum AudioCmd {
Seek { deck: DeckId, position: f32 },
SetCue(DeckId),
GotoCue(DeckId),
// Volume is handled by wpctl; we just record it in state.
// Fader changes are applied directly from app.rs via EffectChain::set_volume.
/// Append a decoded PCM chunk (interleaved stereo f32) to a deck's buffer.
/// `load_gen` must match the deck's current generation to guard against
/// stale decodes from a previous Load.
AppendPcm { deck: DeckId, load_gen: u64, frames: Vec<f32> },
/// Signal that decoding is complete for this generation.
MarkDecodingDone { deck: DeckId, load_gen: u64 },
}
// ─── Status updates ─────────────────────────────────────────────────────────
// ─── Status updates ──────────────────────────────────────────────────────────
#[derive(Debug, Clone)]
pub enum AudioStatus {
Loaded {
deck: DeckId,
/// Sent immediately from file headers — before full decode.
Metadata {
deck: DeckId,
name: String,
artist: Option<String>,
album: Option<String>,
/// Estimated from codec_params; exact after full decode.
duration_samples: u64,
sample_rate: u32,
},
/// Waveform peak data sent in chunks as the file decodes (~4096-sample chunks).
/// Each entry is `[bass_peak, mid_peak, high_peak]` — peak amplitudes after
/// splitting the mono mix through a Linkwitz-Riley-ish crossover (LP 250 Hz,
/// BP 250–4000 Hz, HP 4000 Hz). The UI blends these into per-column color.
WaveformChunk {
deck: DeckId,
bands: Vec<[f32; 3]>,
/// Sample-frame offset of the first peak in this chunk.
frame_offset: u64,
},
/// BPM estimate from `timestretch::detect_bpm` over the first ~60 s of
/// decoded audio. Emitted once per Load, after `Ready`.
Bpm { deck: DeckId, bpm: f32 },
/// Decoded PCM chunk — routed back through the app into AudioCmd::AppendPcm.
PcmChunk {
deck: DeckId,
load_gen: u64,
frames: Vec<f32>,
sample_rate: u32,
name: String,
/// PipeWire object ID of the stream node (for wpctl volume)
stream_node_id: u32,
},
Position {
deck: DeckId,
position: u64,
},
Vu {
deck: DeckId,
l: f32,
r: f32,
/// Full PCM is in memory; deck is ready to play.
Ready {
deck: DeckId,
load_gen: u64,
duration_samples: u64,
sample_rate: u32,
},
Position { deck: DeckId, position: u64 },
Vu { deck: DeckId, l: f32, r: f32 },
TrackEnded(DeckId),
Error {
deck: DeckId,
msg: String,
},
Error { deck: DeckId, msg: String },
}
// ─── Per-deck internal state ──────────────────────────────────────────────────
struct DeckState {
id: DeckId,
id: DeckId,
playing: bool,
pcm: Vec<f32>, // interleaved stereo f32
/// Play was requested before PCM arrived; start as soon as first chunk lands.
pending_play: bool,
/// True once the decode thread has sent every frame (safe to stop at EOF).
decoding_done: bool,
/// Incremented on each Load; guards AppendPcm from a stale decode thread.
load_gen: u64,
pcm: Vec<f32>,
sample_rate: u32,
read_pos: usize, // byte index into pcm (always even)
cue_pos: usize,
/// Track last VU update time to rate-limit updates
/// Frame index into `pcm` (no fractional resampling — the DSP stretcher
/// handles tempo). Advanced by the number of frames consumed each block.
read_pos: usize,
cue_pos: usize,
dsp: DeckDsp,
params: Arc<DeckParams>,
last_vu_update: std::time::Instant,
}
impl DeckState {
fn new(id: DeckId) -> Self {
fn new(id: DeckId, sample_rate: u32, params: Arc<DeckParams>) -> Self {
Self {
id,
playing: false,
pcm: Vec::new(),
sample_rate: 44100,
read_pos: 0,
cue_pos: 0,
playing: false,
pending_play: false,
decoding_done: true,
load_gen: 0,
pcm: Vec::new(),
sample_rate,
read_pos: 0,
cue_pos: 0,
dsp: DeckDsp::new(sample_rate),
params,
last_vu_update: std::time::Instant::now(),
}
}
fn total_frames(&self) -> usize { self.pcm.len() / 2 }
}
// ─── Public handle ─────────────────────────────────────────────────────
// ─── Public handle ────────────────────────────────────────────────────────────
pub struct AudioEngine {
pub cmd_tx: Sender<AudioCmd>,
pub cmd_tx: Sender<AudioCmd>,
pub status_rx: Receiver<AudioStatus>,
/// One `Arc<DeckParams>` per deck — the app writes effect parameters into
/// these atomics directly. The realtime callback snapshots them per block.
pub params: Vec<Arc<DeckParams>>,
}
impl AudioEngine {
pub fn new() -> Result<Self> {
let (cmd_tx, cmd_rx) = bounded::<AudioCmd>(64);
let (status_tx, status_rx) = bounded::<AudioStatus>(512);
let (cmd_tx, cmd_rx) = bounded::<AudioCmd>(64);
let (status_tx, status_rx) = bounded::<AudioStatus>(1024);
let params: Vec<Arc<DeckParams>> = (0..4).map(|_| DeckParams::new()).collect();
let params_for_engine = params.clone();
thread::Builder::new()
.name("audio-engine".into())
.spawn(move || {
if let Err(e) = engine_thread(cmd_rx, status_tx) {
if let Err(e) = engine_thread(cmd_rx, status_tx, params_for_engine) {
tracing::error!("Audio engine crashed: {e}");
}
})?;
Ok(Self { cmd_tx, status_rx })
Ok(Self { cmd_tx, status_rx, params })
}
}
// ─── Engine thread ──────────────────────────────────────────────────────
// ─── Engine thread ────────────────────────────────────────────────────────────
fn engine_thread(cmd_rx: Receiver<AudioCmd>, status_tx: Sender<AudioStatus>) -> Result<()> {
const STREAM_SR: u32 = 44_100;
fn engine_thread(
cmd_rx: Receiver<AudioCmd>,
status_tx: Sender<AudioStatus>,
params: Vec<Arc<DeckParams>>,
) -> Result<()> {
use pipewire::{context::ContextRc, main_loop::MainLoopRc, stream::StreamRc};
let main_loop = MainLoopRc::new(None)?;
let context = ContextRc::new(&main_loop, None)?;
let core = context.connect_rc(None)?;
let context = ContextRc::new(&main_loop, None)?;
let core = context.connect_rc(None)?;
// Shared per-deck state (written by command thread, read by PW callback)
let deck_states: Vec<Arc<Mutex<DeckState>>> = (0..4)
.map(|i| Arc::new(Mutex::new(DeckState::new(DeckId::from_index(i).unwrap()))))
.map(|i| {
let id = DeckId::from_index(i).unwrap();
Arc::new(Mutex::new(DeckState::new(id, STREAM_SR, params[i].clone())))
})
.collect();
// Build one PipeWire stream per deck
let mut _streams: Vec<StreamRc> = Vec::new();
let mut _listeners = Vec::new();
let mut _streams: Vec<StreamRc> = Vec::new();
let mut _listeners = Vec::new();
for (i, ds) in deck_states.iter().enumerate() {
let deck_id = DeckId::from_index(i).unwrap();
let ds_cb = Arc::clone(ds);
let deck_id = DeckId::from_index(i).unwrap();
let ds_cb = Arc::clone(ds);
let status_tx_cb = status_tx.clone();
// Target: the input (sink) side of this deck's filter-chain node
let target_name = format!("input.{}", EffectChain::node_name_for(deck_id));
let props = pipewire::properties::properties! {
*pipewire::keys::MEDIA_TYPE => "Audio",
*pipewire::keys::MEDIA_CATEGORY => "Playback",
*pipewire::keys::MEDIA_ROLE => "Music",
*pipewire::keys::APP_NAME => "djdeck",
*pipewire::keys::NODE_NAME => format!("djdeck.deck.{}", deck_id.node_suffix()),
// Route to our filter-chain sink
"node.target" => target_name,
};
let stream = StreamRc::new(
@@ -172,41 +222,47 @@ fn engine_thread(cmd_rx: Receiver<AudioCmd>, status_tx: Sender<AudioStatus>) ->
if let Some(mut buf) = stream.dequeue_buffer() {
let datas = buf.datas_mut();
if datas.is_empty() {
return;
}
if datas.is_empty() { return; }
let data = &mut datas[0];
let out_bytes = data.data().unwrap();
let n_frames = out_bytes.len() / 8; // f32 stereo
let n_frames = out_bytes.len() / 8;
let out = unsafe {
std::slice::from_raw_parts_mut(out_bytes.as_ptr() as *mut f32, n_frames * 2)
std::slice::from_raw_parts_mut(
out_bytes.as_ptr() as *mut f32,
n_frames * 2,
)
};
let mut vu_l = 0.0f32;
let mut vu_r = 0.0f32;
if ds.playing && !ds.pcm.is_empty() {
for frame in 0..n_frames {
if ds.read_pos + 1 >= ds.pcm.len() {
let total = ds.total_frames();
if ds.read_pos >= total {
if ds.decoding_done {
ds.playing = false;
out[frame * 2] = 0.0;
out[frame * 2 + 1] = 0.0;
let _ = status_tx_cb.try_send(AudioStatus::TrackEnded(ds.id));
continue;
}
out.fill(0.0);
} else {
// Split disjoint field borrows so the DSP can read
// `pcm` while mutating its internal state.
let start = ds.read_pos;
let DeckState { dsp, pcm, params, .. } = &mut **ds;
let consumed = dsp.process(params, pcm, start, out);
ds.read_pos = (start + consumed).min(total);
// VU from the post-DSP output.
let mut vu_l = 0.0f32;
let mut vu_r = 0.0f32;
for i in 0..n_frames {
vu_l = vu_l.max(out[i * 2].abs());
vu_r = vu_r.max(out[i * 2 + 1].abs());
}
let l = ds.pcm[ds.read_pos];
let r = ds.pcm[ds.read_pos + 1];
ds.read_pos += 2;
out[frame * 2] = l;
out[frame * 2 + 1] = r;
vu_l = vu_l.max(l.abs());
vu_r = vu_r.max(r.abs());
let now = std::time::Instant::now();
if now.duration_since(ds.last_vu_update) >= std::time::Duration::from_millis(16) {
let _ = status_tx_cb.try_send(AudioStatus::Vu { deck: ds.id, l: vu_l, r: vu_r });
ds.last_vu_update = now;
}
}
} else {
out.fill(0.0);
@@ -216,29 +272,17 @@ fn engine_thread(cmd_rx: Receiver<AudioCmd>, status_tx: Sender<AudioStatus>) ->
let chunk = data.chunk_mut();
*chunk.offset_mut() = 0;
*chunk.stride_mut() = 8;
*chunk.size_mut() = (n_frames * 8) as u32;
*chunk.size_mut() = (n_frames * 8) as u32;
}
let pos = (ds.read_pos / 2) as u64;
let _ = status_tx_cb.try_send(AudioStatus::Position {
deck: ds.id,
position: pos,
position: ds.read_pos as u64,
});
// Rate-limit VU updates to ~60Hz (16ms) to reduce CPU usage
// while still providing smooth updates during playback
let now = std::time::Instant::now();
if now.duration_since(ds.last_vu_update) >= std::time::Duration::from_millis(16) {
let _ = status_tx_cb.try_send(AudioStatus::Vu {
deck: ds.id,
l: vu_l,
r: vu_r,
});
ds.last_vu_update = now;
}
}
})
.register()?; // Connect the stream with F32LE stereo format
.register()?;
let pod_bytes = make_f32_stereo_pod()?;
let mut params = [pipewire::spa::pod::Pod::from_bytes(&pod_bytes).unwrap()];
stream.connect(
@@ -254,7 +298,7 @@ fn engine_thread(cmd_rx: Receiver<AudioCmd>, status_tx: Sender<AudioStatus>) ->
_streams.push(stream);
}
// Command processing thread — decodes files and mutates deck states
// Command processing thread
let ds_cmd = deck_states.clone();
let st_cmd = status_tx.clone();
thread::spawn(move || {
@@ -267,63 +311,52 @@ fn engine_thread(cmd_rx: Receiver<AudioCmd>, status_tx: Sender<AudioStatus>) ->
Ok(())
}
// ─── Command handler ──────────────────────────────────────────────────────────
fn handle_cmd(cmd: AudioCmd, states: &[Arc<Mutex<DeckState>>], status_tx: &Sender<AudioStatus>) {
match cmd {
AudioCmd::Load { deck, path } => {
let name = path
.file_name()
.and_then(|n| n.to_str())
.unwrap_or("unknown")
.to_string();
match decode_file(&path) {
Ok((pcm, sr)) => {
let total = (pcm.len() / 2) as u64;
{
let mut ds = states[deck as usize].lock().unwrap();
ds.pcm = pcm;
ds.sample_rate = sr;
ds.read_pos = 0;
ds.cue_pos = 0;
ds.playing = false;
}
// stream_node_id: we don't have it from here — the app queries
// pw-dump after connecting. Emit 0 as placeholder; app fills it in.
let _ = status_tx.send(AudioStatus::Loaded {
deck,
duration_samples: total,
sample_rate: sr,
name,
stream_node_id: 0,
});
let load_gen = {
let mut ds = states[deck as usize].lock().unwrap();
ds.pcm.clear();
ds.read_pos = 0;
ds.cue_pos = 0;
ds.playing = false;
ds.pending_play = false;
ds.decoding_done = false;
ds.load_gen = ds.load_gen.wrapping_add(1);
ds.load_gen
};
let st = status_tx.clone();
thread::spawn(move || {
if let Err(e) = decode_file_streaming(&path, deck, load_gen, &st) {
let _ = st.send(AudioStatus::Error { deck, msg: e.to_string() });
}
Err(e) => {
let _ = status_tx.send(AudioStatus::Error {
deck,
msg: e.to_string(),
});
}
}
});
}
AudioCmd::Play(deck) => {
let mut ds = states[deck as usize].lock().unwrap();
if !ds.pcm.is_empty() {
ds.playing = true;
} else {
ds.pending_play = true;
}
}
AudioCmd::Pause(deck) => {
let mut ds = states[deck as usize].lock().unwrap();
ds.playing = false;
ds.playing = false;
ds.pending_play = false;
}
AudioCmd::Stop(deck) => {
let mut ds = states[deck as usize].lock().unwrap();
ds.playing = false;
ds.read_pos = 0;
ds.playing = false;
ds.pending_play = false;
ds.read_pos = 0;
}
AudioCmd::Seek { deck, position } => {
let mut ds = states[deck as usize].lock().unwrap();
let total = ds.pcm.len();
ds.read_pos = ((total as f32 * position.clamp(0.0, 1.0)) as usize / 2) * 2;
let total = ds.total_frames();
ds.read_pos = ((total as f32 * position.clamp(0.0, 1.0)) as usize).min(total);
}
AudioCmd::SetCue(deck) => {
let mut ds = states[deck as usize].lock().unwrap();
@@ -333,15 +366,46 @@ fn handle_cmd(cmd: AudioCmd, states: &[Arc<Mutex<DeckState>>], status_tx: &Sende
let mut ds = states[deck as usize].lock().unwrap();
ds.read_pos = ds.cue_pos;
}
AudioCmd::AppendPcm { deck, load_gen, frames } => {
let mut ds = states[deck as usize].lock().unwrap();
if ds.load_gen != load_gen { return; } // stale decode
let was_empty = ds.pcm.is_empty();
ds.pcm.extend_from_slice(&frames);
if was_empty && ds.pending_play {
ds.playing = true;
ds.pending_play = false;
}
}
AudioCmd::MarkDecodingDone { deck, load_gen } => {
let mut ds = states[deck as usize].lock().unwrap();
if ds.load_gen == load_gen {
ds.decoding_done = true;
}
}
}
}
// ─── Symphonia decode ─────────────────────────────────────────────────────────────
// ─── Symphonia streaming decode ───────────────────────────────────────────────
/// Number of PCM stereo frames per waveform peak bucket (~0.09 s at 44.1 kHz).
const PEAK_BUCKET: usize = 4096;
/// Open file, emit Metadata immediately, then decode to PCM while streaming
/// both waveform peaks and raw PCM chunks every 64 peak-buckets (~6 s).
fn decode_file_streaming(
path: &PathBuf,
deck: DeckId,
load_gen: u64,
status_tx: &Sender<AudioStatus>,
) -> Result<()> {
let name = path
.file_name()
.and_then(|n| n.to_str())
.unwrap_or("unknown")
.to_string();
/// Decode any supported audio file into interleaved stereo f32 PCM.
fn decode_file(path: &PathBuf) -> Result<(Vec<f32>, u32)> {
let file = std::fs::File::open(path)?;
let mss = MediaSourceStream::new(Box::new(file), Default::default());
let mss = MediaSourceStream::new(Box::new(file), Default::default());
let mut hint = Hint::new();
if let Some(ext) = path.extension().and_then(|e| e.to_str()) {
@@ -356,6 +420,7 @@ fn decode_file(path: &PathBuf) -> Result<(Vec<f32>, u32)> {
)?;
let mut format = probed.format;
let track = format
.tracks()
.iter()
@@ -363,33 +428,218 @@ fn decode_file(path: &PathBuf) -> Result<(Vec<f32>, u32)> {
.ok_or_else(|| anyhow!("No audio track found"))?
.clone();
let sr = track.codec_params.sample_rate.unwrap_or(44100);
let sr = track.codec_params.sample_rate.unwrap_or(44100);
let channels = track.codec_params.channels.map(|c| c.count()).unwrap_or(2);
let mut decoder =
symphonia::default::get_codecs().make(&track.codec_params, &DecoderOptions::default())?;
let est_duration = track.codec_params.n_frames.unwrap_or(0);
let (artist, album) = extract_tags(&mut format);
let _ = status_tx.send(AudioStatus::Metadata {
deck,
name,
artist,
album,
duration_samples: est_duration,
sample_rate: sr,
});
let mut decoder = symphonia::default::get_codecs()
.make(&track.codec_params, &DecoderOptions::default())?;
let track_id = track.id;
let mut pcm: Vec<f32> = Vec::new();
let mut pcm_batch: Vec<f32> = Vec::with_capacity(64 * PEAK_BUCKET * 2);
let mut total_frames: u64 = 0;
// Mono accumulator used for BPM detection. Capped at 60 s.
const BPM_MAX_SAMPLES_SECS: u64 = 60;
let bpm_cap = (sr as u64 * BPM_MAX_SAMPLES_SECS) as usize;
let mut mono_for_bpm: Vec<f32> = Vec::with_capacity(bpm_cap.min(2_700_000));
let mut wf = WaveformAccum::new(sr);
loop {
let packet = match format.next_packet() {
Ok(p) => p,
Err(symphonia::core::errors::Error::IoError(_)) => break,
Err(symphonia::core::errors::Error::ResetRequired) => break,
Err(symphonia::core::errors::Error::IoError(_)) => break,
Err(symphonia::core::errors::Error::ResetRequired) => break,
Err(e) => return Err(e.into()),
};
if packet.track_id() != track_id {
continue;
}
if packet.track_id() != track_id { continue; }
let decoded = decoder.decode(&packet)?;
push_samples(&mut pcm, decoded, channels);
push_samples_and_peaks(
&mut pcm_batch,
&mut total_frames,
decoded,
channels,
&mut wf,
&mut mono_for_bpm,
bpm_cap,
deck,
load_gen,
sr,
status_tx,
);
}
Ok((pcm, sr))
wf.flush(&mut |_| {});
if !wf.bands_batch.is_empty() {
let _ = status_tx.send(AudioStatus::WaveformChunk {
deck,
bands: std::mem::take(&mut wf.bands_batch),
frame_offset: wf.bucket_start_frame,
});
}
if !pcm_batch.is_empty() {
let _ = status_tx.send(AudioStatus::PcmChunk {
deck, load_gen,
frames: pcm_batch,
sample_rate: sr,
});
}
let _ = status_tx.send(AudioStatus::Ready {
deck,
load_gen,
duration_samples: total_frames,
sample_rate: sr,
});
// BPM detection on the mono accumulator. Runs synchronously in the decode
// thread, so it doesn't compete with the realtime callback. ~hundreds of
// ms for a 60 s clip; the Ready status above means the track is already
// playable before this fires.
if !mono_for_bpm.is_empty() {
let bpm = timestretch::detect_bpm(&mono_for_bpm, sr);
if bpm.is_finite() && bpm > 0.0 {
let _ = status_tx.send(AudioStatus::Bpm { deck, bpm: bpm as f32 });
}
}
Ok(())
}
fn push_samples(out: &mut Vec<f32>, buf: AudioBufferRef, channels: usize) {
// ─── Per-decode waveform accumulator ──────────────────────────────────────────
//
// Splits the mono mix into bass / mid / high via biquads, then takes the peak
// per band over PEAK_BUCKET frames. Each `bands_batch` entry is one bucket
// `[bass, mid, high]`. Sent off every 64 buckets (~6 s) as a `WaveformChunk`.
struct WaveformAccum {
bass_lp: biquad::DirectForm1<f32>,
mid_bp: biquad::DirectForm1<f32>,
high_hp: biquad::DirectForm1<f32>,
bucket_bass: f32,
bucket_mid: f32,
bucket_high: f32,
bucket_count: usize,
bands_batch: Vec<[f32; 3]>,
frame_offset: u64,
bucket_start_frame: u64,
}
impl WaveformAccum {
fn new(sample_rate: u32) -> Self {
use biquad::{Coefficients, DirectForm1, Hertz, Q_BUTTERWORTH_F32, Type};
let fs = Hertz::<f32>::from_hz(sample_rate as f32).unwrap();
let lp_co = Coefficients::<f32>::from_params(
Type::LowPass, fs, Hertz::<f32>::from_hz(250.0).unwrap(), Q_BUTTERWORTH_F32
).unwrap();
let bp_co = Coefficients::<f32>::from_params(
Type::BandPass, fs, Hertz::<f32>::from_hz(1500.0).unwrap(), 0.7
).unwrap();
let hp_co = Coefficients::<f32>::from_params(
Type::HighPass, fs, Hertz::<f32>::from_hz(4000.0).unwrap(), Q_BUTTERWORTH_F32
).unwrap();
Self {
bass_lp: DirectForm1::<f32>::new(lp_co),
mid_bp: DirectForm1::<f32>::new(bp_co),
high_hp: DirectForm1::<f32>::new(hp_co),
bucket_bass: 0.0,
bucket_mid: 0.0,
bucket_high: 0.0,
bucket_count: 0,
bands_batch: Vec::with_capacity(64),
frame_offset: 0,
bucket_start_frame: 0,
}
}
#[inline]
fn push_sample(&mut self, mono: f32) {
use biquad::Biquad as _;
let b = self.bass_lp.run(mono).abs();
let m = self.mid_bp .run(mono).abs();
let h = self.high_hp.run(mono).abs();
if b > self.bucket_bass { self.bucket_bass = b; }
if m > self.bucket_mid { self.bucket_mid = m; }
if h > self.bucket_high { self.bucket_high = h; }
self.bucket_count += 1;
self.frame_offset += 1;
if self.bucket_count >= PEAK_BUCKET {
// Compress: sqrt expands quiet detail.
self.bands_batch.push([
self.bucket_bass.sqrt(),
self.bucket_mid.sqrt(),
self.bucket_high.sqrt(),
]);
self.bucket_bass = 0.0;
self.bucket_mid = 0.0;
self.bucket_high = 0.0;
self.bucket_count = 0;
}
}
/// Flush any in-progress bucket if non-empty.
fn flush(&mut self, _emit: &mut dyn FnMut(&[[f32; 3]])) {
if self.bucket_count > 0 {
self.bands_batch.push([
self.bucket_bass.sqrt(),
self.bucket_mid.sqrt(),
self.bucket_high.sqrt(),
]);
self.bucket_bass = 0.0;
self.bucket_mid = 0.0;
self.bucket_high = 0.0;
self.bucket_count = 0;
}
}
}
fn extract_tags(format: &mut Box<dyn symphonia::core::formats::FormatReader>) -> (Option<String>, Option<String>) {
let mut artist = None;
let mut album = None;
if let Some(meta) = format.metadata().current() {
for tag in meta.tags() {
match tag.std_key {
Some(StandardTagKey::Artist) => artist = Some(tag.value.to_string()),
Some(StandardTagKey::AlbumArtist) if artist.is_none()
=> artist = Some(tag.value.to_string()),
Some(StandardTagKey::Album) => album = Some(tag.value.to_string()),
_ => {}
}
}
}
(artist, album)
}
#[allow(clippy::too_many_arguments)]
fn push_samples_and_peaks(
pcm_batch: &mut Vec<f32>,
total_frames: &mut u64,
buf: AudioBufferRef,
channels: usize,
wf: &mut WaveformAccum,
mono_for_bpm: &mut Vec<f32>,
bpm_cap: usize,
deck: DeckId,
load_gen: u64,
sample_rate: u32,
status_tx: &Sender<AudioStatus>,
) {
macro_rules! push {
($buf:expr, $scale:expr) => {{
let frames = $buf.frames();
@@ -397,11 +647,35 @@ fn push_samples(out: &mut Vec<f32>, buf: AudioBufferRef, channels: usize) {
let l = *$buf.chan(0).get(f).unwrap_or(&Default::default());
let r = if channels >= 2 {
*$buf.chan(1).get(f).unwrap_or(&Default::default())
} else {
l
};
out.push(l as f32 * $scale);
out.push(r as f32 * $scale);
} else { l };
let lf = l as f32 * $scale;
let rf = r as f32 * $scale;
pcm_batch.push(lf);
pcm_batch.push(rf);
*total_frames += 1;
let mono = (lf + rf) * 0.5;
wf.push_sample(mono);
if mono_for_bpm.len() < bpm_cap {
mono_for_bpm.push(mono);
}
if wf.bands_batch.len() >= 64 {
let _ = status_tx.send(AudioStatus::WaveformChunk {
deck,
bands: std::mem::take(&mut wf.bands_batch),
frame_offset: wf.bucket_start_frame,
});
let _ = status_tx.send(AudioStatus::PcmChunk {
deck,
load_gen,
frames: std::mem::take(pcm_batch),
sample_rate,
});
wf.bucket_start_frame = wf.frame_offset;
}
}
}};
}
@@ -410,13 +684,13 @@ fn push_samples(out: &mut Vec<f32>, buf: AudioBufferRef, channels: usize) {
AudioBufferRef::F32(b) => push!(b, 1.0_f32),
AudioBufferRef::F64(b) => push!(b, 1.0_f32),
AudioBufferRef::S16(b) => push!(b, 1.0 / 32768.0_f32),
AudioBufferRef::S32(b) => push!(b, 1.0 / 2147483648.0_f32),
AudioBufferRef::U8(b) => push!(b, 1.0 / 128.0_f32),
AudioBufferRef::S32(b) => push!(b, 1.0 / 2_147_483_648.0_f32),
AudioBufferRef::U8(b) => push!(b, 1.0 / 128.0_f32),
_ => {}
}
}
// ─── PipeWire format pod helper ───────────────────────────────────────────────
// ─── PipeWire format pod ──────────────────────────────────────────────────────
fn make_f32_stereo_pod() -> Result<Vec<u8>> {
use pipewire::spa::param::ParamType;
@@ -431,15 +705,13 @@ fn make_f32_stereo_pod() -> Result<Vec<u8>> {
&Value::Object(object!(
SpaTypes::ObjectParamFormat,
ParamType::EnumFormat,
property!(FormatProperties::MediaType, Id, MediaType::Audio),
property!(FormatProperties::MediaType, Id, MediaType::Audio),
property!(FormatProperties::MediaSubtype, Id, MediaSubtype::Raw),
property!(FormatProperties::AudioFormat, Id, AudioFormat::F32LE),
property!(FormatProperties::AudioRate, Int, 44100i32),
property!(FormatProperties::AudioFormat, Id, AudioFormat::F32LE),
property!(FormatProperties::AudioRate, Int, STREAM_SR as i32),
property!(FormatProperties::AudioChannels, Int, 2i32),
)),
)?
.0
.into_inner();
)?.0.into_inner();
Ok(bytes)
}