1use std::ffi::OsString;
2use std::io::Read;
3use std::path::PathBuf;
4use std::process::{Child, Command, Stdio};
5use std::sync::Arc;
6use std::sync::atomic::{AtomicBool, Ordering};
7use std::time::Duration;
8
9use crate::mix::PcmQueue;
10
11pub const AHEAD_SECONDS: u32 = 2;
12
13pub fn ahead_limit(rate: u32, channels: u16) -> usize {
14 (rate as usize)
15 .saturating_mul(channels as usize)
16 .saturating_mul(AHEAD_SECONDS as usize)
17}
18
19#[derive(Clone, Debug)]
20pub struct DecodeSpec {
21 pub ffmpeg: PathBuf,
22 pub media: PathBuf,
23 pub start: f32,
24 pub length: f32,
25 pub rate: u32,
26 pub channels: u16,
27 pub tracks: Vec<(i64, f32)>,
28}
29
30#[derive(Clone, Debug)]
31pub struct CommandLine {
32 pub program: PathBuf,
33 pub args: Vec<OsString>,
34}
35
36pub fn mix_graph(tracks: &[(i64, f32)]) -> String {
37 use std::fmt::Write;
38 let mut graph = String::new();
39 for (k, (stream, volume)) in tracks.iter().enumerate() {
40 write!(
41 graph,
42 "[0:{stream}]volume={volume:.3},aformat=channel_layouts=stereo[t{k}];"
43 )
44 .unwrap();
45 }
46 let inputs: String = (0..tracks.len()).map(|k| format!("[t{k}]")).collect();
47 write!(
48 graph,
49 "{inputs}amix=inputs={}:normalize=0[mix]",
50 tracks.len()
51 )
52 .unwrap();
53 graph
54}
55
56pub fn command_line(spec: &DecodeSpec) -> CommandLine {
57 let mut args = Vec::new();
58 push(&mut args, "-nostdin");
59 push(&mut args, "-v");
60 push(&mut args, "error");
61 push(&mut args, "-ss");
62 push(&mut args, &format!("{:.3}", spec.start.max(0.0)));
63 push(&mut args, "-t");
64 push(&mut args, &format!("{:.3}", spec.length.max(0.0)));
65 push(&mut args, "-i");
66 args.push(spec.media.as_os_str().to_os_string());
67 if !spec.tracks.is_empty() {
68 push(&mut args, "-filter_complex");
69 push(&mut args, &mix_graph(&spec.tracks));
70 push(&mut args, "-map");
71 push(&mut args, "[mix]");
72 }
73 push(&mut args, "-f");
74 push(&mut args, "f32le");
75 push(&mut args, "-ac");
76 push(&mut args, &decode_channels(spec).to_string());
77 push(&mut args, "-ar");
78 push(&mut args, &spec.rate.to_string());
79 push(&mut args, "-");
80 CommandLine {
81 program: spec.ffmpeg.clone(),
82 args,
83 }
84}
85
86pub fn decode_channels(spec: &DecodeSpec) -> u16 {
87 if spec.tracks.is_empty() {
88 spec.channels
89 } else {
90 2
91 }
92}
93
94pub(crate) fn spec_at_output(spec: &DecodeSpec, rate: u32, channels: u16) -> DecodeSpec {
95 let mut spec = spec.clone();
96 spec.rate = rate;
97 spec.channels = if spec.tracks.is_empty() { channels } else { 2 };
98 spec
99}
100
101fn push(args: &mut Vec<OsString>, value: &str) {
102 args.push(OsString::from(value));
103}
104
105pub struct Decode {
106 stop: Arc<AtomicBool>,
107 child: Option<Child>,
108 queue: Arc<PcmQueue>,
109 channels: u16,
110 rate: u32,
111}
112
113impl Decode {
114 pub fn start(spec: &DecodeSpec) -> Result<Decode, String> {
115 if !spec.ffmpeg.is_absolute() {
116 return Err("ffmpeg path must be absolute".into());
117 }
118 let channels = decode_channels(spec);
119 if spec.rate == 0 || channels == 0 {
120 return Err("decode rate and channels must be non-zero".into());
121 }
122 let line = command_line(spec);
123 let mut command = Command::new(&line.program);
124 command
125 .args(&line.args)
126 .stdin(Stdio::null())
127 .stdout(Stdio::piped())
128 .stderr(Stdio::null());
129 let mut child = command.spawn().map_err(|e| format!("ffmpeg: {e}"))?;
130 let Some(stdout) = child.stdout.take() else {
131 let _ = child.kill();
132 let _ = child.wait();
133 return Err("ffmpeg returned no audio".into());
134 };
135 let queue = PcmQueue::new();
136 let stop = Arc::new(AtomicBool::new(false));
137 let thread_queue = Arc::clone(&queue);
138 let thread_stop = Arc::clone(&stop);
139 let rate = spec.rate;
140 std::thread::spawn(move || pump(stdout, thread_queue, thread_stop, rate, channels));
141 Ok(Decode {
142 stop,
143 child: Some(child),
144 queue,
145 channels,
146 rate,
147 })
148 }
149
150 pub fn rate(&self) -> u32 {
151 self.rate
152 }
153
154 pub fn channels(&self) -> u16 {
155 self.channels
156 }
157
158 pub fn ended(&self) -> bool {
159 self.queue.reader_ended()
160 }
161
162 pub fn buffered(&self) -> usize {
163 self.queue.len()
164 }
165
166 pub(crate) fn queue(&self) -> Arc<PcmQueue> {
167 Arc::clone(&self.queue)
168 }
169}
170
171impl Drop for Decode {
172 fn drop(&mut self) {
173 self.stop.store(true, Ordering::Relaxed);
174 if let Some(mut child) = self.child.take() {
175 let _ = child.kill();
176 let _ = child.wait();
177 }
178 }
179}
180
181fn pump(
182 mut stdout: impl Read,
183 queue: Arc<PcmQueue>,
184 stop: Arc<AtomicBool>,
185 rate: u32,
186 channels: u16,
187) {
188 let mut framer = LeF32::new();
189 let mut chunk = vec![0u8; 16_384];
190 while !stop.load(Ordering::Relaxed) {
191 if ahead_is_full(queue.len(), rate, channels) {
192 std::thread::sleep(Duration::from_millis(10));
193 continue;
194 }
195 match stdout.read(&mut chunk) {
196 Ok(0) | Err(_) => break,
197 Ok(n) => {
198 let mut decoded = Vec::new();
199 framer.push(&chunk[..n], &mut decoded);
200 queue.push(&decoded);
201 }
202 }
203 }
204 queue.end();
205}
206
207struct LeF32 {
208 pending: [u8; 4],
209 have: usize,
210}
211
212impl LeF32 {
213 fn new() -> Self {
214 Self {
215 pending: [0; 4],
216 have: 0,
217 }
218 }
219
220 fn push(&mut self, mut bytes: &[u8], out: &mut Vec<f32>) {
221 if self.have > 0 {
222 let need = 4 - self.have;
223 let take = need.min(bytes.len());
224 self.pending[self.have..self.have + take].copy_from_slice(&bytes[..take]);
225 self.have += take;
226 bytes = &bytes[take..];
227 if self.have == 4 {
228 out.push(f32::from_le_bytes(self.pending));
229 self.have = 0;
230 }
231 }
232 let whole = bytes.len() - bytes.len() % 4;
233 for chunk in bytes[..whole].chunks_exact(4) {
234 let mut buf = [0u8; 4];
235 buf.copy_from_slice(chunk);
236 out.push(f32::from_le_bytes(buf));
237 }
238 let rest = &bytes[whole..];
239 self.pending[..rest.len()].copy_from_slice(rest);
240 self.have = rest.len();
241 }
242}
243
244pub(crate) fn ahead_is_full(buffered: usize, rate: u32, channels: u16) -> bool {
245 buffered > ahead_limit(rate, channels)
246}
247
248#[cfg(test)]
249mod tests {
250 use super::*;
251
252 fn spec(tracks: Vec<(i64, f32)>, channels: u16) -> DecodeSpec {
253 DecodeSpec {
254 ffmpeg: PathBuf::from("/opt/ffmpeg"),
255 media: PathBuf::from("/media/reel.mov"),
256 start: 1.5,
257 length: 10.0,
258 rate: 48_000,
259 channels,
260 tracks,
261 }
262 }
263
264 fn args(line: &CommandLine) -> Vec<String> {
265 line.args
266 .iter()
267 .map(|arg| arg.to_string_lossy().into_owned())
268 .collect()
269 }
270
271 #[test]
272 fn the_decoder_command_matches_the_reference_graph() {
273 let line = command_line(&spec(vec![(1, 0.4), (2, 1.0)], 1));
274 assert_eq!(line.program, PathBuf::from("/opt/ffmpeg"));
275 assert_eq!(
276 args(&line),
277 [
278 "-nostdin",
279 "-v",
280 "error",
281 "-ss",
282 "1.500",
283 "-t",
284 "10.000",
285 "-i",
286 "/media/reel.mov",
287 "-filter_complex",
288 "[0:1]volume=0.400,aformat=channel_layouts=stereo[t0];[0:2]volume=1.000,aformat=channel_layouts=stereo[t1];[t0][t1]amix=inputs=2:normalize=0[mix]",
289 "-map",
290 "[mix]",
291 "-f",
292 "f32le",
293 "-ac",
294 "2",
295 "-ar",
296 "48000",
297 "-",
298 ]
299 );
300 }
301
302 #[test]
303 fn a_single_file_decodes_at_the_chosen_rate() {
304 let mut single = spec(Vec::new(), 1);
305 single.media = PathBuf::from("/m/a.wav");
306 single.start = -2.5;
307 single.length = 1.25;
308 single.rate = 44_100;
309 let line = command_line(&single);
310 assert_eq!(
311 args(&line),
312 [
313 "-nostdin", "-v", "error", "-ss", "0.000", "-t", "1.250", "-i", "/m/a.wav", "-f",
314 "f32le", "-ac", "1", "-ar", "44100", "-",
315 ]
316 );
317 let live = spec_at_output(&single, 48_000, 2);
318 assert_eq!(live.rate, 48_000);
319 assert_eq!(live.channels, 2);
320 let live_args = args(&command_line(&live));
321 assert!(
322 live_args
323 .windows(2)
324 .any(|pair| pair[0] == "-ar" && pair[1] == "48000")
325 );
326 assert!(
327 live_args
328 .windows(2)
329 .any(|pair| pair[0] == "-ac" && pair[1] == "2")
330 );
331 }
332
333 #[test]
334 fn the_reader_waits_only_above_two_seconds() {
335 assert_eq!(ahead_limit(48_000, 2), 192_000);
336 assert_eq!(AHEAD_SECONDS, 2);
337 assert!(!ahead_is_full(192_000, 48_000, 2));
338 assert!(ahead_is_full(192_001, 48_000, 2));
339 }
340
341 #[test]
342 fn split_reads_keep_f32le_frames_whole() {
343 let mut framer = LeF32::new();
344 let mut out = Vec::new();
345 let first = 0.5f32.to_le_bytes();
346 let second = (-1.0f32).to_le_bytes();
347 let mut bytes = Vec::new();
348 bytes.extend_from_slice(&first);
349 bytes.extend_from_slice(&second);
350 bytes.push(0x01);
351 framer.push(&bytes[..2], &mut out);
352 assert!(out.is_empty());
353 framer.push(&bytes[2..6], &mut out);
354 framer.push(&bytes[6..], &mut out);
355 assert_eq!(out.len(), 2);
356 assert_eq!(out[0].to_bits(), 0.5f32.to_bits());
357 assert_eq!(out[1].to_bits(), (-1.0f32).to_bits());
358 assert_eq!(framer.have, 1);
359 }
360
361 #[test]
362 fn a_relative_ffmpeg_path_is_refused() {
363 let mut spec = spec(Vec::new(), 1);
364 spec.ffmpeg = PathBuf::from("ffmpeg");
365 assert!(matches!(
366 Decode::start(&spec),
367 Err(err) if err.contains("absolute")
368 ));
369 }
370
371 #[test]
372 #[ignore = "needs ffmpeg; set PITO_FFMPEG to an absolute path"]
373 fn ffmpeg_decodes_a_wav_to_f32() {
374 let ffmpeg = std::env::var_os("PITO_FFMPEG").expect("PITO_FFMPEG");
375 let dir = PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../../tmp/sound-ffmpeg");
376 std::fs::create_dir_all(&dir).unwrap();
377 let media = dir.join("tone.wav");
378 let clip = crate::Clip {
379 rate: 8_000,
380 channels: 1,
381 samples: vec![0.0, 0.5, -0.5, 1.0],
382 };
383 std::fs::write(&media, clip.to_wav()).unwrap();
384 let spec = DecodeSpec {
385 ffmpeg: PathBuf::from(ffmpeg),
386 media,
387 start: 0.0,
388 length: 1.0,
389 rate: 8_000,
390 channels: 1,
391 tracks: Vec::new(),
392 };
393 let decode = Decode::start(&spec).unwrap();
394 let deadline = std::time::Instant::now() + Duration::from_secs(5);
395 while !decode.ended() && std::time::Instant::now() < deadline {
396 std::thread::sleep(Duration::from_millis(10));
397 }
398 assert!(decode.ended());
399 let popped = decode.queue().pop_frames(4, 1);
400 assert!(popped.ended);
401 assert_eq!(popped.real, 4);
402 for (got, want) in popped.samples.iter().zip(clip.samples.iter()) {
403 assert!((got - want).abs() < 0.0001);
404 }
405 }
406}