Krolyxon
Commit 50918aa

Branches

Tags

Refs →

rmptuisrcvisualizermod.rs

251 lines · 8.2 KB · Rust

Raw Blame History Download
1//! Reads PCM data from MPD's fifo output and computes spectra.
2//!
3//! A dedicated thread does the I/O and FFT while the visualizer is visible
4//! and publishes each frame (recent samples plus their spectrum) on a `watch`
5//! channel; the UI draws waves from the samples and bars from the spectrum
6//! (see [`bars`]).
7
8mod bars;
9
10use std::{
11 fs::{File, OpenOptions},
12 io::{self, Read},
13 sync::{
14 Arc,
15 atomic::{AtomicBool, Ordering},
16 },
17 thread,
18 time::{Duration, Instant},
19};
20
21pub use bars::Bars;
22use realfft::{RealFftPlanner, RealToComplex};
23use tokio::sync::watch;
24use tracing::warn;
25
26use crate::config::VisualizerConfig;
27
28const FFT_SIZE: usize = 2048;
29/// Samples per channel kept for the wave styles.
30const WAVE_SAMPLES: usize = 1024;
31
32/// One analysis frame.
33#[derive(Clone, Debug, Default)]
34pub struct AudioFrame {
35 /// Normalized magnitude per FFT bin (0 Hz up to Nyquist).
36 pub magnitudes: Vec<f32>,
37 /// Width of one bin in Hz.
38 pub bin_hz: f32,
39 /// The most recent samples of each channel in `-1.0..=1.0`, oldest
40 /// first. Both hold the same data for mono input.
41 pub left: Vec<f32>,
42 pub right: Vec<f32>,
43 /// Why there is no audio, if the fifo cannot be read.
44 pub error: Option<String>,
45}
46
47pub struct Visualizer {
48 active: Arc<AtomicBool>,
49 spectrum: watch::Receiver<Arc<AudioFrame>>,
50}
51
52impl Visualizer {
53 pub fn spawn(config: VisualizerConfig) -> Self {
54 let active = Arc::new(AtomicBool::new(false));
55 let (tx, spectrum) = watch::channel(Arc::new(AudioFrame::default()));
56 let flag = Arc::clone(&active);
57 let spawned = thread::Builder::new()
58 .name("visualizer".into())
59 .spawn(move || Reader::new(config, flag, tx).run());
60 if let Err(error) = spawned {
61 warn!("cannot start visualizer thread: {error}");
62 }
63 Self { active, spectrum }
64 }
65
66 /// Starts or stops reading the fifo.
67 pub fn set_active(&self, active: bool) {
68 self.active.store(active, Ordering::Relaxed);
69 }
70
71 pub fn is_active(&self) -> bool {
72 self.active.load(Ordering::Relaxed)
73 }
74
75 pub fn latest(&self) -> Arc<AudioFrame> {
76 Arc::clone(&self.spectrum.borrow())
77 }
78
79 /// A receiver to await new spectra on.
80 pub fn subscribe(&self) -> watch::Receiver<Arc<AudioFrame>> {
81 self.spectrum.clone()
82 }
83}
84
85struct Reader {
86 config: VisualizerConfig,
87 active: Arc<AtomicBool>,
88 tx: watch::Sender<Arc<AudioFrame>>,
89 fft: Arc<dyn RealToComplex<f32>>,
90 window: Vec<f32>,
91 /// The most recent mono samples, oldest first.
92 samples: Vec<f32>,
93 left: Vec<f32>,
94 right: Vec<f32>,
95}
96
97impl Reader {
98 fn new(
99 config: VisualizerConfig,
100 active: Arc<AtomicBool>,
101 tx: watch::Sender<Arc<AudioFrame>>,
102 ) -> Self {
103 let window = (0..FFT_SIZE)
104 .map(|i| {
105 let phase = 2.0 * std::f32::consts::PI * i as f32 / (FFT_SIZE - 1) as f32;
106 0.5 - 0.5 * phase.cos()
107 })
108 .collect();
109 Self {
110 fft: RealFftPlanner::<f32>::new().plan_fft_forward(FFT_SIZE),
111 config,
112 active,
113 tx,
114 window,
115 samples: vec![0.0; FFT_SIZE],
116 left: vec![0.0; WAVE_SAMPLES],
117 right: vec![0.0; WAVE_SAMPLES],
118 }
119 }
120
121 fn run(mut self) {
122 loop {
123 while !self.active.load(Ordering::Relaxed) {
124 if self.tx.is_closed() {
125 return;
126 }
127 thread::sleep(Duration::from_millis(100));
128 }
129 match open_fifo(&self.config) {
130 Ok(file) => self.read_while_active(file),
131 Err(error) => {
132 self.publish(AudioFrame {
133 error: Some(format!(
134 "cannot open {}: {error}",
135 self.config.fifo.display()
136 )),
137 ..AudioFrame::default()
138 });
139 thread::sleep(Duration::from_secs(1));
140 }
141 }
142 }
143 }
144
145 fn read_while_active(&mut self, mut file: File) {
146 let channels = usize::from(self.config.channels);
147 let frame_bytes = 2 * channels;
148 let interval = Duration::from_secs_f64(1.0 / f64::from(self.config.framerate));
149 let mut buf = vec![0u8; 16 * 1024];
150 let mut pending: Vec<u8> = Vec::new();
151 let mut last_frame = Instant::now();
152 let mut last_data = Instant::now();
153 let mut silent = false;
154
155 // Drop audio that piled up while nobody was reading.
156 while matches!(file.read(&mut buf), Ok(n) if n > 0) {}
157
158 while self.active.load(Ordering::Relaxed) {
159 match file.read(&mut buf) {
160 Ok(0) => thread::sleep(Duration::from_millis(10)),
161 Ok(n) => {
162 pending.extend_from_slice(&buf[..n]);
163 let usable = pending.len() - pending.len() % frame_bytes;
164 for frame in pending[..usable].chunks_exact(frame_bytes) {
165 let sample = |i: usize| {
166 let s = &frame[2 * i..2 * i + 2];
167 f32::from(i16::from_le_bytes([s[0], s[1]])) / f32::from(i16::MAX)
168 };
169 let sum: f32 = (0..channels).map(sample).sum();
170 self.samples.push(sum / channels as f32);
171 self.left.push(sample(0));
172 self.right.push(sample(channels.min(2) - 1));
173 }
174 pending.drain(..usable);
175 keep_last(&mut self.samples, FFT_SIZE);
176 keep_last(&mut self.left, WAVE_SAMPLES);
177 keep_last(&mut self.right, WAVE_SAMPLES);
178 last_data = Instant::now();
179 silent = false;
180 }
181 Err(error) if error.kind() == io::ErrorKind::WouldBlock => {
182 thread::sleep(Duration::from_millis(4));
183 }
184 Err(error) if error.kind() == io::ErrorKind::Interrupted => {}
185 Err(error) => {
186 warn!("reading visualizer fifo failed: {error}");
187 return;
188 }
189 }
190 if last_frame.elapsed() >= interval {
191 last_frame = Instant::now();
192 if last_data.elapsed() > Duration::from_millis(150) {
193 // Paused or stopped: publish silence once so bars fall.
194 if !silent {
195 silent = true;
196 self.samples.fill(0.0);
197 self.left.fill(0.0);
198 self.right.fill(0.0);
199 self.analyze();
200 }
201 } else {
202 self.analyze();
203 }
204 }
205 }
206 }
207
208 fn analyze(&mut self) {
209 let mut input: Vec<f32> = self
210 .samples
211 .iter()
212 .zip(&self.window)
213 .map(|(s, w)| s * w)
214 .collect();
215 let mut output = self.fft.make_output_vec();
216 if self.fft.process(&mut input, &mut output).is_err() {
217 return;
218 }
219 // A full-scale sine yields |X| = N/4 with a Hann window.
220 let scale = 4.0 / FFT_SIZE as f32;
221 let magnitudes = output.iter().map(|c| c.norm() * scale).collect();
222 self.publish(AudioFrame {
223 magnitudes,
224 bin_hz: self.config.sample_rate as f32 / FFT_SIZE as f32,
225 left: self.left.clone(),
226 right: self.right.clone(),
227 error: None,
228 });
229 }
230
231 fn publish(&self, spectrum: AudioFrame) {
232 let _ = self.tx.send(Arc::new(spectrum));
233 }
234}
235
236fn keep_last(samples: &mut Vec<f32>, len: usize) {
237 let excess = samples.len().saturating_sub(len);
238 samples.drain(..excess);
239}
240
241fn open_fifo(config: &VisualizerConfig) -> io::Result<File> {
242 let mut options = OpenOptions::new();
243 options.read(true);
244 #[cfg(unix)]
245 {
246 use std::os::unix::fs::OpenOptionsExt;
247 // Never block, whether or not MPD has the fifo open for writing.
248 options.custom_flags(libc::O_NONBLOCK);
249 }
250 options.open(&config.fifo)
251}