651 lines
15 KiB
Rust
651 lines
15 KiB
Rust
mod config;
|
|
mod tag;
|
|
mod ui;
|
|
|
|
use crate::config::{Config, Transcode};
|
|
use anyhow::{Context, Error, Result};
|
|
use futures::{pin_mut, prelude::*};
|
|
use glib::{GBoxed, GString};
|
|
use gstreamer::{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(Clone, Debug, GBoxed)]
|
|
#[gboxed(type_name = "GBoxErrorWrapper")]
|
|
struct GBoxErrorWrapper(Arc<Error>);
|
|
|
|
impl GBoxErrorWrapper {
|
|
fn new(err: Error) -> Self {
|
|
GBoxErrorWrapper(Arc::new(err))
|
|
}
|
|
}
|
|
|
|
impl StdError for GBoxErrorWrapper {
|
|
fn source(&self) -> Option<&(dyn StdError + 'static)> {
|
|
self.0.source()
|
|
}
|
|
}
|
|
|
|
impl fmt::Display for GBoxErrorWrapper {
|
|
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> StdResult<(), fmt::Error> {
|
|
self.0.fmt(f)
|
|
}
|
|
}
|
|
|
|
#[derive(Debug, derive_more::Display, derive_more::Error)]
|
|
#[display(fmt = "Received error from {}: {} (debug: {:?})", src, error, debug)]
|
|
struct GErrorMessage {
|
|
src: String,
|
|
error: String,
|
|
debug: Option<String>,
|
|
source: glib::Error,
|
|
}
|
|
|
|
fn gmake<T: IsA<Element>>(factory_name: &str) -> Result<T> {
|
|
let res = gstreamer::ElementFactory::make(factory_name, None)
|
|
.with_context(|| format!("Could not make gstreamer Element \"{}\"", factory_name))?
|
|
.downcast()
|
|
.ok()
|
|
.with_context(|| {
|
|
format!(
|
|
"Could not cast gstreamer Element \"{}\" into `{}`",
|
|
factory_name,
|
|
std::any::type_name::<T>()
|
|
)
|
|
})?;
|
|
Ok(res)
|
|
}
|
|
|
|
#[derive(Debug, Clone)]
|
|
pub struct ConversionArgs {
|
|
rel_from_path: PathBuf,
|
|
transcode: Transcode,
|
|
}
|
|
|
|
fn get_conversion_args(config: &Config) -> impl Iterator<Item = Result<ConversionArgs>> + '_ {
|
|
walkdir::WalkDir::new(&config.from)
|
|
.into_iter()
|
|
.filter_map(|e| e.ok())
|
|
.filter(|e| e.file_type().is_file())
|
|
.map(move |e| -> Result<Option<ConversionArgs>> {
|
|
let from_bytes = path_to_bytes(e.path());
|
|
|
|
let transcode = config
|
|
.matches
|
|
.iter()
|
|
.filter(|m| {
|
|
m.regexes
|
|
.iter()
|
|
.any(|regex| regex.is_match(from_bytes.as_ref()))
|
|
})
|
|
.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())
|
|
}
|
|
|
|
#[tokio::main(flavor = "current_thread")]
|
|
async fn main() -> Result<()> {
|
|
task::LocalSet::new()
|
|
.run_until(async move {
|
|
let (ui_queue, ui_fut) = ui::init();
|
|
|
|
let main_handle = async move {
|
|
let ok = task::spawn_local(main_loop(ui_queue))
|
|
.await
|
|
.context("Main task failed")??;
|
|
Result::<_>::Ok(ok)
|
|
};
|
|
|
|
let ui_handle = async move {
|
|
let ok = task::spawn_local(ui_fut)
|
|
.await
|
|
.context("Ui task failed")?
|
|
.context("Ui failed")?;
|
|
Result::<_>::Ok(ok)
|
|
};
|
|
|
|
future::try_join(main_handle, ui_handle).await?;
|
|
Ok(())
|
|
})
|
|
.await
|
|
}
|
|
|
|
async fn main_loop(ui_queue: ui::MsgQueue) -> Result<()> {
|
|
let (config, conv_args) = task::spawn_blocking(|| -> Result<_> {
|
|
gstreamer::init()?;
|
|
gstreamer::tags::register::<tag::MbArtistId>();
|
|
gstreamer::tags::register::<tag::MbAlbumArtistId>();
|
|
|
|
let config = config::config().context("Could not get the config")?;
|
|
|
|
let conv_args = get_conversion_args(&config)
|
|
.collect::<Result<Vec<_>>>()
|
|
.context("Failed loading dir structure")?;
|
|
|
|
Ok((config, conv_args))
|
|
})
|
|
.await
|
|
.context("Init task failed")??;
|
|
|
|
let log_path = Path::new(".")
|
|
.canonicalize()
|
|
.context("Unable to canonicalize path to log file")?
|
|
.join("audio-conv.log");
|
|
|
|
ui_queue.push(ui::Msg::Init {
|
|
task_len: conv_args.len(),
|
|
log_path: log_path.clone(),
|
|
});
|
|
|
|
let concurrent_jobs = config.jobs.unwrap_or_else(|| num_cpus::get());
|
|
|
|
stream::iter(conv_args.into_iter().enumerate())
|
|
.map(Ok)
|
|
.try_for_each_concurrent(concurrent_jobs, |(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!(
|
|
"Transcoding failed for {}",
|
|
args.rel_from_path.display()
|
|
));
|
|
|
|
let mut log_file = match fs::OpenOptions::new()
|
|
.create(true)
|
|
.append(true)
|
|
.open(log_path)
|
|
.await
|
|
{
|
|
Ok(log_file) => log_file,
|
|
Err(fs_err) => {
|
|
let err = err.context(fs_err).context("Unable to open log file");
|
|
return Err(err);
|
|
}
|
|
};
|
|
|
|
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 });
|
|
}
|
|
}
|
|
|
|
Result::<_>::Ok(())
|
|
}
|
|
})
|
|
.await?;
|
|
|
|
ui_queue.push(ui::Msg::Exit);
|
|
|
|
Ok(())
|
|
}
|
|
|
|
async fn transcode(
|
|
config: &Config,
|
|
args: &ConversionArgs,
|
|
task_id: usize,
|
|
queue: &ui::MsgQueue,
|
|
) -> Result<()> {
|
|
let from_path = config.from.join(&args.rel_from_path);
|
|
let mut to_path = config.to.join(&args.rel_from_path);
|
|
|
|
fs::create_dir_all(
|
|
to_path
|
|
.parent()
|
|
.with_context(|| format!("Could not get parent dir for {}", to_path.display()))?,
|
|
)
|
|
.await?;
|
|
|
|
// 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");
|
|
|
|
rm_file_on_err(&to_path_tmp, async {
|
|
match args.transcode {
|
|
Transcode::Copy => {
|
|
fs::copy(&from_path, &to_path_tmp).await.with_context(|| {
|
|
format!(
|
|
"Could not copy file from {} to {}",
|
|
from_path.display(),
|
|
to_path_tmp.display()
|
|
)
|
|
})?;
|
|
}
|
|
_ => {
|
|
to_path.set_extension(args.transcode.extension());
|
|
|
|
transcode_gstreamer(
|
|
&from_path,
|
|
&to_path_tmp,
|
|
args.transcode.clone(),
|
|
task_id,
|
|
queue,
|
|
)
|
|
.await?
|
|
}
|
|
}
|
|
|
|
fs::rename(&to_path_tmp, &to_path).await.with_context(|| {
|
|
format!(
|
|
"Could not rename temporary file {} to {}",
|
|
to_path_tmp.display(),
|
|
to_path.display()
|
|
)
|
|
})
|
|
})
|
|
.await
|
|
}
|
|
|
|
async fn transcode_gstreamer(
|
|
from_path: &Path,
|
|
to_path: &Path,
|
|
transcode: Transcode,
|
|
task_id: usize,
|
|
queue: &ui::MsgQueue,
|
|
) -> Result<()> {
|
|
let file_src: Element = gmake("filesrc")?;
|
|
file_src.set_property("location", &path_to_gstring(&from_path))?;
|
|
|
|
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 to_path_clone = to_path.to_owned();
|
|
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(());
|
|
}
|
|
};
|
|
|
|
let is_audio = src_pad.current_caps().and_then(|caps| {
|
|
caps.structure(0).map(|s| {
|
|
let name = s.name();
|
|
name.starts_with("audio/")
|
|
})
|
|
});
|
|
match is_audio {
|
|
None => {
|
|
return Err(Error::msg(format!(
|
|
"Failed to get media type from pad {}",
|
|
src_pad.name()
|
|
)));
|
|
}
|
|
Some(false) => {
|
|
// not audio pad... ignoring
|
|
return Ok(());
|
|
}
|
|
Some(true) => {}
|
|
}
|
|
|
|
let resample: Element = gmake("audioresample")?;
|
|
// quality from 0 to 10
|
|
resample.set_property("quality", &10)?;
|
|
|
|
let mut dest_elems = vec![
|
|
resample,
|
|
// `audioconvert` converts audio format, bitdepth, ...
|
|
gmake("audioconvert")?,
|
|
];
|
|
|
|
match &transcode {
|
|
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")?);
|
|
}
|
|
|
|
Transcode::Flac { compression } => {
|
|
let encoder: Element = gmake("flacenc")?;
|
|
encoder.set_property_from_str("quality", &compression.to_string());
|
|
dest_elems.push(encoder);
|
|
}
|
|
|
|
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")?);
|
|
}
|
|
|
|
Transcode::Copy => {
|
|
// already handled outside this fn
|
|
unreachable!();
|
|
}
|
|
};
|
|
|
|
let file_dest: gstreamer_base::BaseSink = gmake("filesink")?;
|
|
file_dest.set_property("location", &path_to_gstring(&to_path_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()
|
|
.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();
|
|
|
|
element_error!(
|
|
decodebin,
|
|
gstreamer::LibraryError::Failed,
|
|
("Failed to insert sink"),
|
|
details: details
|
|
);
|
|
}
|
|
});
|
|
|
|
let bus = pipeline.bus().context("Could not get bus for pipeline")?;
|
|
|
|
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) => {
|
|
let pipe_stop_res = pipeline.set_state(gstreamer::State::Null);
|
|
|
|
let err: Error = err
|
|
.details()
|
|
.and_then(|details| {
|
|
if details.name() != "error-details" {
|
|
return None;
|
|
}
|
|
|
|
let err = details
|
|
.get::<&GBoxErrorWrapper>("error")
|
|
.unwrap()
|
|
.clone()
|
|
.into();
|
|
Some(err)
|
|
})
|
|
.unwrap_or_else(|| {
|
|
GErrorMessage {
|
|
src: msg
|
|
.src()
|
|
.map(|s| String::from(s.path_string()))
|
|
.unwrap_or_else(|| String::from("None")),
|
|
error: err.error().to_string(),
|
|
debug: err.debug(),
|
|
source: err.error(),
|
|
}
|
|
.into()
|
|
});
|
|
|
|
if let Err(pipe_err) = pipe_stop_res {
|
|
let err = err.context(pipe_err).context(
|
|
"Unable to set the pipeline to the `Null` state, after error",
|
|
);
|
|
Err(err)
|
|
} else {
|
|
Err(err)
|
|
}
|
|
}
|
|
_ => Ok(true),
|
|
}
|
|
})
|
|
.take_while(|e| {
|
|
if let Ok(false) = e {
|
|
futures::future::ready(false)
|
|
} else {
|
|
futures::future::ready(true)
|
|
}
|
|
})
|
|
.try_for_each(|_| futures::future::ready(Ok(())))
|
|
.await?;
|
|
|
|
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>()
|
|
.map(|time| time.nseconds());
|
|
|
|
let ratio = dur.and_then(|dur| {
|
|
if dur == 0 {
|
|
return None;
|
|
}
|
|
|
|
let pos = decodebin
|
|
.query_position::<ClockTime>()
|
|
.map(|time| time.nseconds());
|
|
|
|
pos.map(|pos| {
|
|
let ratio = pos as f64 / dur as f64;
|
|
ratio.clamp(0.0, 1.0)
|
|
})
|
|
});
|
|
|
|
if let Some(ratio) = ratio {
|
|
queue.push(ui::Msg::TaskProgress { id: task_id, ratio });
|
|
}
|
|
}
|
|
|
|
#[allow(unreachable_code)]
|
|
Result::<_>::Ok(())
|
|
};
|
|
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")?;
|
|
|
|
Ok(())
|
|
}
|
|
|
|
async fn rm_file_on_err<F, T>(path: &Path, f: F) -> Result<T>
|
|
where
|
|
F: Future<Output = Result<T>>,
|
|
{
|
|
match f.await {
|
|
Err(err) => match fs::remove_file(path).await {
|
|
Ok(()) => Err(err),
|
|
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 file {} 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()
|
|
}
|