Merge pull request #317 from marceline-cramer/rust-native-audio-trans…

…coding

Rust-native audio transcoding

Cargo.toml

@@ -139,6 +139,8 @@ toml_edit = "0.19.3"
 arboard = "3.2.0"
 noise = { version = "0.7.0", default-features = false }
 russimp = { version = "1.0.6", features = ['prebuilt'] }
+symphonia = { version = "0.5", default-features = false, features = ["mp3", "pcm", "wav"] }
+vorbis_rs = "0.3.0"
 colored = "2.0.0"
 
 #

crates/build/Cargo.toml

@@ -46,3 +46,6 @@ parking_lot = { workspace = true }
 async-trait = { workspace = true }
 dyn-clonable = { workspace = true }
 cargo_toml = { workspace = true }
+symphonia = { workspace = true }
+vorbis_rs = { workspace = true }
+rand = { workspace = true }

crates/build/src/pipelines/audio/mod.rs

@@ -1,10 +1,6 @@
-use std::process::Stdio;
-
 use ambient_std::asset_url::AssetType;
 use ambient_world_audio::AudioNode;
 use anyhow::Context;
-use futures::FutureExt;
-use tokio::io::{AsyncRead, AsyncReadExt};
 use tracing::{info_span, Instrument};
 
 use super::{
@@ -26,18 +22,18 @@ pub async fn pipeline(ctx: &PipelineCtx) -> Vec<OutAsset> {
 
             let content_url = match file.extension().as_deref() {
                 Some("ogg") => ctx.write_file(&rel_path, contents).await,
-                ext @ Some("wav" | "mp3") => {
+                Some(ext @ "wav") | Some(ext @ "mp3") => {
                     tracing::info!("Processing {ext:?} file");
                     // Make sure to take the contents, to avoid having both the input and output in
                     // memory at once
-                    let contents = ffmpeg_convert(std::io::Cursor::new(contents)).await?;
+                    let contents = symphonia_convert(ext, contents).await?;
                     ctx.write_file(rel_path.with_extension("ogg"), contents).await
                 }
                 other => anyhow::bail!("Audio filetype {:?} is not yet supported", other.unwrap_or_default()),
             };
 
             let root_node = AudioNode::Vorbis { url: content_url.to_string() };
-            let graph_url = ctx.write_file(&rel_path.with_extension("SOUND_GRAPH_EXTENSION"), save_audio_graph(root_node).unwrap()).await;
+            let graph_url = ctx.write_file(&rel_path.with_extension(SOUND_GRAPH_EXTENSION), save_audio_graph(root_node).unwrap()).await;
 
             Ok(vec![
                 OutAsset {
@@ -74,44 +70,111 @@ fn save_audio_graph(root: AudioNode) -> anyhow::Result<Vec<u8>> {
 }
 
 #[tracing::instrument(level = "info", skip(input))]
-async fn ffmpeg_convert<A>(input: A) -> anyhow::Result<Vec<u8>>
-where
-    A: 'static + Send + AsyncRead,
-{
-    let mut child = tokio::process::Command::new("ffmpeg")
-        .args(["-i", "pipe:", "-f", "ogg", "pipe:1"])
-        .stdin(Stdio::piped())
-        .stdout(Stdio::piped())
-        .spawn()
-        .context("Failed to execute ffmpeg")?;
-
-    tracing::info!("Writing to stdin");
-
-    let mut stdin = child.stdin.take().expect("no stdin");
-    let mut stdout = child.stdout.take().expect("no stdout");
-
-    let input = tokio::task::spawn(async move {
-        tokio::pin!(input);
-        tokio::io::copy(&mut input, &mut stdin).await.context("Failed to write to stdin")
-    })
-    .map(|v| v.unwrap());
-
-    let output = async move {
-        let mut output = Vec::new();
-        stdout.read_to_end(&mut output).await.unwrap();
-        Ok(output)
+async fn symphonia_convert(ext: &str, input: Vec<u8>) -> anyhow::Result<Vec<u8>> {
+    use std::num::{NonZeroU32, NonZeroU8};
+
+    use symphonia::core::{
+        codecs::{DecoderOptions, CODEC_TYPE_NULL},
+        errors::Error,
+        formats::FormatOptions,
+        io::MediaSourceStream,
+        meta::MetadataOptions,
+        probe::Hint,
     };
 
-    let status = async { child.wait().await.context("Failed to wait for ffmpeg") };
-
-    tracing::info!("Waiting for ffmpeg to complete");
-    let (_, output, status) = tokio::try_join!(input, output, status)?;
+    use vorbis_rs::{VorbisBitrateManagementStrategy, VorbisEncoder};
+
+    // this symphonia decoding code is largely based on symphonia's examples:
+    // https://github.com/pdeljanov/Symphonia/blob/master/symphonia/examples
+
+    // hint symphonia about what format this file might be
+    let mut hint = Hint::new();
+    hint.with_extension(ext);
+
+    // create a media source stream with default options
+    let media_source = Box::new(std::io::Cursor::new(input));
+    let mss = MediaSourceStream::new(media_source, Default::default());
+
+    // use default metadata and format reader options
+    let meta_opts = MetadataOptions::default();
+    let fmt_opts = FormatOptions::default();
+
+    // probe the audio file for its params
+    let probed = symphonia::default::get_probe().format(&hint, mss, &fmt_opts, &meta_opts).context("Failed to probe audio format")?;
+    let mut format = probed.format;
+
+    // find the default audio track for this file
+    let track = format.tracks().iter().find(|t| t.codec_params.codec != CODEC_TYPE_NULL).context("Failed to select default audio track")?;
+
+    // init an audio decoder with default options
+    let dec_opts = DecoderOptions::default();
+    let mut decoder = symphonia::default::get_codecs().make(&track.codec_params, &dec_opts).context("Failed to create audio decoder")?;
+
+    // randomize an ogg stream serial number
+    let stream_serial: i32 = rand::random();
+
+    // retrieve the sampling rate from the input file
+    let sampling_rate: NonZeroU32 = decoder
+        .codec_params()
+        .sample_rate
+        .context("Expected audio to have sample rate")?
+        .try_into()
+        .context("Audio must have >0 sampling rate")?;
+
+    // retrieve the channel count from the input file
+    let channels = decoder.codec_params().channels.context("Audio does not have any channels")?.count();
+    let channels: NonZeroU8 = (channels as u8).try_into().context("Audio must have >0 channels")?;
+
+    // select a bitrate
+    let bitrate = VorbisBitrateManagementStrategy::QualityVbr { target_quality: 0.9 };
+
+    // create the ogg Vorbis encoder
+    let mut encoder = VorbisEncoder::new(
+        stream_serial,
+        [("", ""); 0], // no tags
+        sampling_rate,
+        channels,
+        bitrate,
+        None,
+        Vec::new(),
+    )?;
+
+    // process all packets in the input file
+    let result = loop {
+        // read the next packet
+        let packet = match format.next_packet() {
+            Ok(packet) => packet,
+            Err(err) => break err,
+        };
+
+        // decode the packet's samples
+        let decoded = match decoder.decode(&packet) {
+            Ok(buf) => buf,
+            Err(err) => break err,
+        };
+
+        // convert the decoded samples to f32 samples
+        let mut block = decoded.make_equivalent::<f32>();
+        decoded.convert(&mut block);
+
+        // get the samples as &[&[f32]]
+        let planes = block.planes();
+        let plane_samples = planes.planes();
+
+        // feed the samples into the encoder
+        encoder.encode_audio_block(plane_samples)?;
+    };
 
-    if !status.success() {
-        anyhow::bail!("FFMPEG conversion failed")
+    // process the error returned by the loop
+    match result {
+        // "end of stream" is non-fatal, ignore it
+        Error::IoError(err) if err.kind() == std::io::ErrorKind::UnexpectedEof && err.to_string() == "end of stream" => {}
+        // return every other kind of error
+        err => return Err(err.into()),
     }
 
-    tracing::info!("Converted to vorbis of {} kb", output.len() as f32 / 1000.0);
-
+    // finish encoding
+    let output = encoder.finish()?;
+    tracing::info!("Encoded {} samples", output.len());
     Ok(output)
 }