Fixed issue with typo in job.

Need to fix process job and run it.
This commit is contained in:
Michael Chung committed 2023-10-11 02:51:47 -04:00
1 parent 97be5da1b5
commit ec03420efc
4 files changed
+136 -46

No files matched your search

Generated
+1 -1
View File
@@ -3155,7 +3155,7 @@ dependencies = [
]
[[package]]
name = "kube-pod-cleanup"
name = "kubectl-pod-cleanup"
version = "0.0.1"
dependencies = [
"config",
@@ -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<JobSuccessResult, ProcessSingleJobError> {
// 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<JobSuccessResult, ProcessSingleJobError> {
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!");
}
}
}
@@ -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<JobArgs, ProcessSingleJobError> {
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: ,
})
}
@@ -43,7 +43,7 @@ static IGNORED_ENVIRONMENT_VARS : Lazy<HashSet<String>> = Lazy::new(|| {
// so we have all the files downloaded for the inputs.
pub struct InferenceArgs<P: AsRef<Path>> {
/// --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,