bump version (v1.0.0)

it writes out the example config into a file

.editorconfig

@@ -0,0 +1,8 @@
+root = true
+
+[*]
+indent_style = space
+indent_size = 4
+charset = utf-8
+trim_trailing_whitespace = true
+insert_final_newline = true

.gitignore

@@ -1,2 +1 @@
 /target/
-/Cargo.lock

Cargo.toml

@@ -1,9 +1,38 @@
 [package]
-name = "audio_conv"
+name = "audio-conv"
-version = "0.0.0"
+version = "1.0.0"
-authors = ["Thomas Heck <t@b128.net>"]
 edition = "2018"
+description = "Copies directory structure and converts audio files in it"
+authors = ["Thomas Heck <t@b128.net>"]
+repository = "https://gitlab.com/chpio/audio-conv"
+license = "MIT OR Apache-2.0"
+categories = [
+    "command-line-utilities",
+    "multimedia",
+    "multimedia::audio",
+    "multimedia::encoding",
+]
+keywords = ["audio", "conversion", "opus", "flac"]
 
 [dependencies]
-stainless_ffmpeg = "0.1"
+gstreamer-audio = { version = "0.16", features = ["v1_10"] }
-rayon = "1.0.2"
+gstreamer = { version = "0.16", features = ["v1_10"] }
+gstreamer-base = { version = "0.16", features = ["v1_10"] }
+glib = "0.10"
+futures = "0.3"
+num_cpus = "1"
+walkdir = "2"
+libc = "0.2"
+anyhow = "1"
+clap = "2"
+serde = { version = "1.0", features = ["derive"] }
+serde_yaml = "0.8"
+regex = "1"
+globset = "0.4"
+derive_more = "0.99"
+tui = { version = "0.13", default-features = false, features = ["crossterm"] }
+
+[dependencies.tokio]
+version = "1"
+default-features = false
+features = ["sync", "rt", "macros", "time", "fs", "io-util"]

example.audio-conv.yaml

@@ -0,0 +1,21 @@
+from: ./music
+to: ./converted_test
+matches:
+  - extensions:
+      - flac
+      - wav
+    # and/or `glob: "**/*.flac"`
+    # and/or `regex: "\.flac$"`
+    # you can also leave it empty for the default extensions
+
+    to:
+      codec: opus
+      bitrate: 160
+      bitrate_type: vbr # or cbr
+
+    # for mp3:
+    # to:
+    #   codec: mp3
+    #   # one of: 8, 16, 24, 32, 40, 48, 56, 64, 80, 96, 112, 128, 160, 192, 224, 256 or 320
+    #   bitrate: 256
+    #   bitrate_type: vbr # or cbr

flake.lock

@@ -0,0 +1,58 @@
+{
+  "nodes": {
+    "flake-utils": {
+      "locked": {
+        "lastModified": 1605370193,
+        "narHash": "sha256-YyMTf3URDL/otKdKgtoMChu4vfVL3vCMkRqpGifhUn0=",
+        "owner": "numtide",
+        "repo": "flake-utils",
+        "rev": "5021eac20303a61fafe17224c087f5519baed54d",
+        "type": "github"
+      },
+      "original": {
+        "owner": "numtide",
+        "repo": "flake-utils",
+        "type": "github"
+      }
+    },
+    "import-cargo": {
+      "locked": {
+        "lastModified": 1594305518,
+        "narHash": "sha256-frtArgN42rSaEcEOYWg8sVPMUK+Zgch3c+wejcpX3DY=",
+        "owner": "edolstra",
+        "repo": "import-cargo",
+        "rev": "25d40be4a73d40a2572e0cc233b83253554f06c5",
+        "type": "github"
+      },
+      "original": {
+        "owner": "edolstra",
+        "repo": "import-cargo",
+        "type": "github"
+      }
+    },
+    "nixpkgs": {
+      "locked": {
+        "lastModified": 1608633860,
+        "narHash": "sha256-AGJfdJCR5jfIt8PqGiENXRqhthrS3Gxy8Wzb3Z2GsS4=",
+        "owner": "NixOS",
+        "repo": "nixpkgs",
+        "rev": "da1b28ab8f361fbe14dc539cd69ce1bfd015fd68",
+        "type": "github"
+      },
+      "original": {
+        "owner": "NixOS",
+        "repo": "nixpkgs",
+        "type": "github"
+      }
+    },
+    "root": {
+      "inputs": {
+        "flake-utils": "flake-utils",
+        "import-cargo": "import-cargo",
+        "nixpkgs": "nixpkgs"
+      }
+    }
+  },
+  "root": "root",
+  "version": 7
+}

flake.nix

@@ -0,0 +1,63 @@
+{
+    description = "Converts audio files";
+
+    inputs = {
+        nixpkgs.url = github:NixOS/nixpkgs;
+        flake-utils.url = "github:numtide/flake-utils";
+        import-cargo.url = github:edolstra/import-cargo;
+    };
+
+    outputs = { self, flake-utils, nixpkgs, import-cargo }:
+        flake-utils.lib.eachDefaultSystem (system:
+            let
+                pkgs = import nixpkgs { inherit system; };
+
+                buildtimeDeps = with pkgs; [
+                    cargo
+                    rustc
+                    pkg-config
+                ];
+
+                runtimeDeps = with pkgs; [
+                    gst_all_1.gstreamer
+
+                    # needed for opus, resample, ...
+                    gst_all_1.gst-plugins-base
+
+                    # needed for flac
+                    gst_all_1.gst-plugins-good
+                ];
+
+                inherit (import-cargo.builders) importCargo;
+            in {
+                defaultPackage = pkgs.stdenv.mkDerivation {
+                    name = "audio-conv";
+                    src = self;
+
+                    nativeBuildInputs = [
+                        # setupHook which makes sure that a CARGO_HOME with vendored dependencies
+                        # exists
+                        (importCargo { lockFile = ./Cargo.lock; inherit pkgs; }).cargoHome
+                    ]
+                        ++ buildtimeDeps;
+
+                    buildInputs = runtimeDeps;
+
+                    buildPhase = ''
+                        cargo build --release --offline
+                    '';
+
+                    installPhase = ''
+                        install -Dm775 ./target/release/audio-conv $out/bin/audio-conv
+                    '';
+                };
+
+                devShell = pkgs.stdenv.mkDerivation {
+                    name = "audio-conv";
+                    buildInputs = [ pkgs.rustfmt pkgs.rust-analyzer ]
+                        ++ buildtimeDeps
+                        ++ runtimeDeps;
+                };
+            }
+        );
+}

shell.nix

@@ -1,15 +0,0 @@
-with import <nixpkgs> {};
-
-stdenv.mkDerivation {
-    name = "audio_conv";
-    buildInputs = [
-        stdenv
-        pkg-config
-        ffmpeg_4
-        clang
-        cargo
-        rustc
-        rls
-    ];
-    LIBCLANG_PATH="${llvmPackages.libclang}/lib";
-}

src/config.rs

@@ -0,0 +1,278 @@
+use anyhow::{Context, Error, Result};
+use globset::GlobBuilder;
+use regex::bytes::{Regex, RegexBuilder};
+use serde::Deserialize;
+use std::{
+    io::Write,
+    path::{Path, PathBuf},
+};
+
+#[derive(Debug)]
+pub struct Config {
+    pub from: PathBuf,
+    pub to: PathBuf,
+    pub matches: Vec<TranscodeMatch>,
+}
+
+#[derive(Debug)]
+pub struct TranscodeMatch {
+    pub regexes: Vec<Regex>,
+    pub to: Transcode,
+}
+
+#[derive(Clone, Debug, Deserialize)]
+#[serde(tag = "codec")]
+pub enum Transcode {
+    #[serde(rename = "opus")]
+    Opus {
+        #[serde(default = "default_opus_bitrate")]
+        bitrate: u16,
+
+        #[serde(default = "bitrate_type_vbr")]
+        bitrate_type: BitrateType,
+    },
+
+    #[serde(rename = "mp3")]
+    Mp3 {
+        #[serde(default = "default_mp3_bitrate")]
+        bitrate: u16,
+
+        #[serde(default = "bitrate_type_vbr")]
+        bitrate_type: BitrateType,
+    },
+}
+
+impl Transcode {
+    pub fn extension(&self) -> &'static str {
+        match self {
+            Transcode::Opus { .. } => "opus",
+            Transcode::Mp3 { .. } => "mp3",
+        }
+    }
+}
+
+fn default_opus_bitrate() -> u16 {
+    160
+}
+
+fn bitrate_type_vbr() -> BitrateType {
+    BitrateType::Vbr
+}
+
+fn default_mp3_bitrate() -> u16 {
+    256
+}
+
+impl Default for Transcode {
+    fn default() -> Self {
+        Transcode::Opus {
+            bitrate: default_opus_bitrate(),
+            bitrate_type: bitrate_type_vbr(),
+        }
+    }
+}
+
+#[derive(Clone, Debug, Deserialize)]
+pub enum BitrateType {
+    #[serde(rename = "cbr")]
+    Cbr,
+    #[serde(rename = "vbr")]
+    Vbr,
+}
+
+#[derive(Debug, Default, Deserialize)]
+struct ConfigFile {
+    from: Option<PathBuf>,
+    to: Option<PathBuf>,
+
+    #[serde(default)]
+    matches: Vec<TranscodeMatchFile>,
+}
+
+#[derive(Debug, Deserialize)]
+struct TranscodeMatchFile {
+    glob: Option<String>,
+    regex: Option<String>,
+
+    #[serde(default)]
+    extensions: Vec<String>,
+
+    to: Transcode,
+}
+
+pub fn config() -> Result<Config> {
+    use clap::{App, Arg, SubCommand};
+
+    let arg_matches = App::new("audio-conv")
+        .version(clap::crate_version!())
+        .about("Converts audio files")
+        .arg(
+            Arg::with_name("config")
+                .short("c")
+                .long("config")
+                .required(false)
+                .takes_value(true)
+                .help("path to an audio-conv config file, defaults to \"audio-conv.yaml\""),
+        )
+        .arg(
+            Arg::with_name("from")
+                .short("f")
+                .long("from")
+                .required(false)
+                .takes_value(true)
+                .help("from directory path"),
+        )
+        .arg(
+            Arg::with_name("to")
+                .short("t")
+                .long("to")
+                .required(false)
+                .takes_value(true)
+                .help("to directory path"),
+        )
+        .subcommand(SubCommand::with_name("init").about("writes an example config"))
+        .get_matches();
+
+    let current_dir = std::env::current_dir().context("could not get current directory")?;
+
+    let config_path = arg_matches.value_of_os("config");
+    let force_load = config_path.is_some();
+    let config_path = config_path
+        .map(AsRef::<Path>::as_ref)
+        .unwrap_or_else(|| AsRef::<Path>::as_ref("audio-conv.yaml"));
+    let config_path = current_dir.join(config_path);
+
+    if let Some("init") = arg_matches.subcommand_name() {
+        std::fs::OpenOptions::new()
+            .write(true)
+            .create_new(true)
+            .open(&config_path)
+            .and_then(|mut f| f.write_all(std::include_bytes!("../example.audio-conv.yaml")))
+            .with_context(|| format!("unable to write config file to {}", config_path.display()))?;
+
+        std::process::exit(0);
+    }
+
+    let config_dir = config_path
+        .parent()
+        .context("could not get parent directory of the config file")?;
+
+    let config_file = load_config_file(&config_path)
+        .with_context(|| format!("failed loading config file \"{}\"", config_path.display()))?;
+
+    if force_load && config_file.is_none() {
+        return Err(Error::msg(format!(
+            "could not find config file \"{}\"",
+            config_path.display()
+        )));
+    }
+
+    let default_regex = RegexBuilder::new("\\.(flac|wav)$")
+        .case_insensitive(true)
+        .build()
+        .expect("failed compiling default match regex");
+
+    let transcode_matches = config_file
+        .as_ref()
+        .map(|config_file| {
+            config_file
+                .matches
+                .iter()
+                .map(|m| {
+                    let glob = m.glob.iter().map(|glob| {
+                        let glob = GlobBuilder::new(glob)
+                            .case_insensitive(true)
+                            .build()
+                            .context("failed building glob")?;
+                        let regex = Regex::new(glob.regex()).context("failed compiling regex")?;
+                        Ok(regex)
+                    });
+
+                    let regex = m.regex.iter().map(|regex| {
+                        let regex = RegexBuilder::new(regex)
+                            .case_insensitive(true)
+                            .build()
+                            .context("failed compiling regex")?;
+                        Ok(regex)
+                    });
+
+                    let extensions = m.extensions.iter().map(|ext| {
+                        let mut ext = regex::escape(ext);
+                        ext.insert_str(0, &"\\.");
+                        ext.push_str("$");
+
+                        let regex = RegexBuilder::new(&ext)
+                            .case_insensitive(true)
+                            .build()
+                            .context("failed compiling regex")?;
+                        Ok(regex)
+                    });
+
+                    let mut regexes = glob
+                        .chain(regex)
+                        .chain(extensions)
+                        .collect::<Result<Vec<_>>>()?;
+
+                    if regexes.is_empty() {
+                        regexes.push(default_regex.clone());
+                    }
+
+                    Ok(TranscodeMatch {
+                        regexes,
+                        to: m.to.clone(),
+                    })
+                })
+                .collect::<Result<Vec<_>>>()
+        })
+        .transpose()?
+        .filter(|matches| !matches.is_empty())
+        .unwrap_or_else(|| {
+            vec![TranscodeMatch {
+                regexes: vec![default_regex],
+                to: Transcode::default(),
+            }]
+        });
+
+    Ok(Config {
+        from: {
+            arg_matches
+                .value_of_os("from")
+                .map(|p| current_dir.join(p))
+                .or_else(|| {
+                    config_file
+                        .as_ref()
+                        .map(|c| c.from.as_ref())
+                        .flatten()
+                        .map(|p| config_dir.join(p))
+                })
+                .ok_or_else(|| Error::msg("\"from\" not configured"))?
+                .canonicalize()
+                .context("could not canonicalize \"from\" path")?
+        },
+        to: arg_matches
+            .value_of_os("to")
+            .map(|p| current_dir.join(p))
+            .or_else(|| {
+                config_file
+                    .as_ref()
+                    .map(|c| c.to.as_ref())
+                    .flatten()
+                    .map(|p| config_dir.join(p))
+            })
+            .ok_or_else(|| Error::msg("\"to\" not configured"))?
+            .canonicalize()
+            .context("could not canonicalize \"to\" path")?,
+        matches: transcode_matches,
+    })
+}
+
+fn load_config_file(path: &Path) -> Result<Option<ConfigFile>> {
+    let mut file = match std::fs::File::open(path) {
+        Ok(file) => file,
+        Err(err) if err.kind() == std::io::ErrorKind::NotFound => return Ok(None),
+        Err(err) => return Err(Error::new(err)),
+    };
+    let config: ConfigFile =
+        serde_yaml::from_reader(&mut file).context("could not parse config file")?;
+    Ok(Some(config))
+}

src/main.rs

@@ -1,277 +1,590 @@
-use std::env;
+mod config;
-use std::path::Path;
+mod ui;
-use std::{fs, io};
 
-use stainless_ffmpeg::{self as ffmpeg, codec, filter, format, frame, media};
+use crate::config::Config;
-use rayon::prelude::*;
+use anyhow::{Context, Error, Result};
+use futures::{future, pin_mut, prelude::*};
+use glib::{subclass::prelude::*, GBoxed, GString};
+use gstreamer::{gst_element_error, prelude::*, Element};
+use gstreamer_base::prelude::*;
+use std::{
+    borrow::Cow,
+    error::Error as StdError,
+    ffi, fmt,
+    fmt::Write as FmtWrite,
+    path::{Path, PathBuf},
+    result::Result as StdResult,
+    sync::Arc,
+    time::Duration,
+};
+use tokio::{fs, io::AsyncWriteExt, task, time::interval};
 
-#[derive(Debug)]
+#[derive(Clone, Debug, GBoxed)]
-enum Error {
+#[gboxed(type_name = "GBoxErrorWrapper")]
-    Io(io::Error),
+struct GBoxErrorWrapper(Arc<Error>);
-    Ffmpeg(ffmpeg::Error),
-    String(String),
-    Str(&'static str),
-}
 
-impl From<ffmpeg::error::Error> for Error {
+impl GBoxErrorWrapper {
-    fn from(v: ffmpeg::Error) -> Error {
+    fn new(err: Error) -> Self {
-        Error::Ffmpeg(v)
+        GBoxErrorWrapper(Arc::new(err))
     }
 }
 
-impl From<io::Error> for Error {
+impl StdError for GBoxErrorWrapper {
-    fn from(v: io::Error) -> Error {
+    fn source(&self) -> Option<&(dyn StdError + 'static)> {
-        Error::Io(v)
+        self.0.source()
     }
 }
 
-impl From<String> for Error {
+impl fmt::Display for GBoxErrorWrapper {
-    fn from(v: String) -> Error {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> StdResult<(), fmt::Error> {
-        Error::String(v)
+        self.0.fmt(f)
     }
 }
 
-impl From<&'static str> for Error {
+#[derive(Debug, derive_more::Display, derive_more::Error)]
-    fn from(v: &'static str) -> Error {
+#[display(fmt = "Received error from {}: {} (debug: {:?})", src, error, debug)]
-        Error::Str(v)
+struct GErrorMessage {
-    }
+    src: String,
+    error: String,
+    debug: Option<String>,
+    source: glib::Error,
 }
 
-fn filter(
+fn gmake<T: IsA<Element>>(factory_name: &str) -> Result<T> {
-    decoder: &codec::decoder::Audio,
+    let res = gstreamer::ElementFactory::make(factory_name, None)
-    encoder: &codec::encoder::Audio,
+        .with_context(|| format!("could not make gstreamer Element \"{}\"", factory_name))?
-) -> Result<filter::Graph, Error> {
+        .downcast()
-    let mut filter = filter::Graph::new();
+        .ok()
-
+        .with_context(|| {
-    let args = format!(
+            format!(
-        "time_base={}:sample_rate={}:sample_fmt={}:channel_layout=0x{:x}",
+                "could not cast gstreamer Element \"{}\" into `{}`",
-        decoder.time_base(),
+                factory_name,
-        decoder.rate(),
+                std::any::type_name::<T>()
-        decoder.format().name(),
+            )
-        decoder.channel_layout().bits()
+        })?;
-    );
+    Ok(res)
-
-    filter.add(&filter::find("abuffer").unwrap(), "in", &args)?;
-    filter.add(&filter::find("abuffersink").unwrap(), "out", "")?;
-
-    {
-        let mut out = filter.get("out").unwrap();
-
-        out.set_sample_format(encoder.format());
-        out.set_channel_layout(encoder.channel_layout());
-        out.set_sample_rate(encoder.rate());
-    }
-
-    filter.output("in", 0)?.input("out", 0)?.parse("anull")?;
-    filter.validate()?;
-
-    if let Some(codec) = encoder.codec() {
-        if !codec
-            .capabilities()
-            .contains(ffmpeg::codec::capabilities::VARIABLE_FRAME_SIZE)
-        {
-            filter
-                .get("out")
-                .unwrap()
-                .sink()
-                .set_frame_size(encoder.frame_size());
-        }
-    }
-
-    Ok(filter)
 }
 
-struct Transcoder {
+#[derive(Debug, Clone)]
-    stream: usize,
+pub struct ConversionArgs {
-    filter: filter::Graph,
+    rel_from_path: PathBuf,
-    decoder: codec::decoder::Audio,
+    transcode: config::Transcode,
-    encoder: codec::encoder::Audio,
 }
 
-fn transcoder(
+fn get_conversion_args(config: &Config) -> impl Iterator<Item = Result<ConversionArgs>> + '_ {
-    ictx: &mut format::context::Input,
+    walkdir::WalkDir::new(&config.from)
-    octx: &mut format::context::Output,
+        .into_iter()
-) -> Result<Transcoder, Error> {
+        .filter_map(|e| e.ok())
-    let input = ictx
+        .filter(|e| e.file_type().is_file())
-        .streams()
+        .map(move |e| -> Result<Option<ConversionArgs>> {
-        .best(media::Type::Audio)
+            let from_bytes = path_to_bytes(e.path());
-        .expect("could not find best audio stream");
-    let mut decoder = input.codec().decoder().audio()?;
 
-    let codec = ffmpeg::encoder::find(octx.format().codec(&"", media::Type::Audio))
+            let transcode = config
-        .expect("failed to find encoder")
+                .matches
-        .audio()?;
+                .iter()
-    let global = octx
+                .filter(|m| {
-        .format()
+                    m.regexes
-        .flags()
+                        .iter()
-        .contains(ffmpeg::format::flag::GLOBAL_HEADER);
+                        .any(|regex| regex.is_match(from_bytes.as_ref()))
-
-    decoder.set_parameters(input.parameters())?;
-
-    let mut output = octx.add_stream(codec)?;
-    let mut encoder = output.codec().encoder().audio()?;
-
-    let channel_layout = codec
-        .channel_layouts()
-        .map(|cls| cls.best(decoder.channel_layout().channels()))
-        .unwrap_or(ffmpeg::channel_layout::STEREO);
-
-    if global {
-        encoder.set_flags(ffmpeg::codec::flag::GLOBAL_HEADER);
-    }
-
-    encoder.set_rate(48_000); //decoder.rate() as i32);
-    encoder.set_channel_layout(channel_layout);
-    encoder.set_channels(channel_layout.channels());
-    encoder.set_format(
-        codec
-            .formats()
-            .expect("unknown supported formats")
-            .next()
-            .unwrap(),
-    );
-
-    encoder.set_time_base((1, 48_000));
-    output.set_time_base((1, 48_000));
-
-    let mut encode_dict = ffmpeg::Dictionary::new();
-    encode_dict.set("vbr", "on");
-    encoder.set_bit_rate(192_000);
-    let encoder = encoder.open_as_with(codec, encode_dict)?;
-    output.set_parameters(&encoder);
-
-    let filter = filter(&decoder, &encoder)?;
-
-    Ok(Transcoder {
-        stream: input.index(),
-        filter,
-        decoder,
-        encoder,
                 })
+                .map(|m| m.to.clone())
+                .next();
+            let transcode = if let Some(transcode) = transcode {
+                transcode
+            } else {
+                return Ok(None);
+            };
+
+            let rel_path = e.path().strip_prefix(&config.from).with_context(|| {
+                format!(
+                    "unable to get relative path for {} from {}",
+                    e.path().display(),
+                    config.from.display()
+                )
+            })?;
+
+            let mut to = config.to.join(&rel_path);
+            to.set_extension(transcode.extension());
+
+            let is_newer = {
+                let from_mtime = e
+                    .metadata()
+                    .map_err(Error::new)
+                    .and_then(|md| md.modified().map_err(Error::new))
+                    .with_context(|| {
+                        format!("unable to get mtime for from file {}", e.path().display())
+                    })?;
+                let to_mtime = to.metadata().and_then(|md| md.modified());
+                match to_mtime {
+                    Ok(to_mtime) => to_mtime < from_mtime,
+                    Err(err) if err.kind() == std::io::ErrorKind::NotFound => true,
+                    Err(err) => {
+                        return Err(err).with_context(|| {
+                            format!("unable to get mtime for to file {}", to.display())
+                        })
+                    }
+                }
+            };
+
+            if is_newer {
+                Ok(Some(ConversionArgs {
+                    rel_from_path: rel_path.to_path_buf(),
+                    transcode,
+                }))
+            } else {
+                Ok(None)
+            }
+        })
+        .filter_map(|e| e.transpose())
 }
 
-fn transcode(input: &Path, output: &Path) -> Result<(), Error> {
+#[tokio::main(flavor = "current_thread")]
-    println!("{:?}", output);
+async fn main() -> Result<()> {
+    task::LocalSet::new()
+        .run_until(async move {
+            let (ui_queue, ui_fut) = ui::init();
 
-    let mut ictx = format::input(&input)?;
+            let main_handle = async move {
-    let original_extension = output
+                let ok = task::spawn_local(main_loop(ui_queue))
-        .extension()
+                    .await
-        .expect("file without extension")
+                    .context("main task failed")??;
-        .to_string_lossy();
+                Result::<_>::Ok(ok)
-    let output_tmp = output.with_extension("tmp");
+            };
-    let mut octx = format::output_as(&output_tmp, &original_extension)?;
-    let mut transcoder = transcoder(&mut ictx, &mut octx)?;
 
-    octx.set_metadata(ictx.metadata().to_owned());
+            let ui_handle = async move {
-    octx.write_header()?;
+                let ok = task::spawn_local(ui_fut)
+                    .await
+                    .context("ui task failed")?
+                    .context("ui failed")?;
+                Result::<_>::Ok(ok)
+            };
 
-    let in_time_base = transcoder.decoder.time_base();
+            future::try_join(main_handle, ui_handle).await?;
+            Ok(())
+        })
+        .await
+}
 
-    let mut frame = frame::Audio::empty();
+async fn main_loop(ui_queue: ui::MsgQueue) -> Result<()> {
-    let mut encoded = ffmpeg::Packet::empty();
+    let (config, conv_args) = task::spawn_blocking(|| -> Result<_> {
+        gstreamer::init()?;
+        let config = config::config().context("could not get the config")?;
 
-    for (stream, mut packet) in ictx.packets() {
+        let conv_args = get_conversion_args(&config)
-        if stream.index() != transcoder.stream {
+            .collect::<Result<Vec<_>>>()
-            continue;
+            .context("failed loading dir structure")?;
-        }
 
-        packet.rescale_ts(stream.time_base(), in_time_base);
+        Ok((config, conv_args))
+    })
+    .await
+    .context("init task failed")??;
 
-        if let Ok(true) = transcoder.decoder.decode(&packet, &mut frame) {
+    let log_path = Path::new(".")
-            transcoder.filter.get("in").unwrap().source().add(&frame)?;
+        .canonicalize()
+        .context("unable to canonicalize path to log file")?
+        .join("audio-conv.log");
 
-            while let Ok(..) = transcoder
+    ui_queue.push(ui::Msg::Init {
-                .filter
+        task_len: conv_args.len(),
-                .get("out")
+        log_path: log_path.clone(),
-                .unwrap()
+    });
-                .sink()
+
-                .frame(&mut frame)
+    stream::iter(conv_args.into_iter().enumerate())
+        .map(Ok)
+        .try_for_each_concurrent(num_cpus::get(), |(i, args)| {
+            let config = &config;
+            let ui_queue = &ui_queue;
+            let log_path = &log_path;
+
+            async move {
+                ui_queue.push(ui::Msg::TaskStart {
+                    id: i,
+                    args: args.clone(),
+                });
+
+                match transcode(config, &args, i, ui_queue).await {
+                    Ok(()) => ui_queue.push(ui::Msg::TaskEnd { id: i }),
+                    Err(err) => {
+                        let err = err.context(format!(
+                            "failed transcoding \"{}\"",
+                            args.rel_from_path.display()
+                        ));
+
+                        let mut log_file = match fs::OpenOptions::new()
+                            .create(true)
+                            .append(true)
+                            .open(log_path)
+                            .await
                         {
-                if let Ok(true) = transcoder.encoder.encode(&frame, &mut encoded) {
+                            Ok(log_file) => log_file,
-                    encoded.set_stream(0);
+                            Err(fs_err) => {
-                    encoded.write_interleaved(&mut octx)?;
+                                let err = err.context(fs_err).context("Unable to open log file");
-                }
+                                return Err(err);
-                unsafe {
-                    ffmpeg::ffi::av_frame_unref(frame.as_mut_ptr());
                             }
+                        };
+
+                        let mut err_str = String::new();
+                        if let Err(write_err) = write!(&mut err_str, "{:?}\n", err) {
+                            let err = err.context(format!(
+                                "Unable to format transcoding error for logging (write error: {})",
+                                write_err
+                            ));
+                            return Err(err);
                         }
+
+                        log_file
+                            .write_all(err_str.as_ref())
+                            .await
+                            .map_err(|fs_err| {
+                                err.context(format!(
+                                    "Unable to write transcoding error to log file (fs error: {})",
+                                    fs_err
+                                ))
+                            })?;
+
+                        ui_queue.push(ui::Msg::TaskError { id: i });
                     }
                 }
 
-    transcoder.filter.get("in").unwrap().source().flush()?;
+                Result::<_>::Ok(())
-
-    while let Ok(..) = transcoder
-        .filter
-        .get("out")
-        .unwrap()
-        .sink()
-        .frame(&mut frame)
-    {
-        if let Ok(true) = transcoder.encoder.encode(&frame, &mut encoded) {
-            encoded.set_stream(0);
-            encoded.write_interleaved(&mut octx)?;
-        }
             }
+        })
+        .await?;
 
-    if let Ok(true) = transcoder.encoder.flush(&mut encoded) {
+    ui_queue.push(ui::Msg::Exit);
-        encoded.set_stream(0);
-        encoded.write_interleaved(&mut octx)?;
-    }
-
-    octx.write_trailer()?;
-
-    fs::rename(output_tmp, output)?;
 
     Ok(())
 }
 
-fn transcode_path(input: &Path, output: &Path) -> Result<(), Error> {
+async fn transcode(
-    input.read_dir()?.par_bridge().try_for_each(|entry| {
+    config: &Config,
-        let entry = entry?;
+    args: &ConversionArgs,
-        let file_type = entry.file_type()?;
+    task_id: usize,
-        let new_input = input.join(entry.file_name());
+    queue: &ui::MsgQueue,
-        let mut new_output = output.join(entry.file_name());
+) -> Result<()> {
-        if file_type.is_dir() {
+    let from_path = config.from.join(&args.rel_from_path);
-            transcode_path(new_input.as_ref(), new_output.as_ref())?;
+    let mut to_path = config.to.join(&args.rel_from_path);
-        } else if file_type.is_file() {
+    to_path.set_extension(args.transcode.extension());
-            if entry.path().extension().unwrap() != "flac" {
+
-                // println!("not flac input: {:?}", entry.path());
+    let file_src: Element = gmake("filesrc")?;
+    file_src.set_property("location", &path_to_gstring(&from_path))?;
+
+    // encode into a tmp file first, then rename to actuall file name, that way we're writing
+    // "whole" files to the intended file path, ignoring partial files in the mtime check
+    let to_path_tmp = to_path.with_extension("tmp");
+
+    let decodebin: Element = gmake("decodebin")?;
+
+    let src_elems: &[&Element] = &[&file_src, &decodebin];
+
+    let pipeline = gstreamer::Pipeline::new(None);
+
+    pipeline.add_many(src_elems)?;
+    Element::link_many(src_elems)?;
+
+    // downgrade pipeline RC to a weak RC to break the reference cycle
+    let pipeline_weak = pipeline.downgrade();
+
+    let transcode_args = args.transcode.clone();
+
+    let to_path_tmp_clone = to_path_tmp.clone();
+
+    decodebin.connect_pad_added(move |decodebin, src_pad| {
+        let insert_sink = || -> Result<()> {
+            let pipeline = match pipeline_weak.upgrade() {
+                Some(pipeline) => pipeline,
+                None => {
+                    // pipeline already destroyed... ignoring
                     return Ok(());
                 }
-            fs::create_dir_all(&output)?;
+            };
-            new_output.set_extension("opus");
+
-            let in_mtime = new_input.metadata()?.modified()?;
+            let is_audio = src_pad.get_current_caps().and_then(|caps| {
-            let out_mtime = new_output.metadata().and_then(|md| md.modified());
+                caps.get_structure(0).map(|s| {
-            match out_mtime {
+                    let name = s.get_name();
-                Ok(out_mtime) => {
+                    name.starts_with("audio/")
-                    if out_mtime < in_mtime {
+                })
-                        transcode(new_input.as_ref(), new_output.as_ref())?;
+            });
+            match is_audio {
+                None => {
+                    return Err(Error::msg(format!(
+                        "Failed to get media type from pad {}",
+                        src_pad.get_name()
+                    )));
                 }
+                Some(false) => {
+                    // not audio pad... ignoring
+                    return Ok(());
                 }
-                Err(e) => {
+                Some(true) => {}
-                    if e.kind() == io::ErrorKind::NotFound {
+            }
-                        transcode(new_input.as_ref(), new_output.as_ref())?;
+
+            let resample: Element = gmake("audioresample")?;
+            // quality from 0 to 10
+            resample.set_property("quality", &7)?;
+
+            let mut dest_elems = vec![
+                resample,
+                // `audioconvert` converts audio format, bitdepth, ...
+                gmake("audioconvert")?,
+            ];
+
+            match &transcode_args {
+                config::Transcode::Opus {
+                    bitrate,
+                    bitrate_type,
+                } => {
+                    let encoder: Element = gmake("opusenc")?;
+                    encoder.set_property(
+                        "bitrate",
+                        &i32::from(*bitrate)
+                            .checked_mul(1_000)
+                            .context("bitrate overflowed")?,
+                    )?;
+                    encoder.set_property_from_str(
+                        "bitrate-type",
+                        match bitrate_type {
+                            config::BitrateType::Vbr => "1",
+                            config::BitrateType::Cbr => "0",
+                        },
+                    );
+
+                    dest_elems.push(encoder);
+                    dest_elems.push(gmake("oggmux")?);
+                }
+                config::Transcode::Mp3 {
+                    bitrate,
+                    bitrate_type,
+                } => {
+                    let encoder: Element = gmake("lamemp3enc")?;
+                    // target: "1" = "bitrate"
+                    encoder.set_property_from_str("target", "1");
+                    encoder.set_property("bitrate", &i32::from(*bitrate))?;
+                    encoder.set_property(
+                        "cbr",
+                        match bitrate_type {
+                            config::BitrateType::Vbr => &false,
+                            config::BitrateType::Cbr => &true,
+                        },
+                    )?;
+
+                    dest_elems.push(encoder);
+                    dest_elems.push(gmake("id3v2mux")?);
+                }
+            };
+
+            let file_dest: gstreamer_base::BaseSink = gmake("filesink")?;
+            file_dest.set_property("location", &path_to_gstring(&to_path_tmp_clone))?;
+            file_dest.set_sync(false);
+            dest_elems.push(file_dest.upcast());
+
+            let dest_elem_refs: Vec<_> = dest_elems.iter().collect();
+            pipeline.add_many(&dest_elem_refs)?;
+            Element::link_many(&dest_elem_refs)?;
+
+            for e in &dest_elems {
+                e.sync_state_with_parent()?;
+            }
+
+            let sink_pad = dest_elems
+                .get(0)
+                .unwrap()
+                .get_static_pad("sink")
+                .expect("1. dest element has no sinkpad");
+            src_pad.link(&sink_pad)?;
+
+            Ok(())
+        };
+
+        if let Err(err) = insert_sink() {
+            let details = gstreamer::Structure::builder("error-details")
+                .field("error", &GBoxErrorWrapper::new(err))
+                .build();
+
+            gst_element_error!(
+                decodebin,
+                gstreamer::LibraryError::Failed,
+                ("Failed to insert sink"),
+                details: details
+            );
+        }
+    });
+
+    let bus = pipeline.get_bus().context("pipe get bus")?;
+
+    fs::create_dir_all(
+        to_path
+            .parent()
+            .with_context(|| format!("could not get parent dir for {}", to_path.display()))?,
+    )
+    .await?;
+
+    rm_file_on_err(&to_path_tmp, async {
+        pipeline
+            .set_state(gstreamer::State::Playing)
+            .context("Unable to set the pipeline to the `Playing` state")?;
+
+        let stream_processor = async {
+            bus.stream()
+                .map::<Result<bool>, _>(|msg| {
+                    use gstreamer::MessageView;
+
+                    match msg.view() {
+                        // MessageView::Progress() => {
+
+                        // }
+                        MessageView::Eos(..) => {
+                            // we need to actively stop pulling the stream, that's because stream will
+                            // never end despite yielding an `Eos` message
+                            Ok(false)
+                        }
+                        MessageView::Error(err) => {
+                            pipeline.set_state(gstreamer::State::Null).context(
+                                "Unable to set the pipeline to the `Null` state, after error",
+                            )?;
+
+                            let err = err
+                                .get_details()
+                                .and_then(|details| {
+                                    if details.get_name() != "error-details" {
+                                        return None;
+                                    }
+
+                                    let err = details
+                                        .get::<&GBoxErrorWrapper>("error")
+                                        .unwrap()
+                                        .map(|err| err.clone().into())
+                                        .expect("error-details message without actual error");
+                                    Some(err)
+                                })
+                                .unwrap_or_else(|| {
+                                    GErrorMessage {
+                                        src: msg
+                                            .get_src()
+                                            .map(|s| String::from(s.get_path_string()))
+                                            .unwrap_or_else(|| String::from("None")),
+                                        error: err.get_error().to_string(),
+                                        debug: err.get_debug(),
+                                        source: err.get_error(),
+                                    }
+                                    .into()
+                                });
+                            Err(err)
+                        }
+                        _ => Ok(true),
+                    }
+                })
+                .take_while(|e| {
+                    if let Ok(false) = e {
+                        futures::future::ready(false)
                     } else {
-                        return Err(e.into());
+                        futures::future::ready(true)
+                    }
+                })
+                .try_for_each(|_| futures::future::ready(Ok(())))
+                .await
+                .context("failed converting")?;
+
+            Result::<_>::Ok(())
+        };
+        pin_mut!(stream_processor);
+
+        let mut progress_interval = interval(Duration::from_millis(ui::UPDATE_INTERVAL_MILLIS / 2));
+        let progress_processor = async {
+            use gstreamer::ClockTime;
+
+            loop {
+                progress_interval.tick().await;
+
+                let dur = decodebin
+                    .query_duration::<ClockTime>()
+                    .and_then(|time| time.nanoseconds());
+
+                let ratio = dur.and_then(|dur| {
+                    if dur == 0 {
+                        return None;
+                    }
+
+                    let pos = decodebin
+                        .query_position::<ClockTime>()
+                        .and_then(|time| time.nanoseconds());
+
+                    pos.map(|pos| {
+                        let ratio = pos as f64 / dur as f64;
+                        ratio.max(0.0).min(1.0)
+                    })
+                });
+
+                if let Some(ratio) = ratio {
+                    queue.push(ui::Msg::TaskProgress { id: task_id, ratio });
                 }
             }
-            }
+
-        } else {
+            #[allow(unreachable_code)]
-            Err(format!(
+            Result::<_>::Ok(())
-                "Unsupported file type `{:?}` (maybe symlink?)",
+        };
-                new_input
+        pin_mut!(progress_processor);
-            ))?;
+
-        }
+        future::try_select(stream_processor, progress_processor)
+            .await
+            .map_err(|err| err.factor_first().0)?;
+
+        pipeline
+            .set_state(gstreamer::State::Null)
+            .context("Unable to set the pipeline to the `Null` state")?;
+
+        fs::rename(&to_path_tmp, &to_path).await?;
+
         Ok(())
     })
+    .await
 }
 
-fn main() -> Result<(), Error> {
+async fn rm_file_on_err<F, T>(path: &Path, f: F) -> F::Output
-    ffmpeg::init()?;
+where
-
+    F: Future<Output = Result<T>>,
-    let input = env::args().nth(1).expect("missing input");
+{
-    let output = env::args().nth(2).expect("missing output");
+    match f.await {
-    transcode_path(input.as_ref(), output.as_ref())?;
+        Err(err) => match fs::remove_file(path).await {
-
+            Ok(..) => Err(err),
-    Ok(())
+            Err(fs_err) if fs_err.kind() == std::io::ErrorKind::NotFound => Err(err),
+            Err(fs_err) => {
+                let err = err
+                    .context(fs_err)
+                    .context(format!("removing {} failed", path.display()));
+                Err(err)
+            }
+        },
+        res @ Ok(..) => res,
+    }
+}
+
+fn path_to_bytes(path: &Path) -> Cow<'_, [u8]> {
+    // https://stackoverflow.com/a/59224987/5572146
+    #[cfg(unix)]
+    {
+        use std::os::unix::ffi::OsStrExt;
+        Cow::Borrowed(path.as_os_str().as_bytes())
+    }
+
+    #[cfg(windows)]
+    {
+        // NOT TESTED
+        // FIXME: test and post answer to https://stackoverflow.com/questions/38948669
+        use std::os::windows::ffi::OsStrExt;
+        let buf: Vec<u8> = path
+            .as_os_str()
+            .encode_wide()
+            .map(|char| char.to_ne_bytes())
+            .flatten()
+            .collect();
+        Cow::Owned(buf)
+    }
+}
+
+fn path_to_gstring(path: &Path) -> GString {
+    let buf = path_to_bytes(path);
+    ffi::CString::new(buf)
+        .expect("Path contained null byte")
+        .into()
 }

src/ui.rs

@@ -0,0 +1,283 @@
+use crate::ConversionArgs;
+use anyhow::{Context, Result};
+use futures::Future;
+use std::{
+    borrow::Cow, cell::RefCell, collections::HashMap, io, mem, path::PathBuf, rc::Rc,
+    time::Duration,
+};
+use tokio::{task, time::interval};
+use tui::{backend::CrosstermBackend, Terminal};
+
+pub const UPDATE_INTERVAL_MILLIS: u64 = 100;
+
+#[derive(Debug)]
+pub enum Msg {
+    Init { task_len: usize, log_path: PathBuf },
+    Exit,
+    TaskStart { id: usize, args: ConversionArgs },
+    TaskEnd { id: usize },
+    TaskProgress { id: usize, ratio: f64 },
+    TaskError { id: usize },
+}
+
+#[derive(Debug, Clone)]
+pub struct MsgQueue {
+    inner: Rc<RefCell<Vec<Msg>>>,
+}
+
+impl MsgQueue {
+    fn new() -> MsgQueue {
+        MsgQueue {
+            inner: Rc::new(RefCell::new(Vec::new())),
+        }
+    }
+
+    pub fn push(&self, msg: Msg) {
+        self.inner.borrow_mut().push(msg);
+    }
+
+    fn swap_inner(&self, other: &mut Vec<Msg>) {
+        let mut inner = self.inner.borrow_mut();
+        mem::swap(&mut *inner, other)
+    }
+}
+
+struct State {
+    terminal: Terminal<CrosstermBackend<io::Stdout>>,
+    log_path: Option<PathBuf>,
+    task_len: Option<usize>,
+    ended_tasks: usize,
+    running_tasks: HashMap<usize, Task>,
+    has_rendered: bool,
+    has_errored: bool,
+}
+
+impl State {
+    fn new() -> Result<State> {
+        let terminal = Terminal::new(CrosstermBackend::new(io::stdout()))
+            .context("Unable to create ui terminal")?;
+
+        Ok(State {
+            terminal,
+            log_path: None,
+            task_len: None,
+            ended_tasks: 0,
+            running_tasks: HashMap::new(),
+            has_rendered: false,
+            has_errored: false,
+        })
+    }
+
+    fn process_msg(&mut self, msg: Msg) -> Result<bool> {
+        match msg {
+            Msg::Init { task_len, log_path } => {
+                self.task_len = Some(task_len);
+                self.log_path = Some(log_path);
+            }
+            Msg::Exit => return Ok(false),
+            Msg::TaskStart { id, args } => {
+                self.running_tasks.insert(
+                    id,
+                    Task {
+                        id,
+                        ratio: None,
+                        args,
+                    },
+                );
+            }
+            Msg::TaskEnd { id } => {
+                self.running_tasks
+                    .remove(&id)
+                    .context("unable to remove finished task; could't find task")?;
+                self.ended_tasks += 1;
+            }
+            Msg::TaskProgress { id, ratio } => {
+                let mut task = self
+                    .running_tasks
+                    .get_mut(&id)
+                    .context("Unable to update task progress; could't find task")?;
+                task.ratio = Some(ratio);
+            }
+            Msg::TaskError { id } => {
+                // TODO
+                self.running_tasks
+                    .remove(&id)
+                    .context("unable to remove errored task; could't find task")?;
+                self.ended_tasks += 1;
+                self.has_errored = true;
+            }
+        }
+
+        Ok(true)
+    }
+
+    fn render(&mut self) -> Result<()> {
+        use tui::{
+            layout::{Constraint, Direction, Layout, Rect},
+            style::{Color, Modifier, Style},
+            text::Text,
+            widgets::{Block, Borders, Gauge, Paragraph},
+        };
+
+        let task_len = if let Some(task_len) = self.task_len {
+            task_len
+        } else {
+            return Ok(());
+        };
+
+        if task_len == 0 {
+            return Ok(());
+        }
+
+        let tasks_ended = self.ended_tasks;
+
+        let mut running_tasks: Vec<_> = self.running_tasks.values().cloned().collect();
+
+        running_tasks.sort_by_key(|task| task.id);
+
+        if !self.has_rendered {
+            self.terminal.clear().context("cleaning ui failed")?;
+            self.has_rendered = true;
+        }
+
+        let error_text = match self.has_errored {
+            true => {
+                let text: Cow<'static, str> = self
+                    .log_path
+                    .as_ref()
+                    .map(|lp| {
+                        let text = format!("Error(s) occurred and were logged to {}", lp.display());
+                        Cow::Owned(text)
+                    })
+                    .unwrap_or_else(|| Cow::Borrowed("Error(s) occurred"));
+                Some(text)
+            }
+            false => None,
+        };
+
+        self.terminal
+            .draw(|f| {
+                let chunks = Layout::default()
+                    .direction(Direction::Vertical)
+                    .margin(1)
+                    .constraints([Constraint::Percentage(90), Constraint::Percentage(10)].as_ref())
+                    .split(f.size());
+
+                let mut task_rect = chunks[0];
+
+                if error_text.is_some() {
+                    task_rect.height -= 3;
+                }
+
+                for (row, task) in running_tasks
+                    .into_iter()
+                    .take(task_rect.height as usize / 2)
+                    .enumerate()
+                {
+                    f.render_widget(
+                        Gauge::default()
+                            .label(task.args.rel_from_path.to_string_lossy().as_ref())
+                            .gauge_style(
+                                Style::default()
+                                    .fg(Color::White)
+                                    .bg(Color::Black)
+                                    .add_modifier(Modifier::ITALIC),
+                            )
+                            .ratio(task.ratio.unwrap_or(0.0)),
+                        Rect::new(
+                            task_rect.x,
+                            task_rect.y + row as u16 * 2,
+                            task_rect.width,
+                            1,
+                        ),
+                    );
+                }
+
+                if let Some(error_text) = error_text {
+                    f.render_widget(
+                        Paragraph::new(Text::raw(error_text)).style(
+                            Style::default()
+                                .fg(Color::Red)
+                                .bg(Color::Black)
+                                .add_modifier(Modifier::BOLD),
+                        ),
+                        Rect::new(task_rect.x, task_rect.height + 1, task_rect.width, 2),
+                    );
+                }
+
+                f.render_widget(
+                    Gauge::default()
+                        .block(
+                            Block::default()
+                                .borders(Borders::ALL)
+                                .title("Overall Progress"),
+                        )
+                        .gauge_style(
+                            Style::default()
+                                .fg(Color::White)
+                                .bg(Color::Black)
+                                .add_modifier(Modifier::ITALIC),
+                        )
+                        .ratio(tasks_ended as f64 / task_len as f64),
+                    chunks[1],
+                );
+            })
+            .context("rendering ui failed")?;
+
+        Ok(())
+    }
+}
+
+#[derive(Debug, Clone)]
+struct Task {
+    id: usize,
+    ratio: Option<f64>,
+    args: ConversionArgs,
+}
+
+pub fn init() -> (MsgQueue, impl Future<Output = Result<()>>) {
+    let queue = MsgQueue::new();
+
+    let queue_clone = queue.clone();
+    let fut = async move {
+        let mut interval = interval(Duration::from_millis(UPDATE_INTERVAL_MILLIS));
+        let mut wrapped = Some((Vec::new(), State::new()?));
+
+        loop {
+            interval.tick().await;
+
+            let (mut current_queue, mut state) = wrapped.take().context("`wrapped` is None")?;
+
+            queue_clone.swap_inner(&mut current_queue);
+
+            let render_res = task::spawn_blocking(move || -> Result<_> {
+                let mut exit = false;
+                for msg in current_queue.drain(..) {
+                    if !state.process_msg(msg)? {
+                        exit = true;
+                    }
+                }
+
+                state.render()?;
+
+                if exit {
+                    Ok(None)
+                } else {
+                    Ok(Some((current_queue, state)))
+                }
+            })
+            .await
+            .context("ui update task failed")?
+            .context("ui update failed")?;
+
+            match render_res {
+                Some(s) => wrapped = Some(s),
+                None => break,
+            }
+        }
+
+        Result::<_>::Ok(())
+    };
+
+    (queue, fut)
+}