diff --git a/Cargo.lock b/Cargo.lock index c79561df90..424e595ff6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3155,7 +3155,7 @@ dependencies = [ ] [[package]] -name = "kube-pod-cleanup" +name = "kubectl-pod-cleanup" version = "0.0.1" dependencies = [ "config", 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 4f59c8a509..494983dca3 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 @@ -1,6 +1,6 @@ use anyhow::anyhow; -use log::{error, info}; - +use log::{ error, info }; +use std::time::Instant; use mysql_queries::queries::generic_inference::job::list_available_generic_inference_jobs::AvailableInferenceJob; use crate::job::job_loop::job_success_result::JobSuccessResult; @@ -9,62 +9,104 @@ use crate::job_dependencies::JobDependencies; // 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, - pub job: &'a AvailableInferenceJob, + pub job_dependencies: &'a JobDependencies, + pub job: &'a AvailableInferenceJob, } -pub async fn process_job(args: VALLEXProcessJobArgs<'_>) -> Result { +// pub async fn download_voice_embedding_from_hash(object_path: str, +// bucket_hash: str, +// private_bucket:&str) -> Result<> { +// let paths = BucketPathUnifier::default_paths(); +// let object_path = paths(hash, 0); +// let filesystem_path = format!("{}_weights.npz", name); +// private_bucket_client.download_file_to_disk(object_path, filesystem_path).await?; +// } - // get job args and dependencies and execute them with the inputs - let job = args.job; - let deps = args.job_dependencies; +// query using the token then grab the bucket hash +pub async fn process_job( + args: VALLEXProcessJobArgs<'_> +) -> Result { + - let job_progress_reporter = args.job_dependencies - .job_progress_reporter - .new_generic_inference(job.inference_job_token.as_str()) - .map_err(|e| ProcessSingleJobError::Other(anyhow!(e)))?; + let job = args.job; + let deps = args.job_dependencies; - // validate the inputs on the job shouldn't this happen at json - //let job_args = validate_job(job)?; + // get job args + let text = match job.maybe_raw_inference_text { + Some(value) => value, + None => { + Err(ProcessSingleJobError::InvalidJob(anyhow!("Missing Text for Inference"))) + } + }; - // Need to download the models - info!("Download models (if not present)..."); + // get some globals + let job_progress_reporter = deps.job_progress_reporter + .new_generic_inference(job.inference_job_token.as_str()) + .map_err(|e| ProcessSingleJobError::Other(anyhow!(e)))?; - 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; + // validate the inputs + // let job_args = validate_job(job)?; - if let Err(e) = result { - error!("could not download: {:?}", e); - return Err(ProcessSingleJobError::from_anyhow_error(e)) + // 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)); + } } - } - // Download embeddings files - - // rinr inference + // Download embeddings file using embedding token - // ==================== TEMP DIR ==================== // - // create a temp dir + // run inference + let work_temp_dir = format!("temp_zeroshot_inference_{}", 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))?; - // run inference + // delete temp dir files - // upload audio to bucket + // ==================== UPLOAD AUDIO TO BUCKET ==================== // + let maybe_args = job.maybe_inference_args + .as_ref() + .map(|args| args.args.as_ref()) + .flatten(); - // deletetemp dir files + let workdir = work_temp_dir.path().to_path_buf(); + let stderr_output_file = work_temp_dir.path().join("stderr.txt"); + let inference_start_time = Instant::now(); + + let command_exit_status = args.job_dependencies + .job_type_details + .vall_e_x + .inference_command + .execute_inference(InferenceArgs { + input_embedding: , // name of the embedding.npz in the tmp dir + input_text: , // name of the text + /// --result_file: path to final file output + output_file_name: , // output file name in the output folder + stderr_output_file: , + }); - // save result ? + // upload audio to public bucket ( for voice ) + // return ok + // Err(ProcessSingleJobError::InvalidJob(anyhow!("this job flow is not yet implemented"))) - Err(ProcessSingleJobError::InvalidJob(anyhow!("this job flow is not yet implemented"))) } - - #[cfg(test)] - mod test { +#[cfg(test)] +mod test { #[test] fn values() { - print!("{}","Hello!"); + print!("{}", "Hello!"); } - } \ No newline at end of file +} diff --git a/crates/service/job/inference_job/src/job/job_types/tts/vall_e_x/validate_job.rs b/crates/service/job/inference_job/src/job/job_types/tts/vall_e_x/validate_job.rs index ebc70db63a..8a7ab36aac 100644 --- a/crates/service/job/inference_job/src/job/job_types/tts/vall_e_x/validate_job.rs +++ b/crates/service/job/inference_job/src/job/job_types/tts/vall_e_x/validate_job.rs @@ -1,9 +1,57 @@ use anyhow::anyhow; +use mysql_queries::payloads::generic_inference_args::generic_inference_args::{InferenceCategoryAbbreviated, PolymorphicInferenceArgs}; use mysql_queries::queries::generic_inference::job::list_available_generic_inference_jobs::AvailableInferenceJob; +use mysql_queries::payloads::generic_inference_args::tts_payload::{TTSArgs}; -use crate::job::{job_loop::process_single_job_error::ProcessSingleJobError, job_types::lipsync::sad_talker::validate_job::JobArgs}; +use crate::job::job_loop::process_single_job_error::ProcessSingleJobError; +use crate::job::job_types::lipsync::sad_talker::validate_job::JobArgs; + +pub struct JobArgs<'a> { + text: String, + voice_token: String +} pub fn validate_job(job: &AvailableInferenceJob) -> Result { - Err(ProcessSingleJobError::InvalidJob(anyhow!("job not yet written"))) + let inference_category = job.maybe_inference_args.as_ref() + .map(|args|args.inference_category) + .flatten(); + + match inference_category { + Some(InferenceCategoryAbbreviated::TextToSpeech) => {}, + Some(category) => { + return Err(ProcessSingleJobError::from_anyhow_error(anyhow!("wrong inference category for job: {:?}", category))); + }, + None => { + return Err(ProcessSingleJobError::from_anyhow_error(anyhow!("no inference category for job!"))); + } + }; + + let inference_args = job.maybe_inference_args + .as_ref() + .map(|args| args.args.as_ref()) + .flatten(); + + let inference_args = match inference_args { + Some(args) => args, + None => { + return Err(ProcessSingleJobError::from_anyhow_error(anyhow!("no inference args for job!"))); + } + }; + + let ttsArgs = match inference_args { + PolymorphicInferenceArgs::Tts(inference_args) => inference_args, + _ => { + return Err(ProcessSingleJobError::from_anyhow_error(anyhow!("wrong inner args for job!"))); + } + }; + + + Ok(JobArgs { + text:, + voice_token: , + }) + + + } 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 7eb2f12737..3b75a16f65 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 @@ -43,7 +43,7 @@ static IGNORED_ENVIRONMENT_VARS : Lazy> = Lazy::new(|| { // so we have all the files downloaded for the inputs. pub struct InferenceArgs> { /// --driven_audio: path to the input audio - pub input_embedding: P, // name of the embedding.npz in the prompts folder? + pub input_embedding: P, // name of the embedding.npz in the tmp dir pub input_text: P, // name of the text /// --result_file: path to final file output pub output_file_name: P, // output file name in the output folder @@ -135,7 +135,7 @@ impl VallEXInferenceCommand { // Override for --checkpoint_dir at inference time let alternate_checkpoint_dir = easyenv::get_env_pathbuf_optional( - "SAD_TALKER_ALTERNATE_CHECKPOINT_PATH"); + "VALL_E_X_ALTERNATE_CHECKPOINT_PATH"); Ok(Self { root_code_directory, @@ -353,7 +353,7 @@ impl VallEXCreateEmbeddingCommand { // Override for --checkpoint_dir at inference time let alternate_checkpoint_dir = easyenv::get_env_pathbuf_optional( - "SAD_TALKER_ALTERNATE_CHECKPOINT_PATH"); + "VALL_E_X_CHECKPOINT_PATH"); Ok(Self { root_code_directory,