diff --git a/crates/lib/mysql_queries/src/queries/media_files/insert_media_file_from_zero_shot_tts.rs b/crates/lib/mysql_queries/src/queries/media_files/insert_media_file_from_zero_shot_tts.rs index 6c860e8d30..60529bdc0c 100644 --- a/crates/lib/mysql_queries/src/queries/media_files/insert_media_file_from_zero_shot_tts.rs +++ b/crates/lib/mysql_queries/src/queries/media_files/insert_media_file_from_zero_shot_tts.rs @@ -33,7 +33,7 @@ pub struct InsertArgs<'a> { pub worker_cluster: &'a str, } -pub async fn insert_media_file_from_face_zero_shot( +pub async fn insert_media_file_from_zero_shot( args: InsertArgs<'_> ) -> AnyhowResult<(MediaFileToken, u64)> { diff --git a/crates/lib/mysql_queries/src/queries/voice_designer/voices/create_voice.rs b/crates/lib/mysql_queries/src/queries/voice_designer/voices/create_voice.rs index 6bc926bb6d..64242257b6 100644 --- a/crates/lib/mysql_queries/src/queries/voice_designer/voices/create_voice.rs +++ b/crates/lib/mysql_queries/src/queries/voice_designer/voices/create_voice.rs @@ -10,6 +10,14 @@ use tokens::tokens::users::UserToken; use crate::queries::generic_synthetic_ids::transactional_increment_generic_synthetic_id::transactional_increment_generic_synthetic_id; + + +// Model category? Zero shot? +// Model type? end to end TTS / VC vs cascading TTS / VC +// Model version? 1? +// Encoding Type? npz ? +// + pub struct CreateVoiceArgs<'a> { pub dataset_token: &'a ZsVoiceDatasetToken, diff --git a/crates/service/job/inference_job/src/job/job_types/tts/vall_e_x/process_job.rs b/crates/service/job/inference_job/src/job/job_types/tts/vall_e_x/process_job.rs index 7157958381..6026a473b9 100644 --- a/crates/service/job/inference_job/src/job/job_types/tts/vall_e_x/process_job.rs +++ b/crates/service/job/inference_job/src/job/job_types/tts/vall_e_x/process_job.rs @@ -9,6 +9,7 @@ use log::{ error, info, warn }; use mysql_queries::queries::media_files::get_media_file::MediaFile; use mysql_queries::queries::media_files::insert_media_file_from_zero_shot_tts::InsertArgs; use mysql_queries::queries::media_files::insert_media_file_from_zero_shot_tts::insert_media_file_from_face_zero_shot; +use mysql_queries::queries::voice_designer::datasets::get_dataset::ZsDataset; use mysql_queries::queries::voice_designer::datasets::get_dataset::get_dataset_by_token; use mysql_queries::queries::voice_designer::datasets::list_datasets::DatasetRecordForList; use mysql_queries::queries::voice_designer::voice_samples::list_dataset_samples_for_dataset_token::DatasetSampleRecordForList; @@ -16,7 +17,11 @@ use mysql_queries::queries::voice_designer::voice_samples::list_dataset_samples_ use mysql_queries::queries::voice_designer::voice_samples::get_dataset_sample; use mysql_queries::queries::voice_designer::voice_samples::get_dataset_sample::get_dataset_sample_by_token; use mysql_queries::queries::voice_designer::voice_samples::list_dataset_samples_for_dataset_token::list_dataset_samples_for_dataset_token; +use mysql_queries::queries::voice_designer::voices::create_voice; +use mysql_queries::queries::voice_designer::voices::create_voice::create_voice; +use mysql_queries::queries::voice_designer::voices::create_voice::CreateVoiceArgs; use mysql_queries::queries::voice_designer::voices::get_voice::get_voice_by_token; +use tokens::tokens::users::UserToken; use std::path::PathBuf; use std::time::Instant; use mysql_queries::queries::generic_inference::job::list_available_generic_inference_jobs::AvailableInferenceJob; @@ -27,7 +32,8 @@ use crate::job::job_loop::job_success_result::ResultEntity; use crate::job::job_loop::process_single_job_error::ProcessSingleJobError; use crate::job::job_types::tts::vall_e_x::validate_job::validate_job; -use crate::job::job_types::tts::vall_e_x::vall_e_x_inference_command::CreateVoiceArgs; +use crate::job::job_types::tts::vall_e_x::vall_e_x_inference_command::CreateVoiceInferenceArgs; + use crate::job_dependencies::JobDependencies; use crate::job::job_types::tts::vall_e_x::vall_e_x_inference_command::InferenceArgs; use cloud_storage::bucket_path_unifier::BucketPathUnifier; @@ -35,17 +41,19 @@ use cloud_storage::bucket_path_unifier::BucketPathUnifier; use super::validate_job::JobType; const BUCKET_FILE_PREFIX: &str = "fakeyou_"; -const BUCKET_FILE_EXTENSION: &str = ".wav"; -const MIME_TYPE: &str = "audio/wav"; +const BUCKET_FILE_EXTENSION: &str = ".npz"; +const MIME_TYPE: &str = "application/x-binary"; + pub struct VoiceFile { pub filesystem_path: PathBuf, } - +const BUCKET_FILE_PREFIX: &str = "fakeyou_"; +const BUCKET_FILE_EXTENSION: &str = ".wav"; +const MIME_TYPE: &str = "audio/wav"; pub struct AudioFile { pub filesystem_path: PathBuf, } - pub async fn download_voice_embedding_from_hash( bucket_hash: &str, name: &str, @@ -65,17 +73,16 @@ pub async fn download_voice_embedding_from_hash( let voice_file = VoiceFile { filesystem_path: PathBuf::from(&path.clone()), }; - Ok(voice_file) } -pub async fn download_audio_from_hash( +pub async fn download_audio_from_hash( bucket_hash: &str, name: &str, private_bucket_client: &BucketClient, path: &PathBuf -)-> Result { +) -> Result { let unifer = BucketPathUnifier::default_paths(); let object_path = unifer.zero_shot_tts_speaker_encoding(bucket_hash, 0); @@ -93,7 +100,6 @@ pub async fn download_audio_from_hash( Ok(audio_file) } - // This will download everything get into the root host OS then ... will invoke inference using the pathes from the files invoked pub struct VALLEXProcessJobArgs<'a> { pub job_dependencies: &'a JobDependencies, @@ -104,7 +110,6 @@ pub async fn process_create_voice( args: VALLEXProcessJobArgs<'_>, dataset_token: String ) -> Result { - let deps = args.job_dependencies; let job = args.job; let mysql_pool = &deps.mysql_pool; @@ -113,92 +118,163 @@ pub async fn process_create_voice( .new_generic_inference(job.inference_job_token.as_str()) .map_err(|e| ProcessSingleJobError::Other(anyhow!(e)))?; - let voice_dataset_token = tokens::tokens::zs_voice_datasets::ZsVoiceDatasetToken(dataset_token); + let voice_dataset_token = tokens::tokens::zs_voice_datasets::ZsVoiceDatasetToken(dataset_token); - // download data set files into a temp directory - let work_temp_dir = format!("/tmp/temp_zeroshot_create_voice_{}", job.id.0); - // NB: TempDir exists until it goes out of scope, at which point it should delete from filesystem. - let work_temp_dir = args.job_dependencies.fs.scoped_temp_dir_creator_for_work - .new_tempdir(&work_temp_dir) - .map_err(|e| ProcessSingleJobError::from_io_error(e))?; - - let workdir = work_temp_dir.path().to_path_buf(); + // download data set files into a temp directory + let work_temp_dir = format!("/tmp/temp_zeroshot_create_voice_{}", job.id.0); + // NB: TempDir exists until it goes out of scope, at which point it should delete from filesystem. + let work_temp_dir = args.job_dependencies.fs.scoped_temp_dir_creator_for_work + .new_tempdir(&work_temp_dir) + .map_err(|e| ProcessSingleJobError::from_io_error(e))?; - let result = list_dataset_samples_for_dataset_token(&voice_dataset_token, false, &mysql_pool).await; + let workdir = work_temp_dir.path().to_path_buf(); - let dataset: Vec; + let voice_dataset = get_dataset_by_token(&voice_dataset_token, false, &mysql_pool).await; + let single_dataset: ZsDataset; - match result { - Ok(val) => { - dataset = val; - }, - Err(e) => { - return Err(ProcessSingleJobError::from_anyhow_error(e)); - } + match voice_dataset { + Ok(val) => { + single_dataset = val; } - - let mut downloaded_dataset:Vec = Vec::new(); - - for (index,record) in dataset.iter().enumerate() { - - let prefix: Option<&str> = record.maybe_public_bucket_prefix.as_ref().map(|s| s.as_str()); - let extension: Option<&str> = record.maybe_public_bucket_extension.as_ref().map(|s| s.as_str()); - - let audio_media_file = MediaFileBucketPath::from_object_hash( - &record.public_bucket_directory_hash, - prefix, - extension, - ); - - let file_name_wav = format!("{}.wav", index); - let mut file_path = PathBuf::new(); - file_path.push(workdir.clone()); - file_path.push(file_path.clone()); - - // TODO we might want to catch the error and not include the pathes into download dataset - let _ = deps.public_bucket_client.download_file_to_disk(audio_media_file.to_full_object_pathbuf(), &file_path).await; - - downloaded_dataset.push(file_path.clone()); + Err(e) => { + return Err(ProcessSingleJobError::from_anyhow_error(e)); } + } - // Need to download the models - info!("Download models (if not present)..."); + let result = list_dataset_samples_for_dataset_token( + &voice_dataset_token, + false, + &mysql_pool + ).await; + let dataset: Vec; - for downloader in deps.job_type_details.vall_e_x.downloaders.all_downloaders() { - let result = downloader.download_if_not_on_filesystem( - &args.job_dependencies.private_bucket_client, - &args.job_dependencies.fs.scoped_temp_dir_creator_for_downloads - ).await; - - if let Err(e) = result { - error!("could not download: {:?}", e); - return Err(ProcessSingleJobError::from_anyhow_error(e)); - } + match result { + Ok(val) => { + dataset = val; } + Err(e) => { + return Err(ProcessSingleJobError::from_anyhow_error(e)); + } + } + + let mut downloaded_dataset: Vec = Vec::new(); + + for (index, record) in dataset.iter().enumerate() { + let prefix: Option<&str> = record.maybe_public_bucket_prefix.as_ref().map(|s| s.as_str()); + let extension: Option<&str> = record.maybe_public_bucket_extension + .as_ref() + .map(|s| s.as_str()); + + let audio_media_file = MediaFileBucketPath::from_object_hash( + &record.public_bucket_directory_hash, + prefix, + extension + ); + + let file_name_wav = format!("{}.wav", index); + let mut file_path = PathBuf::new(); + file_path.push(workdir.clone()); + file_path.push(file_path.clone()); + + // TODO we might want to catch the error and not include the pathes into download dataset + let _ = deps.public_bucket_client.download_file_to_disk( + audio_media_file.to_full_object_pathbuf(), + &file_path + ).await; + println!("FilePath to clone voice: {}", file_path); + downloaded_dataset.push(file_path.clone()); + } + + // Need to download the models + info!("Download models (if not present)..."); + + for downloader in deps.job_type_details.vall_e_x.downloaders.all_downloaders() { + let result = downloader.download_if_not_on_filesystem( + &args.job_dependencies.private_bucket_client, + &args.job_dependencies.fs.scoped_temp_dir_creator_for_downloads + ).await; + + if let Err(e) = result { + error!("could not download: {:?}", e); + return Err(ProcessSingleJobError::from_anyhow_error(e)); + } + } job_progress_reporter - .log_status("running inference") - .map_err(|e| ProcessSingleJobError::Other(e))?; + .log_status("running inference") + .map_err(|e| ProcessSingleJobError::Other(e))?; let inference_start_time = Instant::now(); - // TODO: fix + let audio_files = join_paths(path); + + // TODO: fix with proper name + let name: String = single_dataset.title; let output_file_name = String::from("s.npz"); // Run Inference - let command_exit_status = args.job_dependencies.job_type_details.vall_e_x.create_embedding_command.execute_inference( - CreateVoiceArgs { + let command_exit_status = + args.job_dependencies.job_type_details.vall_e_x.create_embedding_command.execute_inference( + job::job_types::tts::vall_e_x::vall_e_x_inference_command::CreateVoiceInferenceArgs { output_embedding_path: &workdir, output_embedding_name: output_file_name, - audio_files: String::from(""), // " " + audio_files: audio_files, stderr_output_file: String::from("zero_shot_create_voice.txt"), } ); - let inference_duration = Instant::now().duration_since(inference_start_time); + // upload media to public bucket - Err(ProcessSingleJobError::Other(anyhow!("Error"))) + info!("Uploading media ..."); + + let result_bucket_location = MediaFileBucketPath::generate_new( + Some(BUCKET_FILE_PREFIX), + Some(BUCKET_FILE_EXTENSION) + ); + + let result_bucket_object_pathbuf = result_bucket_location.to_full_object_pathbuf(); + + let mut finished_file = work_temp_dir.path().to_path_buf(); + finished_file.push(&output_file_name); + + args.job_dependencies.private_bucket_client + .upload_filename_with_content_type( + &result_bucket_object_pathbuf, + &finished_file, + &MIME_TYPE + ) + .await // TODO: We should check the mimetype to make sure bad payloads can't get uploaded + .map_err(|e| ProcessSingleJobError::Other(e))?; + + + // TODO fix + let creator_ip_address = String::from("127.0.0.1"); + let creator_user_token = String::from("creator_user_token"); + + let bucket_hash = String::from("bucket hash"); + // insert record + let voice_token = create_voice(CreateVoiceArgs { + dataset_token: &dataset_token, + model_category: "vc", + model_type: "vall-e-x", + model_version: 0, + model_encoding_type: "encodec", + voice_title: &voice_name, + bucket_hash: &bucket_hash, + maybe_creator_user_token: Some(&creator_user_token), + creator_ip_address: &creator_ip_address, + creator_set_visibility: &Default::default(), + mysql_pool, + }).await; + + Ok(JobSuccessResult { + maybe_result_entity: Some(ResultEntity { + entity_type: InferenceResultType::MediaFile, + entity_token: media_file_token.to_string(), + }), + inference_duration, + }) } pub async fn process_inference_voice( @@ -334,7 +410,7 @@ pub async fn process_inference_voice( job_progress_reporter.log_status("done").map_err(|e| ProcessSingleJobError::Other(e))?; // insert into db the record - let (media_file_token, id) = insert_media_file_from_face_zero_shot(InsertArgs { + let (media_file_token, id) = insert_media_file_from_zero_shot(InsertArgs { pool: &args.job_dependencies.mysql_pool, job: &job, maybe_mime_type: Some(&MIME_TYPE), @@ -390,24 +466,26 @@ pub async fn process_job( } } -fn join_paths(paths: Vec)-> String { - paths.into_iter() - .map(|p| format!("\"{}\"",p.display())) - .collect::>() - .join(" ") +fn join_paths(paths: Vec) -> String { + paths + .into_iter() + .map(|p| format!("\"{}\"", p.display())) + .collect::>() + .join(" ") } mod tests { - use super::*; #[test] - fn test_length_zero() { - let paths = vec![ - PathBuf::from("/home/tensor/code/TTSDockerContainer/Vall-E-mount/input/20.wav"), - PathBuf::from("/home/tensor/code/TTSDockerContainer/Vall-E-mount/input/21.wav") - ]; + fn test_path_build() { + let paths = vec![ + PathBuf::from("/home/tensor/code/TTSDockerContainer/Vall-E-mount/input/20.wav"), + PathBuf::from("/home/tensor/code/TTSDockerContainer/Vall-E-mount/input/21.wav") + ]; - let value = join_paths(paths); - let expected = "\"/home/tensor/code/TTSDockerContainer/Vall-E-mount/input/20.wav\" \"/home/tensor/code/TTSDockerContainer/Vall-E-mount/input/21.wav\""; - assert_eq!(value,expected ); + let value = join_paths(paths); + let expected = + "\"/home/tensor/code/TTSDockerContainer/Vall-E-mount/input/20.wav\" \"/home/tensor/code/TTSDockerContainer/Vall-E-mount/input/21.wav\""; + println!(expected); + assert_eq!(value, expected); } } diff --git a/crates/service/job/inference_job/src/job/job_types/tts/vall_e_x/vall_e_x_inference_command.rs b/crates/service/job/inference_job/src/job/job_types/tts/vall_e_x/vall_e_x_inference_command.rs index ffb8c5550d..2883327203 100644 --- a/crates/service/job/inference_job/src/job/job_types/tts/vall_e_x/vall_e_x_inference_command.rs +++ b/crates/service/job/inference_job/src/job/job_types/tts/vall_e_x/vall_e_x_inference_command.rs @@ -46,7 +46,7 @@ pub struct InferenceArgs> { pub stderr_output_file: String, } -pub struct CreateVoiceArgs> { +pub struct CreateVoiceInferenceArgs> { pub output_embedding_path: P, pub output_embedding_name: String, pub audio_files: String,